nearpod/kafka

docker-infra: Kafka 镜像文档

特别说明

• 本镜像基于Alpine Linux。对于JVM镜像，这可能会导致意外后果，一切可能中断。效果因人而异（YMMV）
• 本项目使用Zulu，这是一个经过全面测试、兼容性验证且可信的OpenJDK 9、8及更早版本的二进制发行版。Zulu使用条款

Kafka

本项目包含一个Docker镜像，旨在使用StatefulSets在Kubernetes上部署Apache Kafka。

主要灵感来源于：https://github.com/kubernetes/contrib/tree/master/statefulsets/kafka

Kubernetes Kafka K8SKafka

本项目包含一个Docker镜像，旨在使用StatefulSets在Kubernetes上部署Apache Kafka。

限制

1. 必须使用持久卷（Persistent Volumes）。emptyDirs可能导致数据丢失。
2. 目前通常无法实现存储介质I/O隔离。考虑使用Pod Anti-Affinity规则将资源密集型工作负载部署在不同节点上。

Docker镜像

本仓库中的Docker镜像基于Ubuntu 16.04，使用基于1.8 JVM的OpenJDK JRE最新版本（JDK 8u111）和基于Scala 2.11的Kafka最新稳定版本（10.2.0）。Ubuntu镜像比BusyBox或Alpine大，但后两者包含mucl或ulibc，需要针对非glibc的libc运行时构建自定义OpenJDK版本。虽然有基于Alpine和BusyBox的更小Kafka镜像，但Kafka、JVM和glibc之间的交互更易于理解和调试。

镜像设计为以非root用户运行Kafka JVM进程。默认用户为kafka，UID 1000，GID 1000。Kafka安装在/opt/kafka目录，所有配置位于/opt/kafka/config，可执行文件位于/opt/kafka/bin。由于/opt/kafka/bin中的脚本实现限制，无法将其符号链接到/user/bin目录，因此/opt/kafka/bin目录已添加到PATH环境变量。

ZooKeeper

Kafka需要Apache ZooKeeper用于 broker 配置存储和协调。可在https://github.com/kubernetes/contrib/tree/master/statefulsets/zookeeper%E6%89%BE%E5%88%B0%E5%9C%A8Kubernetes%E4%B8%8A%E9%83%A8%E7%BD%B2ZooKeeper%E9%9B%86%E7%BE%A4%E7%9A%84%E7%A4%BA%E4%BE%8B%E3%80%82%E6%B5%8B%E8%AF%95%E7%8E%AF%E5%A2%831-3%E5%8F%B0%E6%9C%8D%E5%8A%A1%E5%99%A8%E8%B6%B3%E5%A4%9F%EF%BC%8C%E7%94%9F%E4%BA%A7%E7%8E%AF%E5%A2%83%E5%BB%BA%E8%AE%AE%E8%87%B3%E5%B0%91%E9%83%A8%E7%BD%B25%E5%8F%B0%EF%BC%8C%E4%BB%A5%E5%AE%B9%E5%BF%8D%E7%BB%B4%E6%8A%A4%E6%9C%9F%E9%97%B4%E7%9A%84%E5%8D%95%E8%8A%82%E7%82%B9%E6%95%85%E9%9A%9C%E3%80%82%E8%8B%A5%E5%9C%A8Kubernetes%E4%B8%8A%E8%BF%90%E8%A1%8CZooKeeper%EF%BC%8C%E5%BB%BA%E8%AE%AE%E4%B8%BA%E6%AF%8F%E4%B8%AAKafka%E9%9B%86%E7%BE%A4%E4%BD%BF%E7%94%A8%E7%8B%AC%E7%AB%8B%E7%9A%84ZooKeeper%E9%9B%86%E7%BE%A4%E3%80%82%E7%94%9F%E4%BA%A7%E7%8E%AF%E5%A2%83%E4%B8%AD%EF%BC%8C%E6%AF%8F%E4%B8%AAZooKeeper%E6%9C%8D%E5%8A%A1%E5%99%A8%E5%BA%94%E8%87%B3%E5%B0%91%E6%9C%892 GiB堆内存，Pod保留内存至少4 GiB。ZooKeeper对CPU要求不高，每服务器2个CPU通常足够。若在云 provider（如GCP、Azure或AWS）上运行Kubernetes，应为ZooKeeper PV配置快速存储类。由于PV由网络附加存储支持，将预写日志与快照目录隔离几乎无收益。

Headless Service

Kafka StatefulSet需要Headless Service控制Kafka broker的网络域。以下YAML创建Headless Service，允许发现broker并暴露9093端口供客户端连接。

yaml
apiVersion: v1
kind: Service
metadata:
  name: kafka-svc
  labels:
    app: kafka
spec:
  ports:
  - port: 9093
    name: server
  clusterIP: None
  selector:
    app: kafka

StatefulSet

Kafka StatefulSet在Kubernetes集群上部署可配置数量的副本。StatefulSet的serviceName必须与Headless Service匹配，并指定所需的broker数量。

