Fork me on GitHub

KAFKA版本滚动升级

字数统计: 1.5k字   |   阅读时长≈ 7分
环境

现有环境(已有现有环境)

主机名 IP 角色
node111/brokerid:111 10.0.10.111 kafka 0.8.2.2/zookeeper
node112/brokerid:112 10.0.10.112 kafka 0.8.2.2/zookeeper
node113/brokerid:113 10.0.10.113 kafka 0.8.2.2/zookeeper

升级后环境

主机名 IP 角色
node111/brokerid:111 10.0.10.111 kafka 2.1.0/zookeeper
node112/brokerid:112 10.0.10.112 kafka 2.1.0/zookeeper
node113/brokerid:113 10.0.10.113 kafka 2.1.0/zookeeper
查看当前KAFKA版本号
1
2
3
4
5
6
[root@node111 config]# cd /opt/kafka
[root@node111 kafka]# find ./libs/ -name \*kafka_\* | head -1 | grep -o '\kafka[^\n]*'
kafka_2.10-0.8.2.2.jar


注：这里 2.10是scala版本 0.8.2.2是kafka版本
升级JDK
1
2
新版本KAFKA要求JDK版本≥1.8,故需要提前升级JDK，否则会报如下经典错误：
kafka/Kafka : Unsupported major.minor version 52.0
1
2
tar -zxf jdk-8u201-linux-x64.tar.gz 
mv jdk1.8.0_201 /opt/
1
2
3
4
#JDK
export JAVA_HOME=/opt/jdk1.8.0_201
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
1
2
source /etc/profile
java -version
一、配置KFAKA-2.1.0

0.8.2核心配置-以node111为例

1
2
3
4
broker.id=111
port=9092
log.dirs=/var/log/kafka
zookeeper.connect=10.0.10.111:2181,10.0.10.112:2181,10.0.10.113:2181

2.1.0核心配置-以node111为例

1
2
3
4
5
6
broker.id=111
listeners=PLAINTEXT://10.0.10.111:9092
log.dirs=/var/log/kafka
zookeeper.connect=10.0.10.111:2181,10.0.10.112:2181,10.0.10.113:2181
inter.broker.protocol.version=0.8.2.2
log.message.format.version=0.8.2.2

重复以上步骤，将node112和node113 KAFKA安装配置完毕。

注意事项

1
2
3
4
5
-- 新版本与旧版本的log.dirs指定相同的路径，否则会导致信息丢失
-- 新版本于旧版本broker.id保持一致。
-- 需要在新版本的server.properties中，写入以下信息，这两个参数很重要！
     inter.broker.protocol.version=旧版本号
     log.message.format.version=旧版本号
二、 关闭老实例，启动新实例
  • 对于0.8.1.x版本
1
2
3
4
在Kafka-0.8.1.x版本中，Kafka源码中/kafka/admin/目录下有ShutdownBroker这个类，可以通过这个类实现优雅下线

方法：
sh kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper 172.18.055.21:2181 --broker 111 --num.retries 3 --retry.interval.ms 600
  • 对于0.8.2+版本
1
2
3
afka.admin.ShutdownBroker这个类被移除了，因此不能通过上述方法实现停机。

在Kafka-0.8.2+版本中，可以直接运行kill pid，但不要kill -9 pid。

将node111节点老版本KAFKA实例停止运行

1
2
3
4
[root@node111 config]# jps -m
30174 Kafka /opt/kafka/config/server.properties

[root@node111 config]# kill 30174   执行命令后可能需要稍等一会儿

启动node111节点新版本KAFKA实例

1
2
【新版本支持】以守护方式启动kafka
[root@node111]# /opt/kafka_2.12-2.1.0/bin/kafka-server-start.sh -daemon /opt/kafka_2.12-2.1.0/config/server.properties

其他节点也是通过这种方法实现服务停止和服务启动

验证kafka新版本

