kafka二进制部署（Kraft方式,SASL认证）

集群节点地址示例：
192.168.242.203192.168.242.204192.168.242.205public-1public-2public-3节点1节点2节点3

1. 每个节点关闭防火墙

systemctl  stop firewalld.service
systemctl  disable firewalld.service

2. 检查每个节点中是否已安装JDK

2.1 查看是否安装jdk

java -version

2.2 如未安装jdk，安装方法

(1) 创建目录

mkdir -p /opt/jdk

(2) 上传java安装包到/opt/jdk

(3) 解压安装包

tar -xvf openlogic-openjdk-8u422-b05-linux-x64.tar.gz

(4) 重命名

mv openlogic-openjdk-8u422-b05-linux-x64 openjdk1.8

(5) 添加环境变量

vi /etc/profile
export JAVA_HOME=/opt/jdk/openjdk1.8
export PATH=$JAVA_HOME/bin:$PATH
source /etc/profile

(6) 检查jdk是否安装成功

java -version

3.修改主机名

(节点1执行)  hostnamectl set-hostname public-1 
(节点2执行)  hostnamectl set-hostname public-2
(节点3执行)  hostnamectl set-hostname public-3

4. 修改每个节点hosts

每个节点执行

cat >> /etc/hosts << EOF
192.168.242.203 public-1
192.168.242.204 public-2
192.168.242.205 public-3
EOF

5. 节点1安装kafka

(1) 创建kafka解压目录

mkdir -p /data/kafka

(2) 进入到 /data/kafka

cd /data/kafka

(3) 上传kafka_2.13-3.6.2.tgz到 /data/kafka目录下

(4) 解压kafka_2.13-3.6.2.tgz 到 /data/kafka

tar -xzvf kafka_2.13-3.6.2.tgz

(5) 新建数据目录

mkdir -p /data/kafka/kafka-data
mkdir -p /data/kafka/kafka-logs

(6) 进入到kafka_2.13-3.6.2/config/kraft 修改 Kraft 协议配置文件server.properties

vi server.properties节点唯一标识node.id=1集群信息
controller.quorum.voters=1@public-1:9093,2@public-2:9093,3@public-3:9093
节点信息listeners=SASL_PLAINTEXT://public-1:9092,CONTROLLER://public-1:9093公开监听信息advertised.listeners=SASL_PLAINTEXT://192.168.242.203:9092broker间通信名inter.broker.listener.name=SASL_PLAINTEXT安全协议映射见下文数据目录log.dirs=/data/kafka/kafka-data日志清理频率(时)log.retention.hours=3分区数num.partitions=3认证配置见下文
安全协议映射：

listener.security.protocol.map=CONTROLLER:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

认证配置：

sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
sasl.mechanism=PLAIN
allow.everyone.if.no.acl.found=false
super.users=User:admin
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
sasl.mechanism.controller.protocol=PLAIN

6. 节点2、节点3按步骤5进行配置

注意修改配置信息

节点2:

node.id=2
listeners=SASL_PLAINTEXT://public-2:9092,CONTROLLER://public-2:9093
advertised.listeners= SASL_PLAINTEXT://192.168.242.204:9092

节点3:

node.id=3
listeners=SASL_PLAINTEXT://public-3:9092,CONTROLLER://public-3:9093
advertised.listeners= SASL_PLAINTEXT://192.168.242.205:9092

其他配置和步骤5一样

7. 创建Kraft账号密码认证文件

每个节点都要创建

cd /data/kafka/kafka_2.13-3.6.2/config/kraft
vi kafka_server_jaas.conf

内容为：

KafkaServer {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="admin"
   password="password123!"
   user_user1="password123!"
   user_user2="password123!";
};

注意body中最后一行一定要有 ; 符号

8. 格式化存储目录

(1) 在任意一个节点上执行如下命令，用于生成一个唯一的集群 ID

cd /data/kafka/kafka_2.13-3.6.2/bin
./kafka-storage.sh random-uuid

(2) 所有节点执行如下命令，注意其中的 ID替换成上述生成的

cd /data/kafka/kafka_2.13-3.6.2
./bin/kafka-storage.sh format -t uXXhRtiEQd6rSQ8hsR82Qg -c ./config/kraft/server.properties

(3) 格式化操作完成之后，定义的 log.dirs 目录下多出 meta.properties 文件,文件中存储了当前的 kafka 节点的 id，当前节点属于哪个集群

9. 配置kafka服务的启动脚本

所有节点执行

cd /data/kafka/kafka_2.13-3.6.2/bin
cp kafka-server-start.sh kafka-server-start-sasl.sh

vi kafka-server-start-sasl.sh 将创建的kafka_server_jaas.conf地址添加到启动文件，修改部分为:

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx2G -Xms2G -Djava.security.auth.login.config=/data/kafka/kafka_2.13-3.6.2/config/kraft/kafka_server_jaas.conf"
fi

可以根据实际jvm参数在此处调整

10. 修改kafka日志输出目录