Kafka KRaft 模式

Kafka KRaft 架构说明、集群部署、存储目录格式化和启停脚本。

Kafka KRaft 模式

1、Kafka-Kraft架构

左图为Kafka现有架构，元数据在zookeeper中，运行时动态选举controller，由controller进行Kafka集群管理。右图为kraft模式架构（实验性），不再依赖zookeeper集群，而是用三台controller节点代替zookeeper，元数据保存在controller中，由controller直接进行Kafka集群管理。

这样做的好处有以下几个：

Kafka不再依赖外部框架，而是能够独立运行；
controller管理集群时，不再需要从zookeeper中先读取数据，集群性能上升；
由于不依赖zookeeper，集群扩展时不再受到zookeeper读写能力限制；

controller不再动态选举，而是由配置文件规定。这样我们可以有针对性的加强controller节点的配置，而不是像以前一样对随机controller节点的高负载束手无策。

2、Kafka-Kraft集群部署

1）解压

[root@kafka102 software]$ tar -xzvf kafka_2.12-3.2.1.tgz -C /opt/module/

2）重名名

[root@kafka102 software]$ cd /opt/module/
[root@kafka102 module]$ mv kafka_2.12-3.2.1/ kafka-kraft-3.2.1/

3）修改server.properties文件

注意

,9094是gitlab的监听端口,停止gitlab

[root@kafka102 module]$ cd kafka-kraft-3.2.1/
[root@kafka102 kafka-kraft-3.2.1]$ vim config/kraft/server.properties
#kafka的角色（controller相当于主机、broker节点相当于从机，主机类似zk功能）
process.roles=broker,controller
node.id=2
controller.quorum.voters=2@kafka102:9094,3@kafka103:9094,4@kafka104:9094
advertised.listeners=PLAINTEXT://kafka102:9092
log.dirs=/opt/module/kafka-kraft-3.2.1/datas

问题

[2022-08-29 19:29:02,149] ERROR Exiting Kafka due to fatal exception (kafka.Kafka$)
java.lang.IllegalArgumentException: requirement failed: If process.roles contains just the 'broker' role, the node id 3 must not be included in the set of voters controller.quorum.voters=Set(2, 3, 4)
    at scala.Predef$.require(Predef.scala:281)
    at kafka.server.KafkaConfig.validateValues(KafkaConfig.scala:2075)
    at kafka.server.KafkaConfig.<init>(KafkaConfig.scala:1997)
    at kafka.server.KafkaConfig.<init>(KafkaConfig.scala:1471)
    at kafka.server.KafkaConfig$.fromProps(KafkaConfig.scala:1394)
    at kafka.Kafka$.buildServer(Kafka.scala:67)
    at kafka.Kafka$.main(Kafka.scala:87)
    at kafka.Kafka.main(Kafka.scala)

4）分发kafka-kraft

[root@kafka102 kafka-kraft-3.2.1]$ cd ..
[root@kafka102 module]$ xsync.sh kafka-kraft-3.2.1/

5）修改kafka103 /kafka104配置文件

[root@kafka103 kafka-kraft-3.2.1]$ vim config/kraft/server.properties
node.id=3
advertised.listeners=PLAINTEXT://kafka103:9092

[root@kafka104 kafka-kraft-3.2.1]$ vim config/kraft/server.properties
node.id=4
advertised.listeners=PLAINTEXT://kafka104:9092

在kafka103和kafka104上需要对node.id相应改变，值需要和controller.quorum.voters对应。

在kafka103和kafka104上需要根据各自的主机名称，修改相应的advertised.Listeners地址。

6）初始化集群数据目录

（1）首先生成存储目录唯一ID

[root@kafka102 kafka-kraft-3.2.1]$ bin/kafka-storage.sh random-uuid
QInSeC9bTwKVvMPJzsfe2A

（2）用该ID格式化kafka存储目录（三台节点）

[root@kafka102 kafka-kraft-3.2.1]$ bin/kafka-storage.sh format -t lb9OcpYJQZqOH37_wxTP2g -c config/kraft/server.properties

[root@kafka103 kafka-kraft-3.2.1]$ bin/kafka-storage.sh format -t LZgCQK4QQF-DVu3QGQX4-A -c config/kraft/server.properties

[root@kafka104 kafka-kraft-3.2.1]$ bin/kafka-storage.sh format -t LZgCQK4QQF-DVu3QGQX4-A -c config/kraft/server.properties

7）启动集群

[root@kafka102 kafka-kraft-3.2.1]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties

[root@kafka103 kafka-kraft-3.2.1]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties

[root@kafka104 kafka-kraft-3.2.1]$ bin/kafka-server-start.sh -daemon config/kraft/server.properties

8）停止集群

[root@kafka102 kafka-kraft-3.2.1]$ bin/kafka-server-stop.sh
[root@kafka103 kafka-kraft-3.2.1]$ bin/kafka-server-stop.sh
[root@kafka104 kafka-kraft-3.2.1]$ bin/kafka-server-stop.sh

3、Kafka-Kraft集群启动停止脚本

在/home/kafka/bin or /usr/local/bin/目录下创建文件xkafka-kraft.sh脚本文件

1）创建脚本

[root@kafka102 kafka-kraft-3.2.1]$ cd /usr/local/bin/
[root@kafka102 bin]$ sudo touch xkafka-kraft.sh
[root@kafka102 bin]$ sudo chown kafka:kafka xkafka-kraft.sh
[root@kafka102 bin]$ chmod 755 xkafka-kraft.sh
[root@kafka102 bin]$ vim xkafka-kraft.sh
#!/bin/bash

user=`whoami`
kafka_kraft_home='/opt/module/kafka-kraft-3.2.1'

case $1 in
    "start"){
        for i in kafka102 kafka103 kafka104
        do
            echo " --------启动 ${user}@$i kafka-kraft -------"
            ssh ${user}@$i "source /etc/profile; ${kafka_kraft_home}/bin/kafka-server-start.sh -daemon ${kafka_kraft_home}/config/kraft/server.properties"
        done
    };;
    "stop"){
        for i in kafka102 kafka103 kafka104
        do
            echo " --------停止 ${user}@$i kafka-kraft -------"
            ssh ${user}@$i "source /etc/profile; nohup ${kafka_kraft_home}/bin/kafka-server-stop.sh ${kafka_kraft_home}/config/kraft/server.properties > /dev/null 2>&1 &"
            sleep 5
        done
    };;
    *){
        echo "你启动的姿势不对"
        echo "  start   启动kafka kraft集群"
        echo "  stop    停止kafka kraft集群"
    };;
esac

2）启动集群命令

[root@kafka102 ~]$ xkafka-kraft.sh start

3）停止集群命令

[root@kafka102 ~]$ xkafka-kraft.sh stop