Apache Kafka是一种流行的分布式流式消息平台。Kafka生产者将数据写入分区主题，这些主题通过可配置的副本存储到broker群集上。消费者来消费存储在broker的分区生成的数据。
前提
kafka需要ZooKeeper，如果你还没安装，则先安装zookeeper。
Kafka安装
首先，创建 kafka.yaml
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
name: kafka-pdb
spec:
selector:
matchLabels:
app: kafka
maxUnavailable: 1
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kafka
spec:
serviceName: kafka-hs
replicas: 3
podManagementPolicy: Parallel
updateStrategy:
type: RollingUpdate
selector:
matchLabels:
app: kafka
template:
metadata:
labels:
app: kafka
spec:
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"
terminationGracePeriodSeconds: 300
containers:
- name: k8skafka
imagePullPolicy: IfNotPresent
image: kubebiz/kafka:0.10.2.1
resources:
requests:
memory: "1Gi"
cpu: "0.5"
ports:
- containerPort: 9093
name: server
command:
- sh
- -c
- "exec kafka-server-start.sh /opt/kafka/config/server.properties \
--override broker.id=${HOSTNAME##*-} \
--override listeners=PLAINTEXT://:9093 \
--override zookeeper.connect=zk-cs.default.svc.cluster.local:2181 \
--override log.dirs=/var/lib/kafka \
--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 "
env:
- name: KAFKA_HEAP_OPTS
value : "-Xmx512M -Xms512M"
- name: KAFKA_OPTS
value: "-Dlogging.level=INFO"
volumeMounts:
- name: datadir
mountPath: /var/lib/kafka
readinessProbe:
exec:
command:
- sh
- -c
- "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server=localhost:9093"
securityContext:
runAsUser: 1000
fsGroup: 1000
volumeClaimTemplates:
- metadata:
name: datadir
spec:
accessModes: [ "ReadWriteOnce" ]
resources:
requests:
storage: 10Gi
然后创建它
$ kubectl apply -f kafka.yaml
service "kafka-hs" created
poddisruptionbudget "kafka-pdb" created
statefulset "kafka" created
kafka通过zk-cs服务连接ZooKeeper集群，如果你还没安装，则先安装zookeeper。
等待所有pod的状态为Running：
$ kubectl get po -l app=kafka -w
NAME READY STATUS RESTARTS AGE
kafka-0 0/1 Pending 0 0s
kafka-0 0/1 Pending 0 0s
kafka-1 0/1 Pending 0 0s
kafka-1 0/1 Pending 0 0s
kafka-2 0/1 Pending 0 0s
kafka-0 0/1 ContainerCreating 0 0s
kafka-2 0/1 Pending 0 0s
kafka-1 0/1 ContainerCreating 0 0s
kafka-1 0/1 Running 0 11s
kafka-0 0/1 Running 0 19s
kafka-1 1/1 Running 0 23s
kafka-0 1/1 Running 0 32s
如果你观察Pod创建，您会注意到，Kafka集群使用了podManagementPolicy: Parallel策略。
测试集群
可以使用kubectl run来执行kafka-topics.sh脚本来创建名为test的主题。
kubectl run -ti --image=kubebiz/kafka:0.10.2.1 createtopic --restart=Never --rm -- kafka-topics.sh --create \
--topic test \
--zookeeper zk-cs.default.svc.cluster.local:2181 \
--partitions 1 \
--replication-factor 3
现在，使用kubectl run来执行kafka-console-consumer.sh来消费消息：
kubectl run -ti --image=kubebiz/kafka:0.10.2.1 consumer --restart=Never --rm -- kafka-console-consumer.sh --topic test --bootstrap-server kafka-0.kafka-hs.default.svc.cluster.local:9093
在打开一个新的命令行窗口，运行生产者：
kubectl run -ti --image=kubebiz/kafka:0.10.2.1 producer --restart=Never --rm \
-- kafka-console-producer.sh --topic test --broker-list kafka-0.kafka-hs.default.svc.cluster.local:9093,kafka-1.kafka-hs.default.svc.cluster.local:9093,kafka-2.kafka-hs.default.svc.cluster.local:9093 done;
第二个终端（terminal）的输出会出现在第一个终端中。如果您在更新集群时继续生产和消费消息，您会注意到没有消息丢失。当更新单个broker时，您可能会看到错误消息，因为分区的leader会发生变化，但客户端会重新尝试，直到消息被提交。
更新Kafka集群
StatefulSet 更新和 DaemonSet 更新一样，都是通过设置相应API对象的spec.updateStrategy来配置的。当更新策略设置为OnDelete时，只有当 StatefulSet 或 DaemonSet 中的 Pod 被删除时，各个控制器才会创建新的 Pod 。当更新策略设置为RollingUpdate时，当对 DaemonSet 或 StatefulSet 的spec.template字段进行修改时，控制器将删除并重新创建Pod。您可以使用滚动更新来更改 StatefulSet 或 DaemonSet 中 Pods 的配置（通过环境变量或命令行参数）、资源请求、资源限制、容器镜像、标签和/或注释。请注意，所有更新都是破坏性的，总是需要销毁和重新创建 DaemonSet 或 StatefulSet 中的每个 Pod 。StatefulSet 滚动更新与 DaemonSet 滚动更新的不同之处在于，Pod的终止和创建是有序的、有顺序的。
你可以使用patch调整kafka StatefulSet的配置，比如将CPU资源减少到250m。
kubectl patch sts kafka --type='json' -p='[{"op": "replace", "path": "/spec/template/spec/containers/0/resources/requests/cpu", "value":"250m"}]'
如果你观察StatefulSet中Pod的状态，你会发现每个Pod都是按照相反的顺序被删除和重新创建的（从序数最大的Pod开始，然后到最小的）。控制器等待每个更新的Pod运行并准备好后再更新后续的Pod。
$kubectl get po -lapp=kafka -w
NAME READY STATUS RESTARTS AGE
kafka-0 1/1 Running 0 13m
kafka-1 1/1 Running 0 13m
kafka-2 1/1 Running 0 13m
kafka-2 1/1 Terminating 0 14m
kafka-2 0/1 Terminating 0 14m
kafka-2 0/1 Terminating 0 14m
kafka-2 0/1 Terminating 0 14m
kafka-2 0/1 Pending 0 0s
kafka-2 0/1 Pending 0 0s
kafka-2 0/1 ContainerCreating 0 0s
kafka-2 0/1 Running 0 10s
kafka-2 1/1 Running 0 21s
kafka-1 1/1 Terminating 0 14m
kafka-1 0/1 Terminating 0 14m
kafka-1 0/1 Terminating 0 14m
kafka-1 0/1 Terminating 0 14m
kafka-1 0/1 Pending 0 0s
kafka-1 0/1 Pending 0 0s
kafka-1 0/1 ContainerCreating 0 0s
kafka-1 0/1 Running 0 11s
kafka-1 1/1 Running 0 21s
kafka-0 1/1 Terminating 0 14m
kafka-0 0/1 Terminating 0 14m
kafka-0 0/1 Terminating 0 14m
kafka-0 0/1 Terminating 0 14m
kafka-0 0/1 Pending 0 0s
kafka-0 0/1 Pending 0 0s
kafka-0 0/1 ContainerCreating 0 0s
kafka-0 0/1 Running 0 10s
kafka-0 1/1 Running 0 22s
请注意，在更新过程中，计划外的中断不会导致无意的更新。也就是说，StatefulSet控制器将始终以正确的版本重新创建Pod，以确保保留更新的顺序。如果一个Pod被删除，如果它已经被更新，它将从StatefulSet的spec.template的更新版本创建。如果Pod还没有更新，它将从StatefulSet的spec.template的上一个版本创建。我们将在下面的章节中进一步探讨这个问题。
阶段性更新（Staging an update）
处理部署和配置修改的方式，您可能希望或需要先进行对StatefulSet的更新，然后再进行部署。您可以通过为RollingUpdate设置分区来完成此操作。当StatefulSet控制器在StatefulSet的updateStrategy中检测到分区时，它将仅将StatefulSet的spec.template的更新版本应用于序数大于或等于该分区值的Pod。
可以通过给kafka StatefulSet打patch，以将分区添加到RollingUpdate更新策略。如果您将分区设置为大于或等于StatefulSet的spec.replicas的数字（如下所示），则您对StatefulSet的spec.template执行的任何后续更新都将分阶段发布，但StatefulSet控制器将不会开始滚动更新。
$ kubectl patch sts kafka -p '{"spec":{"updateStrategy":{"type":"RollingUpdate","rollingUpdate":{"partition":3}}}}'
statefulset "kafka" patched
如果你将StatefulSet打上patch，将请求的CPU设置为0.3，你会发现没有一个Pods被更新。
$ kubectl patch sts kafka --type='json' -p='[{"op": "replace", "path": "/spec/template/spec/containers/0/resources/requests/cpu", "value":"0.3"}]'
statefulset "kafka" patched
即使你删除一个Pod并等待StatefulSet控制器重新创建它，你会注意到Pod是以当前的CPU请求重新创建的。
$ kubectl delete po kafka-1
pod "kafka-1" deleted
$ kubectl get po kafka-1 -w
NAME READY STATUS RESTARTS AGE
kafka-1 0/1 ContainerCreating 0 10s
kafka-1 0/1 Running 0 19s
kafka-1 1/1 Running 0 21s
$ kubectl get po kafka-1 -o yaml
apiVersion: v1
kind: Pod
metadata:
...
resources:
requests:
cpu: 250m
memory: 1Gi
金丝雀（Rolling out a canary）
通常，我们要在全局应用之前，先在应用程序的单个实例上验证镜像更新或配置更改。如果将上面创建的partition修改为2，StatefulSet控制器将推出一个canary，可用于验证更新是否按预期工作。
$ kubectl patch sts kafka -p '{"spec":{"updateStrategy":{"type":"RollingUpdate","rollingUpdate":{"partition":2}}}}'
statefulset "kafka" patched
您可以观看StatefulSet控制器更新kafka-2 Pod，并在更新完成后暂停。
$ kubectl get po -lapp=kafka -w
NAME READY STATUS RESTARTS AGE
kafka-0 1/1 Running 0 50m
kafka-1 1/1 Running 0 10m
kafka-2 1/1 Running 0 29s
kafka-2 1/1 Terminating 0 34s
kafka-2 0/1 Terminating 0 38s
kafka-2 0/1 Terminating 0 39s
kafka-2 0/1 Terminating 0 39s
kafka-2 0/1 Pending 0 0s
kafka-2 0/1 Pending 0 0s
kafka-2 0/1 Terminating 0 20s
kafka-2 0/1 Terminating 0 20s
kafka-2 0/1 Pending 0 0s
kafka-2 0/1 Pending 0 0s
kafka-2 0/1 ContainerCreating 0 0s
kafka-2 0/1 Running 0 19s
kafka-2 1/1 Running 0 22s
分阶段滚动（Phased roll outs）
类似于金丝雀，你可以根据阶段性进行（如线性、几何或指数性推出）更新。
如果你把partition设置为1，StatefulSet控制器就会多更新一个broker。
$ kubectl patch sts kafka -p '{"spec":{"updateStrategy":{"type":"RollingUpdate","rollingUpdate":{"partition":1}}}}'
statefulset "kafka" patched
如果你把它设置为0，StatefulSet控制器就会更新最终的broker并完成更新。
$ kubectl patch sts kafka -p '{"spec":{"updateStrategy":{"type":"RollingUpdate","rollingUpdate":{"partition":0}}}}'
statefulset "kafka" patched
请注意，您不必将partition递减一。对于大型的StatefulSet（例如，具有100个副本的StatefulSet），可以使用像100、99、90、50、0的进度。在这种情况下，分阶段进行更新，部署Canary，部署到10个实例，更新50%的Pod，然后完成更新。
博主，用了本文的yaml文件和 https://www.kubebiz.com/KubeBiz/kafka 的都发现个问题，就是kafka节点一直都是未就绪
用以下命令看看卡在哪个地方：
kubectl get events -n my-space 或者 kubectl describe pods kafka-0 -n my-space一般是存储分配不到，或者镜像还在拉取。
[root@k8s01 kafka]# kubectl get -n tools-env po -l app=kafka NAME READY STATUS RESTARTS AGE kafka-0 0/1 Running 0 14m kafka-1 1/1 Running 0 15m kafka-2 1/1 Running 0 15m总是会掉一会，然后又就绪了，导致logstash配置bootstrap_servers时候kafka-0就有点问题。
zk安装好了？
Kubernetes(k8s)运行ZooKeeper，一个分布式系统协调器
好了的 我测试消息都能正常消费的 就是kafka-0这个节点会是不是的未就绪 然后又恢复
唯一有点疑问的是，看您kafak的里面端口是9093 我使用用的是9092 这个应该不是这个问题的关键 感觉
events里面没有失败的原因吗
kubectl logs -f kafka-0我记得好像有机器不太好，健康检查的超时时间调大就可以了（默认是1秒）
readinessProbe: exec: command: - sh - -c - "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server=localhost:9093" timeoutSeconds: 5好的，我去设置观察下。谢谢您的回复！
我发现如果不指定storageClassName，pvc模板并不会使用已有的sc生成pv，导致pod一直pending...
k8s 1.20.11 部署缺少
selector: matchLabels: app: kafka嗯，k8s的版本，所以对参数有区别，可以参考：https://www.kubebiz.com/KubeBiz/kafka
里面有k8s的版本选择，会自动补全。
这些镜像老是下载不下来怎么办？有没有什么方法
访问不到国外，封掉了，可以换成kubebiz的镜像。
参考：https://www.kubebiz.com/KubeBiz/kafka
这样部署外部服务可以访问吗？
外部要识别pod的ip才行，否则是不行。
kafka作为一个需要大量IO的消息中间件，应该不被建议塞进容器中，容易造成数据安全的问题吧
不会的，容器只是一层皮，网络或io也好都是走路由或iptables，本身没差的。
只是大家对容器不熟悉，总觉得有问题，却又说不出来什么问题。
我之前看到kafka用的是pagecache的内存，所以这里有一个以为，只设置jvm的内存参数，当pagecache使用超过了container的limit，会是怎样的情况？
超出container的部分会主动要求释放，释放的了，安全，释放不了，oom。
所以pagecache这部分的内存的gc的动作是不受jvm的gc的机制的，是受操作系统控制的？
是的，利用的是os的机制。
你好，我使用K8S部署kafka用的是wurstmeister/kafka这个镜像。
我起了我起了3个节点。每个pod限制内存上限是2G。
这三个节点我部署在了一个node节点上，（一台主机4核16g。并且这台主机只负责部署kafka和flink）。
压测的时候也没问题。但是跑上一段时间会导致整个node节点挂掉。（第一次运行不到一周，后面升级了K8S版本坚持了半个多月）。
不知道自己那些地方没有配置好。
或者说我换你这个镜像的话。用在生产的话配置有什么推荐修改的么？
我怀疑你限制了内存上限，但是容器里的kafka的jvm配置可能会大于你的限制，导致超过就oom了给kill了。
你可以观察一下先。
我这个是官方推荐的配置方式。放心使用