一、升级目的

为了使用zstd压缩算法。

二、准备

提前下载升级包，并放在/opt下

cd /opt && wget https://archive.apache.org/dist/kafka/2.1.0/kafka_2.11-2.1.0.tgz

三、集群环境

1、节点数：172.18.1.1 ~ 172.18.1.10，由10个节点组成kafka集群

2、安装目录：/opt/kafka_2.11-2.0.0

四、升级方式

轮询升级

五、修改目标版本kafka_2.11-2.1.0的配置

先操作172.18.1.1

修改配置，添加以下内容

# cd /opt/kafka_2.11-2.1.0 && vim config/server.propertie 
inter.broker.protocol.version = 2.0.0

注意：同步其他配置到目标版本，特别是数据存储目录

六、重启kafka

1、 停止旧版本kafka 2.0.0，使用kafka自带stop脚本bin下的kafka-server-stop.sh

2、启动新版本kafka 2.1.0

cd /opt/kafka_2.11-2.1.0 && ./bin/kafka-server-start.sh -daemon config/server.properties

3、测试topic test ，确认收发消息正常

# 发送消息（注意端口号为配置文件里面的端口号）
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

#消费消息（可能端口号与配置文件保持一致，或与发送端口保持一致）
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning   #加了--from-beginning 重头消费所有的消息
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test         #不加--from-beginning 从最新的一条消息开始消费

4、其他节点重复以上操作步骤

七、再次修改目标版本kafka_2.11-2.1.0的配置

先操作172.18.1.1

修改配置，将inter.broker.protocol.version修改为 inter.broker.protocol.version = 2.1.0

八、重启kafka 2.1.0

cd /opt/kafka_2.11-2.1.0

# 停止
./bin/kafka-server-stop.sh

# 确认kafka已停止
ps aux |grep kafka

# 启动
./bin/kafka-server-start.sh -daemon config/server.properties

测试topic test ，确认收发消息正常

# 发送消息（注意端口号为配置文件里面的端口号）
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

#消费消息（可能端口号与配置文件保持一致，或与发送端口保持一致）
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning   #加了--from-beginning 重头消费所有的消息
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test         #不加--from-beginning 从最新的一条消息开始消费

至此，升级kafka已操作完成。