kafka 三节点部署

## 环境

|节点名|IP|
|-|-|
|kafka01|10.15.4.244|
|kafka02|10.15.4.245|
|kafka03|10.15.4.138|

|依赖|版本|
|-|-|
|Linux Server| Centos 9 |
|ZooKeeper|3.8.1|
|kafka|3.5.0|

## 部署

每个节点均需操作

创建相关目录
```bash
mkdir /root/kafka
mkdir /root/kafka/logs
```

下载 kafka
```bash
# https://downloads.apache.org/kafka/3.5.1/kafka_2.13-3.5.1.tgz
cd /root/kafka
wget https://downloads.apache.org/kafka/3.5.0/kafka_2.13-3.5.0.tgz
tar zxvf kafka_2.13-3.5.0.tgz
```

修改配置
```bash
cd /root/kafka/kafka_2.13-3.5.0
vim config/server.properties

# 修改以下内容
log.dirs=/root/kafka/logs
zookeeper.connect=10.15.4.244:2181,10.15.4.245:2181,10.15.4.138:2181

# 分节点配置
# 配置在节点 0 上
broker.id=0
listeners=PLAINTEXT://10.15.4.244:9092

# 配置在节点 1 上
broker.id=1
listeners=PLAINTEXT://10.15.4.245:9092

# 配置在节点 2 上
broker.id=2
listeners=PLAINTEXT://10.15.4.138:9092

# 以提供其他生产、消费程序的远程端口连接
# host.name=128.1.136.129

# 让 topic 可以真的删除
delete.topic.enable=true

# 禁止 consumer 或者 producer 使用时自动创建
auto.create.topics.enable=false

```

hosts

如果是内网 IP，可以考虑在 /etc/hosts 里配置内网 ip 映射到自己，比如
```
192.168.1.14 192.168.1.14
```

防火墙
```bash
# 开放端口
firewall-cmd --add-port=9092/tcp --permanent

# 重新加载防火墙配置
firewall-cmd --reload
```

每个节点上启动 kafka
```
cd /root/kafka/kafka_2.13-3.5.0
./bin/kafka-server-start.sh config/server.properties &
# 或者守护进程
./bin/kafka-server-start.sh -daemon config/server.properties

# 启动成功输出示例(最后几行)
[2023-06-30 15:57:37,184] INFO Kafka commitId: c97b88d5db4de28d (org.apache.kafka.common.utils.AppInfoParser)
[2023-06-30 15:57:37,184] INFO Kafka startTimeMs: 1688111857182 (org.apache.kafka.common.utils.AppInfoParser)
[2023-06-30 15:57:37,185] INFO [KafkaServer id=1] started (kafka.server.KafkaServer)
```

## 测试

创建 topic

```
# 在 kafka01(Broker) 上创建测试 Topic: test-topic，这里我们指定了3个副本、1个分区
./bin/kafka-topics.sh --create --bootstrap-server 10.15.4.244:9092 --replication-factor 3 --partitions 1 --topic test-topic
# Topic 在 kafka01 上创建后也会同步到集群中另外两个 Broker: kafka02, kafka03
```

查看 topic
```
# 我们可以通过命令列出指定 Broker 的 topic
./bin/kafka-topics.sh --list --bootstrap-server 10.15.4.245:9092
```

发送消息
```
# 这里我们向 Broker(id=0) 的 Topic=test-topic 发送消息
./bin/kafka-console-producer.sh --broker-list 10.15.4.244:9092 --topic test-topic

#消息内容
> test message
```

消费消息
```
# 在 Kafka02 上消费 Broker03 的消息
./bin/kafka-console-consumer.sh --bootstrap-server 10.15.4.138:9092 --topic test-topic --from-beginning
```

```
# 在 Kafka03 上消费 Broker02 的消息
./bin/kafka-console-consumer.sh --bootstrap-server 10.15.4.245:9092 --topic test-topic --from-beginning
```

然后均能收到消息
```
test message
```

这是因为这两个消费消息的命令是建立了两个不同的Consumer
如果我们启动Consumer指定Consumer Group Id就可以作为一个消费组协同工，1个消息同时只会被一个Consumer消费到
```
./bin/kafka-console-consumer.sh --bootstrap-server 10.15.4.138:9092 --topic test-topic --from-beginning --group testgroup

./bin/kafka-console-consumer.sh --bootstrap-server 10.15.4.245:9092 --topic test-topic --from-beginning --group testgroup
```

## 认证

参考 https://juejin.cn/post/7024713687986339853
参考 https://kafka.apache.org/documentation/#security_sasl_scram_brokerconfig

