在 kafka-0 节点中创建一个名为 Message 的 topic，3个分区，3个副本：
[@k8s-master1 ~]# kubectl exec -it kafka-0 -n ns-kafka -- /bin/bash bash-4.4# kafka-topics.sh --create --topic Message --zookeeper zk-inner-service.ns-kafka.svc.cluster.local:2181 --partitions 3 --replication-factor 3 Created topic Message.在 zk-1 节点中查看是否存在这个 Topic：
[@k8s-master1 ~]# kubectl exec -it zk-1 -n ns-kafka -- zkCli.sh Connecting to localhost:2181 [zk: localhost:2181(CONNECTED) 0] ls / [admin, brokers, cluster, config, consumers, controller, controller_epoch, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper] [zk: localhost:2181(CONNECTED) 1] ls /brokers [ids, seqid, topics] [zk: localhost:2181(CONNECTED) 3] ls /brokers/topics [Message]可以看到 Zookeeper 中已经存在这个 Topic 了。
3.3 模拟生产者和消费者首先 在 kafka-1 上模拟生产者向 Message 中写入消息：
[@k8s-master1 ~]# kubectl exec -it kafka-1 -n ns-kafka -- /bin/bash bash-4.4# kafka-console-producer.sh --topic Message --broker-list kafka-0.kafka-service.ns-kafka.svc.cluster.local:9092,kafka-1.kafka-service.ns-kafka.svc.cluster.local:9092,kafka-2.kafka-service.ns-kafka.svc.cluster.local:9092 >This is a test message >Welcome to Kafka然后在 kafka-2 中模拟消费者消费这些信息：
[@k8s-master1 ~]# kubectl exec -it kafka-2 -n ns-kafka -- /bin/bash bash-4.4# kafka-console-consumer.sh --topic Message --bootstrap-server kafka-0.kafka-service.ns-kafka.svc.cluster.local:9092,kafka-1.kafka-service.ns-kafka.svc.cluster.local:9092,kafka-2.kafka-service.ns-kafka.svc.cluster.local:9092 --from-beginning This is a test message Welcome to Kafka可以正常生产消息和消费消息，代表 Kafka 集群运行正常。
十、FAQ 1. 如何在 yaml 文件中指定要创建的 Topic在 yaml 文件中指定如下 env，即可在 Pod 启动自动创建 Topic：
env: - name: KAFKA_CREATE_TOPICS value: "Topic1:1:3,Topic2:1:1:compact"上面的内容代表 Topic1 会有 1 个分区，3个副本，Topic2 会有 1 个分区，1 个副本并且副本的 cleanup.policy 设置为 compact。
自动创建 Topic 一定要设置 KAFKA_CREATE_TOPICS 变量，然后会由 create_topic.sh 脚本（镜像中存在）根据变量内容自动创建。
2. 为 Topic 设置的 compaction 不生效可参考网址：