Kafka 3.3.1 Kraft 多端口协议搭建
序
Kafka 3.3.1 已经出来挺久了，很多公司还停留在 1.X/2.X 甚至 0.8 版本的 kafka，不是说不能用，但是用起来真的糟糕，况且现在 Kraft 已经正式推出了，早就该更新了。
本篇文章从实践角度出发，使用真实的搭建手册改编，带领大家搭建多网卡、多端口、多认证的 Kafka 3.3.1 on Kraft。
预计效果，内部通信的非认证、内部数据传输的 sasl-plain、外部数据消费的 sasl-ssl。
版本
Kafka 2.12-3.3.1 with KRaft
auth: huangyichun maggot
服务器环境
在搭建时，必须明确我们集群的情况。本手册使用双网卡进行配置演示，每台服务器有两个网卡（*下面 outip 为虚拟网卡）
再生产环境，搭建任何集群都应该有操作步骤，如这篇文档一样。
快速配置虚拟网卡步骤
cd /etc/sysconfig/network-scripts # 配置目录
cp ifcfg-ens33 ifcfg-ens33:0 # centos7 默认为ens33
vim ifcfg-ens33\:0 # 编辑对应虚拟网卡
# 更改以下配置：
DEVICE="ens33:0"
NAME="ens33:0"
IPADDR="192.168.111.xxx"
systemctl restart network # 重启网络服务
ifconfig # 验证
配置 ansible hosts
vim /etc/ansible/hosts
# 追加以下内容
[in_ip_kafka3]
192.168.111.128
192.168.111.129
192.168.111.130
[in_ip_kafka3:vars]
ansible_ssh_user='root'
ansible_ssh_pass='root'
如果还未进行 ansible 安装，请移步：
todo ansible 安装手册
环境验证
主目录
/usr/local/kafka/kafka_2.12-3.3.1
数据目录
log.dirs=/data1/kafka/data,/data2/kafka/data
当前虚机并没有这么多挂载盘，只做演示作用
查看磁盘挂载状态
ansible in_ip_kafka3 -m shell -a "df -h"
# 如果可以，在挂载盘的时候最好将kafka data盘 atime关闭。
查看相关内核参数
ansible in_ip_kafka3 -m shell -a "ulimit -a"
# 特别关心 openfiles 参数，一般在大数据中，在 102400 级别
安装 java 14.0.2
安装目录
/opt/jdk/jdk-14.0.2
下载路径记得贴一下 todo
其他 jdk 也是可以的
分发并解压：
ansible in_ip_kafka3 -m copy -a "src=/opt/jdk/jdk-14.0.2_linux-x64_bin.tar.gz dest=/opt/jdk/"
ansible in_ip_kafka3 -m shell -a "cd /opt/jdk; tar -zxvf jdk-14.0.2_linux-x64_bin.tar.gz"
验证 java 安装：
ansible in_ip_kafka3 -m shell -a "ls -ltr /opt/jdk/"
ansible in_ip_kafka3 -m shell -a "/opt/jdk/jdk-14.0.2/bin/java -version"
安装 kafka3.3.1
下载地址
kafka.apache.org
分发解压
ansible in_ip_kafka3 -m copy -a "src=kafka_2.12-3.3.1.tgz dest=/opt/kafka/"
ansible in_ip_kafka3 -m shell -a "cd /opt/kafka; tar -zxvf kafka_2.12-3.3.1.tgz"
ansible in_ip_kafka3 -m shell -a "ls -ltr /opt/kafka"
kraft conf 目录
/opt/kafka/kafka_2.12-3.3.1/config/kraft/server.properties
ansible in_ip_kafka3 -m shell -a "ls -ltr /opt/kafka/kafka_2.12-3.3.1/config/kraft/server.properties"
node.id 映射;node.id 是 kafka 中的节点标识，选举投票也会使用这个 id 进行判断
192.168.111.128 -> 111128
192.168.111.129 -> 111129
192.168.111.130 -> 111130
controller.quorum.voters
controller.quorum.voters=111128@192.168.111.128:9093,111129@192.168.111.129:9093,111130@192.168.111.130:9093
配置更改
vim /opt/kafka/kafka_2.12-3.3.1/config/kraft/server.properties
# The role of this server. Setting this puts us in KRaft mode
process.roles=broker,controller
# The node id associated with this instance's roles
# 采用后16位作为nodeid不足十进制3位的补0,如：192.168.111.128 -> 111128
node.id=${NODE_ID}
# The connect string for the controller quorum
controller.quorum.voters=111128@192.168.111.128:9093,111129@192.168.111.129:9093
# The address the socket server listens on.
listeners=INNER://0.0.0.0:9091,OUTTER://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
# Name of listener used for communication between brokers.
inter.broker.listener.name=CONTROLLER
# Listener name, hostname and port the broker will advertise to clients.
# If not set, it uses the value for "listeners".
advertised.listeners=INNER://${INNER_IP}:9091,OUTTER://${OUTTER_IP}:9092
# A comma-separated list of the names of the listeners used by the controller.
# If no explicit mapping set in `listener.security.protocol.map`, default will be using PLAINTEXT protocol
# This is required if running in KRaft mode.
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,INNER:PLAINTEXT,OUTTER:PLAINTEXT
############################# Log Basics #############################
# A comma separated list of directories under which to store log files
log.dirs=/data1/kafka/data,/data2/kafka/data
配置分发：
ansible in_ip_kafka3 -m copy -a "src=/opt/kafka/kafka_2.12-3.3.1/config/kraft/server.properties dest=/opt/kafka/kafka_2.12-3.3.1/config/kraft/server.properties"
别只分发了不改配置项… node.id & advertised.listeners
配置释义
在 Kafka3.X 中，正式开启了 Kraft 的使用，与之前的 zookeeper 模式其实基本一致，只是将元数据搬迁到了 kafka 内部中进行管理。所以新增了部分选项，需要指定每个节点其作用。
process.roles
可以配置 broker、controller，标示在 KRaft 模式下，该节点充当什么角色：
- broker 当数据存储节点
- controller 当元数据管理节点
- broker,controller 两者都
- 没有设置，假定使用 zookeeper 进行元数据原理。
controller.quorum.voters
必须所有节点都设置该配置，标识哪些节点是 Quorum 的投票节点，必须包含所有的投票节点。跟 zookeeper 配置中的 zookeeper.connect 配置一样。
controller.quorum.voters=111128@192.168.111.128:9093,111129@192.168.111.129:9093
格式为 node.id@ip:port
- node.id
上面的映射规则映射出来的 id，一般建议使用后 255*255 补齐零，当然原则是避免重复。
- ip:port
这个 ip:port 在双网卡时请填写 CONTROLLER 的 ip:port，使用这个进行 kafka 集群内部的通讯规则。
inter.broker.listener.name
kafka 内部通讯时使用哪个 listener。一般建议配置没有进行加密的通讯端口，然后防火墙进行相关配置，仅开放对应的网段进行访问。
advertised.listeners
对外开放访问的接口
advertised.listeners=INNER:// I N N E R I P : 9091 , O U T T E R : / / {INNER_IP}:9091,OUTTER:// INNERIP:9091,OUTTER://{OUTTER_IP}:9092
格式为 LISTENER_NAME://ip:port[,……]
- LISTENER_NAME
在 listeners 中定义的访问类型
- ip:port
这个 ip:port 在双网卡时请填写 LISTENER_NAME 的 ip:port，但如果之前绑定的是 0.0.0.0，请指定到具体的端口！
listener.security.protocol.map
配置具体的 LISTENER_NAME 对应的加密方式
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,INNER:PLAINTEXT,OUTTER:PLAINTEXT
格式为 LISTENER_NAME : 加密方式
这里的加密方式分为：
- PLAINTEXT
- SSL
- SASL_PLAINTEXT
- SASL_SSL
这里我们都先使用 PLAINTEXT 进行配置，后续会将 OUTTER 的机密方式进行更改
controller.listener.names
进行选举等操作时占用的通讯端口，配置为 listeners 中的某一个。
log.dirs
配置数据的存储目录，在 kafka 中，log 就是数据的意思。
log.dirs=/data1/kafka/data,/data2/kafka/data
这里我们使用两个挂载盘进行配置，如果是只有一个挂载盘，那么不用这样进行配置。
JVM 相关配置
MEM 使用更改
vim /opt/kafka/kafka_2.12-3.3.1/bin/kafka-server-start.sh
# 找到 KAFKA_HEAP_OPTS, 添加下面的配置参数：
KAFKA_HEAP_OPTS="-Xmx2G -Xms2G -Xmn1G -XX:MetaspaceSize=256M"
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi
分发到其他节点：
ansible in_ip_kafka3 -m copy -a "src=/opt/kafka/kafka_2.12-3.3.1/bin/kafka-server-start.sh dest=/opt/kafka/kafka_2.12-3.3.1/bin/kafka-server-start.sh"
日志相关配置
默认 kafka 日志输出没有指定位置，可以在这里进行配置
vim /opt/kafka/kafka_2.12-3.3.1/bin/kafka-run-class.sh
# 搜索到 LOG_DIR 处，然后添加下面一行
LOG_DIR=/data1/kafka/logs
if [ "x$LOG_DIR" = "x" ]; then
LOG_DIR="$base_dir/logs"
fi
# 搜索到 JAVA_HOME，将 JAVA 命令指向刚刚安装14的位置
# Which java to use
if [ -z "$JAVA_HOME" ]; then
JAVA="java"
else
JAVA="$JAVA_HOME/bin/java"
fi
# 添加下面一行
JAVA="/opt/jdk/jdk-14.0.2/bin/java"
分发 kafka-run-class.sh 到其他节点
ansible in_ip_kafka3 -m copy -a "src=/opt/kafka/kafka_2.12-3.3.1/bin/kafka-run-class.sh dest=/opt/kafka/kafka_2.12-3.3.1/bin/kafka-run-class.sh"
初始化 kafka 集群
严禁在已经生产中进行此步骤，会清空所有已经存在数据！
随机集群标识
随机一个 uuid 作为 kafka 集群标识：
bash /opt/kafka/kafka_2.12-3.3.1/bin/kafka-storage.sh random-uuid
输出一个结果：
M6ZxYfO3TByOQQoc1YdBQw
那么我们在此集群使用 M6ZxYfO3TByOQQoc1YdBQw 为整个集群的标识。
节点 format 操作
此操作会格式化 server.properties 中的 log.dir 路径，严禁在已经生产中使用！
/opt/kafka/kafka_2.12-3.3.1/bin/kafka-storage.sh format -t M6ZxYfO3TByOQQoc1YdBQw -c /opt/kafka/kafka_2.12-3.3.1/config/kraft/server.properties
ansible 操作：
ansible in_ip_kafka3 -m shell -a "rm -rf /data*/kafka"
ansible in_ip_kafka3 -m shell -a "/opt/kafka/kafka_2.12-3.3.1/bin/kafka-storage.sh format -t M6ZxYfO3TByOQQoc1YdBQw -c /opt/kafka/kafka_2.12-3.3.1/config/kraft/server.properties"
启动命令
ansible in_ip_kafka3 -m shell -a "/opt/kafka/kafka_2.12-3.3.1/bin/kafka-server-start.sh -daemon /opt/kafka/kafka_2.12-3.3.1/config/kraft/server.properties"
# 查看是否有对应日志，且正常
ansible in_ip_kafka3 -m shell -a "ls -ltr /data1/kafka/logs"
ansible in_ip_kafka3 -m shell -a "tail /data1/kafka/logs/*"
ansible in_ip_kafka3 -m shell -a "jps"
排查错误
Connection to node 111130 (/192.168.111.129:9093) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
如果发现有这个问题，请查看是否关闭掉了防火墙。
ansible in_ip_kafka3 -m shell -a "systemctl status firewalld"
关闭步骤：
ansible in_ip_kafka3 -m shell -a "systemctl stop firewalld; systemctl mask firewalld"