yaml
apiVersion: apps/v1beta1
kind: StatefulSet
metadata:
  name: kafka
spec:
  serviceName: kafka-svc
  replicas: 3
  ...

配置

本节详细说明Kafka集群的配置。

Broker配置

每个broker的配置通过命令行标志覆盖默认配置生成。以下是Kafka文档中的高、中重要性配置参数。

yaml
kafka-server-start.sh /opt/kafka/config/server.properties --override broker.id=${HOSTNAME##*-} \
          --override listeners=PLAINTEXT://:9093 \
          --override zookeeper.connect=zk-0.zk-svc.default.svc.cluster.local:2181,zk-1.zk-svc.default.svc.cluster.local:2181,zk-2.zk-svc.default.svc.cluster.local:2181 \
          --override auto.create.topics.enable=true \
          --override auto.leader.rebalance.enable=true \
          --override background.threads=10 \
          --override compression.type=producer \
          --override delete.topic.enable=false \
          --override leader.imbalance.check.interval.seconds=300 \
          --override leader.imbalance.per.broker.percentage=10 \
          --override log.flush.interval.messages=9223372036854775807 \
          --override log.flush.offset.checkpoint.interval.ms=60000 \
          --override log.flush.scheduler.interval.ms=9223372036854775807 \
          --override log.retention.bytes=-1 \
          --override log.retention.hours=168 \
          --override log.roll.hours=168 \
          --override log.roll.jitter.hours=0 \
          --override log.segment.bytes=1073741824 \
          --override log.segment.delete.delay.ms=60000 \
          --override message.max.bytes=1000012 \
          --override min.insync.replicas=1 \
          --override num.io.threads=8 \
          --override num.network.threads=3 \
          --override num.recovery.threads.per.data.dir=1 \
          --override num.replica.fetchers=1 \
          --override offset.metadata.max.bytes=4096 \
          --override offsets.commit.required.acks=-1 \
          --override offsets.commit.timeout.ms=5000 \
          --override offsets.load.buffer.size=5242880 \
          --override offsets.retention.check.interval.ms=600000 \
          --override offsets.retention.minutes=1440 \
          --override offsets.topic.compression.codec=0 \
          --override offsets.topic.num.partitions=50 \
          --override offsets.topic.replication.factor=3 \
          --override offsets.topic.segment.bytes=104857600 \
          --override queued.max.requests=500 \
          --override quota.consumer.default=9223372036854775807 \
          --override quota.producer.default=9223372036854775807 \
          --override replica.fetch.min.bytes=1 \
          --override replica.fetch.wait.max.ms=500 \
          --override replica.high.watermark.checkpoint.interval.ms=5000 \
          --override replica.lag.time.max.ms=10000 \
          --override replica.socket.receive.buffer.bytes=65536 \
          --override replica.socket.timeout.ms=30000 \
          --override request.timeout.ms=30000 \
          --override socket.receive.buffer.bytes=102400 \
          --override socket.request.max.bytes=104857600 \
          --override socket.send.buffer.bytes=102400 \
          --override unclean.leader.election.enable=true \
          --override zookeeper.session.timeout.ms=6000 \
          --override zookeeper.set.acl=false \
          --override broker.id.generation.enable=true \
          --override connections.max.idle.ms=600000 \
          --override controlled.shutdown.enable=true \
          --override controlled.shutdown.max.retries=3 \
          --override controlled.shutdown.retry.backoff.ms=5000 \
          --override controller.socket.timeout.ms=30000 \
          --override default.replication.factor=1 \
          --override fetch.purgatory.purge.interval.requests=1000 \
          --override group.max.session.timeout.ms=300000 \
          --override group.min.session.timeout.ms=6000 \
          --override inter.broker.protocol.version=0.10.2-IV0 \
          --override log.cleaner.backoff.ms=15000 \
          --override log.cleaner.dedupe.buffer.size=134217728 \
          --override log.cleaner.delete.retention.ms=86400000 \
          --override log.cleaner.enable=true \
          --override log.cleaner.io.buffer.load.factor=0.9 \
          --override log.cleaner.io.buffer.size=524288 \
          --override log.cleaner.io.max.bytes.per.second=1.7976931348623157E308 \
          --override log.cleaner.min.cleanable.ratio=0.5 \
          --override log.cleaner.min.compaction.lag.ms=0 \
          --override log.cleaner.threads=1 \
          --override log.cleanup.policy=delete \
          --override log.index.interval.bytes=4096 \
          --override log.index.size.max.bytes=10485760 \
          --override log.message.timestamp.difference.max.ms=9223372036854775807 \
          --override log.message.timestamp.type=CreateTime \
          --override log.preallocate=false \
          --override log.retention.check.interval.ms=300000 \
          --override max.connections.per.ip=2147483647 \
          --override num.partitions=1 \
          --override producer.purgatory.purge.interval.requests=1000 \
          --override replica.fetch.backoff.ms=1000 \
          --override replica.fetch.max.bytes=1048576 \
          --override replica.fetch.response.max.bytes=10485760 \
          --override reserved.broker.max.id=1000

• broker.id从StatefulSet Pod的序号索引提取。
• listeners配置必须指定headless service指示的端口（此处为9093）。
• zookeeper.connect是ZooKeeper集群中服务器的host:port对逗号分隔列表。

操作系统镜像调优

生产环境中，必须配置基础OS镜像以允许足够数量的文件描述符来应对工作负载。

• 对于每个broker，(分区数) * (分区大小/段大小)决定了broker在任何时候打开的文件数。必须确保这不会导致broker进程因耗尽允许的文件描述符而终止。

CPU

典型的生产Kafka broker部署运行在双处理器Xeon上，每个核心有多个硬件线程。但CPU不太可能成为瓶颈。8 CPU部署足以提供良好性能。建议从2-4 CPU开始模拟工作负载，然后逐步增加。CPU极不可能成为部署瓶颈。

内存

Kafka大量利用OS页缓存缓冲数据。要充分理解Kafka与Linux容器的交互，应阅读此文档和此文档。特别重要的是理解mem cgroup对页缓存的计算和隔离。

若主要关注隔离和性能，应执行以下操作：

• 确定要缓冲的数据秒数t（时间）。
• 确定部署的总写入吞吐量tp（存储/时间）。
• tp * t给出应保留的内存存储需求，应将其设置为容器的内存请求。

磁盘

磁盘吞吐量是用户使用Kafka时最常见的瓶颈。由于持久卷由网络附加存储支持，在大多数情况下，吞吐量在每个节点基础上被限制，与附加到节点的持久卷数量无关。例如，若在GKE或GCP的Kubernetes集群上部署Kafka，使用标准PD类型，每个实例的最大持续吞吐量为120 MB/s（写入）和180 MB/s（读取）。若有多个应用程序，每个都挂载持久卷，这些数值代表总可实现吞吐量。若发现争用，应考虑使用Pod Anti-Affinity规则确保资源密集型工作负载不共置在同一节点。

Pod亲和性

StatefulSet的PodTemplateSpec中的Kafka Pod包含Pod Anti-Affinity和Pod Affinity规则。

yaml
    affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: "app"
                    operator: In
                    values:
                    - kafka
              topologyKey: "kubernetes.io/hostname"
        podAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
             - weight: 1
               podAffinityTerm:
                 labelSelector:
                    matchExpressions:
                      - key: "app"
                        operator: In
                        values:
                        - zk
                 topologyKey: "kubernetes.io/hostname"

Pod Anti-Affinity规则确保两个Kafka Broker不会在同一节点上启动。这并非绝对必要，但有助于在节点故障时提供更强的可用性保证，并缓解磁盘吞吐量瓶颈。

Pod Affinity规则尝试将Kafka和ZooKeeper共置在同一节点。Kafka broker数量可能多于ZooKeeper服务器，但Kubernetes调度器会在可能的情况下，在尊重Pod Anti-Affinity规则强制的硬分布的同时，尝试共置Kafka broker和ZooKeeper服务器。此优化旨在最小化ZooKeeper集群和Kafka集群之间的网络I/O。但若出现磁盘争用，表达Pod Anti-Affinity规则确保ZooKeeper服务器和Kafka broker不调度到同一节点同样有效。

测试

测试部署的最简单方法是创建主题并使用控制台生产者和消费者。使用kubectl exec在其中一个broker上执行bash shell。

shell
> kubectl exec -ti kafka-0 -- bash

通过命令行使用kafka-topics.sh创建主题：

shell
> kafka-topics.sh --create \
--topic test \
--zookeeper zk-0.zk-svc.default.svc.cluster.local:2181,zk-1.zk-svc.default.svc.cluster.local:2181,zk-2.zk-svc.default.svc.cluster.local:2181 \
--partitions 3 \
--replication-factor 2

运行控制台消费者：

shell
> kafka-console-consumer.sh --topic test --bootstrap-server localhost:9093

使用kubectl exec在另一个broker上执行bash shell（可使用同一broker，但使用不同broker可证明系统跨多个节点工作）：

shell
> kubectl exec -ti kafka-1 -- bash

运行控制台生产者并通过输入生成几条消息。每次按Enter键会将消息刷新到消费者：

shell
> kafka-console-producer.sh --topic test --broker-list localhost:9093
hello
I like kafka
goodbye

在运行控制台消费者的终端将看到消息：

shell
> kafka-console-consumer.sh --topic test --bootstrap-server localhost:9093
hello
I like kafka
goodbye

水平扩展

可使用kubectl scale水平扩展集群。以下命令将broker数量扩展到2个：

shell
> kubectl scale statefulset kafka --replicas=2

注意，当向上或向下扩展Kafka集群时，必须使用kafka-reassign-partitions.sh确保扩展后数据正确复制和分配。