### 新增管理员 
```
# 新增
./bin/kafka-configs.sh -bootstrap-server 10.15.4.138:9092 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=Dev123456],SCRAM-SHA-512=[password=Dev123456]' --entity-type users --entity-name admin

# 查看
./bin/kafka-configs.sh -bootstrap-server 10.15.4.138:9092  --describe --entity-type users --entity-name admin

# 删除
./bin/kafka-configs.sh -bootstrap-server 10.15.4.138:9092 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name admin
```

### 启用 ACL

```
vim config/server.properties

# 多 broker，则每个 broker 均需如此配置
listeners=SASL_PLAINTEXT://10.15.4.245:9092
sasl.enabled.mechanisms=SCRAM-SHA-256
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:admin
```

### 配置 jaas

```
touch config/kafka-broker.jaas

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="Dev123456";
};
KafkaClient {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="Dev123456";
};
```

### 修改启动脚本

```
vim bin/kafka-server-start.sh

# 在前面几行增加如下启动参数
export KAFKA_OPTS=" $KAFKA_OPTS -Djava.security.auth.login.config=/home/abab/infra/kafka/kafka_2.13-3.5.0/config/kafka-broker.jaas"
```

重启 broker

### command-line

```
touch config/cmd-config

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="Dev123456";
```

因为加入身份验证，所以之后脚本的执行，需要带上 `--command-config config/cmd-config`

### 新增用户

增加两个用户，一个生产者，后面用来给topic发送消息，一个叫消费者，后面用来消费消息。

```
./bin/kafka-configs.sh --bootstrap-server 10.15.4.244:9092 --alter --add-config 'SCRAM-SHA-256=[password=Dev123456],SCRAM-SHA-512=[password=Dev123456]' --entity-type users --entity-name md_producer --command-config config/cmd-config

./bin/kafka-configs.sh --bootstrap-server 10.15.4.244:9092 --alter --add-config 'SCRAM-SHA-256=[password=Dev123456],SCRAM-SHA-512=[password=Dev123456]' --entity-type users --entity-name md_consumer --command-config config/cmd-config
```

### 授权

参考 https://docs.confluent.io/platform/current/kafka/authorization.html

```
./bin/kafka-acls.sh --bootstrap-server 10.15.4.244:9092 --add --allow-principal User:"md_producer" --producer --topic md --topic snapshot --topic order --topic transaction --command-config config/cmd-config

./bin/kafka-acls.sh --bootstrap-server 10.15.4.244:9092 --add --allow-principal User:"md_consumer" --consumer --topic md --topic snapshot --topic order --topic transaction --group '*' --command-config config/cmd-config
```

### 消费和生产脚本

加入授权后，消费和生产脚本需要指定权限配置，才能继续使用了

```
./bin/kafka-console-consumer.sh --bootstrap-server 10.15.4.244:9092 --topic md --consumer.config config/cmd-config
./bin/kafka-console-producer.sh --bootstrap-server 10.15.4.244:9092 --topic md --producer.config config/cmd-config
```

## 常用配置

## 其他

```
# 删除 topic
./bin/kafka-topics.sh --bootstrap-server 10.15.4.244:9092 --delete --topic test-topic
```

```
# 删除用户
./bin/kafka-configs.sh --bootstrap-server 10.15.4.244:9092 --alter --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name md_topic_creater --command-config config/cmd-config
```

```

# 增加用户对 topic 的创建和删除权限
./bin/kafka-acls.sh --bootstrap-server 10.15.4.244:9092 --add --allow-principal User:"md_topic_operator" --operation create --operation delete --topic test-topic --command-config config/cmd-config

# 删除用户对 topic 的创建和删除权限
./bin/kafka-acls.sh --bootstrap-server 10.15.4.244:9092 --remove --allow-principal User:"md_topic_operator" --operation create --operation delete --topic test-topic --command-config config/cmd-config

```
## rebalance and auto commit

http://mbukowicz.github.io/kafka/2020/09/12/implementing-kafka-consumer-in-java.html

## 开启 jxm

修改bin/kafka-server-start.sh脚本，第一行增加export JMX_PORT="9988"即可

```
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    export JMX_PORT="9988"
fi
```

## 问题

1. brokers 的 listeners 配置成内网 ip 时，开启 ACL 鉴权吼，访问过慢导致超时

参考 https://blog.csdn.net/LuckFu/article/details/124020426 和 https://blog.csdn.net/zhuan_long/article/details/104835368 
大致意思时，kafka 访问内网 ip 时，把 ip 当作域名访问了，所以绕了一大圈，导致过慢。最后在 /etc/hosts 添加 192.168.1.14 192.168.1.14，把自己映射到自己，来解决的。灵感来自于第二篇博客的评论区

逆风起笔