1 介绍

Apache Kafka Raft 是一种共识协议，它的引入是为了消除 Kafka 对 ZooKeeper 的元数据管理的依赖，被社区称之为 Kafka Raft metadata mode，简称 KRaft 模式。本文基于最新的 3.1.1 版本快速部署一个测试版本的 kafka, 若需要部署老版本kafka 可以参考博文 kafka笔记1–基础介绍与集群安装

此处安装3.1.1 版本的kafka，截止2023-10，3.5.1版本也可以通过本文方法来安装

2 部署测试

2.1 部署

在 downloads.apache.org/kafka/ 下载 kafka， 也可以在官方下载，官方可能网速比较慢。

解压后按照如下方式初始化即可:

配置 server.properties
$ vim config/kraft/server.properties
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://192.168.31.61:9092 # 此处要更改为实际ip，否则只能在本机访问

生成集群 ID
$ bin/kafka-storage.sh random-uuid
1kwB-zWkQfmc7Lc9wMuiLQ

格式化存储目录
$ bin/kafka-storage.sh format -t 1kwB-zWkQfmc7Lc9wMuiLQ -c config/kraft/server.properties
Formatting /tmp/kraft-combined-logs

启功节点服务
$ bin/kafka-server-start.sh config/kraft/server.properties
加上 -daemon 直接后台启动
$ bin/kafka-server-start.sh -daemon config/kraft/server.properties

2.2 测试

通过如下命令创建、查看topic，开启生产和消费者。

创建topic：
$ bin/kafka-topics.sh --create --topic k8s-log --partitions 1 --replication-factor 1 --bootstrap-server 127.0.0.1:9092
Created topic k8s-log.

查看topic：
$ bin/kafka-topics.sh --list --bootstrap-server  127.0.0.1:9092
__consumer_offsets
k8s-log

查看group：
$ bin/kafka-consumer-groups.sh --list --bootstrap-server  127.0.0.1:9092
logstash-k8s-log

查看topic详细信息：
$ bin/kafka-topics.sh --describe --topic k8s-log --bootstrap-server 127.0.0.1:9092
Topic: k8s-log	TopicId: s08iH5PcRk-B-PTkBMk1kQ	PartitionCount: 1	ReplicationFactor: 1	Configs: segment.bytes=1073741824
	Topic: k8s-log	Partition: 0	Leader: 1	Replicas: 1	Isr: 1

开启消费者：
$ bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic k8s-log

开启生产者：
$ bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic k8s-log

删除topic：
$ bin/kafka-topics.sh --delete --topic k8s-log --bootstrap-server 127.0.0.1:9092

删除group：
$/bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --delete --group ck-sre_k8s_sh_log
Deletion of requested consumer groups ('k8s-log') was successful.

如下图，在 producer 中输入数据后，可以在consumer消费并输出数据。

3 注意事项

1. advertised.listeners=PLAINTEXT://localhost:9092 需要更改为实际ip，否则只能在本机访问

2. 使用kraft 的kafka需要配置 config/kraft/server.properties 文件

3. kafka 查看分区消息数量 - GetOffsetShell脚本方式

   虽然kafka.tools.GetOffsetShell工具的使用方法没有记录在官方文档中，但使用它却非常简单。假设我也查询名为`test`的topic的消息数，具体命令如下：
   kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic k8s-log --time -1
   -1表示查询test各个分区当前最大的消息位移值(注意，这里的位移不只是consumer端的位移，而是指消息在每个分区的位置)
   
   如果你要查询曾经生产过的最大消息数，那么只运行上面这条命令然后把各个分区的结果相加就可以了。但如果你需要查询当前集群中该topic的消息数，那么还需要运行下面这条命令：
   kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic k8s-log --time -2
   -2表示去获取当前各个分区的
   

4. kafka 调整指定 groupid 的offset

   kafka-consumer-groups.sh --bootstrap-server ip01:9092,ip02:9092,ip03:9092 --group  ck-sls_nginx_ingress --reset-offsets --to-earliest --topic sls-nginx-ingress --execute
   最早： --to-earliest
   最新： --to-latest
   

5. 配置 kafka 自动重启crontab任务

   01: 完成 run_kafka.sh 脚本
   vim /your-path/run_kafka.sh
   #!/bin/bash
   source /root/.bashrc
   
   sleep 5
   jps|grep Kafka
   if [ $? -eq 0 ];then
   echo 'kafka is running!'
   else
   echo 'we will start kafka'
   cd /apps/svr/your-kafka-path
   bin/kafka-server-start.sh -daemon config/kraft/server.properties
   fi
   
   02: 配置crontab 定时任务
   */5 * * * * bash /your-path/run_kafka.sh
   
   注意：
   由于crontab无法引用用户配置的环境变量(~/.bashrc中JAVA_HOME登变量)，run_kafka.sh 中需要 source /root/.bashrc (若不是通过root配置java环境变量，按需更改为启动用户即可)
   

4 说明

软件环境：
Ubuntu 20.04 Desktop
kafka_2.12-3.1.1
参考文档：
downloads.apache.org/kafka
官方文档 – Kafka 3.0 Documentation
Kafka高可用 — KRaft集群搭建
Kafka入门系列(十三) 如何查看topic消息数