1
2
3
[root@node111 ~]# cd /opt/kafka_2.12-2.1.0/
[root@node111 kafka_2.12-2.1.0]# find ./libs/ -name \*kafka_\* | head -1 | grep -o '\kafka[^\n]*'
kafka_2.12-2.1.0.jar
三、更新协议版本
1
当集群中所有KAFKA节点都升级完毕后，就开始逐个升级protocol版本了，通过inter.broker.protocol.version这个配置选项，设置为2.1.0
1
sed -i 's/inter.broker.protocol.version=0.8.2.2/inter.broker.protocol.version=2.1.0/' /opt/kafka_2.12-2.1.0/config/server.properties
1
2
3
验证配置文件修改
grep "inter.broker.protocol.version" /opt/kafka_2.12-2.1.0/config/server.properties
inter.broker.protocol.version=2.1.0
1
2
3
4
重启KAFKA节点使新协议生效
/opt/kafka_2.12-2.1.0/bin/kafka-server-stop.sh
jps -m
/opt/kafka_2.12-2.1.0/bin/kafka-server-start.sh -daemon /opt/kafka_2.12-2.1.0/config/server.properties

其他kafka节点也需要逐一修改和节点重启

四、更新日志格式版本
1
通过log.message.format.version参数更新日志格式版本，设置为2.1.0
1
sed -i 's/log.message.format.version=0.8.2.2/log.message.format.version=2.1.0/' /opt/kafka_2.12-2.1.0/config/server.properties
1
2
3
验证配置文件修改
grep "log.message.format.version" /opt/kafka_2.12-2.1.0/config/server.properties
log.message.format.version=2.1.0
1
2
3
4
重启KAFKA节点使新协议生效
/opt/kafka_2.12-2.1.0/bin/kafka-server-stop.sh
jps -m
/opt/kafka_2.12-2.1.0/bin/kafka-server-start.sh -daemon /opt/kafka_2.12-2.1.0/config/server.properties

其他kafka节点也需要逐一修改和节点重启

五、验证新版本KAFKA数据
1
2
3
4
5
查看topic list
[root@node111 kafka_2.10-0.8.2.2]# bin/kafka-topics.sh --zookeeper 10.0.10.112:2181 --list
__consumer_offsets
garywu
garywu2
1
2
3
4
5
6
7
8
9
10
11
12
13
查看topic
[root@node111 kafka_2.10-0.8.2.2]# bin/kafka-topics.sh --describe --zookeeper 10.0.10.112:2181 --topic garywu2
Topic:garywu2	PartitionCount:10	ReplicationFactor:3	Configs:
	Topic: garywu2	Partition: 0	Leader: 111	Replicas: 111,112,113	Isr: 113,112,111
	Topic: garywu2	Partition: 1	Leader: 112	Replicas: 112,113,111	Isr: 113,112,111
	Topic: garywu2	Partition: 2	Leader: 113	Replicas: 113,111,112	Isr: 113,112,111
	Topic: garywu2	Partition: 3	Leader: 111	Replicas: 111,113,112	Isr: 113,112,111
	Topic: garywu2	Partition: 4	Leader: 112	Replicas: 112,111,113	Isr: 113,112,111
	Topic: garywu2	Partition: 5	Leader: 113	Replicas: 113,112,111	Isr: 113,112,111
	Topic: garywu2	Partition: 6	Leader: 111	Replicas: 111,112,113	Isr: 113,112,111
	Topic: garywu2	Partition: 7	Leader: 112	Replicas: 112,113,111	Isr: 113,112,111
	Topic: garywu2	Partition: 8	Leader: 113	Replicas: 113,111,112	Isr: 113,112,111
	Topic: garywu2	Partition: 9	Leader: 111	Replicas: 111,113,112	Isr: 113,112,111
注意事项
1
2
3
4
(1) 如果为生产环境不接受宕机，那么请严格按照此步骤进行滚动升级
(2) 当brokers成功升级后，设定版本协议、日志格式版本以及重启操作，可以在业务影响较小的时间段进行逐一重启。
(3) 由于kafka新版本与0.8+版本的客户端API兼容，因此在完成brokers升级后，即便protocal和log日志格式版本设置为旧版本，也不影响调用。
(4) KAFKA broker的PID被kill之后，会触发KAFKA Partation的leader重新选举，在选举过程中，producer发送的消息会发送失败；与此同时，consumer也无法消费到发送失败的消息。

参考：石头头头的博客

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true


不论我写了什么...
其实都是错的......