Kafka节点变化怎么办?

broker节点数量发生变化时,需要对已有topics的分区和副本进行重新分配,按照以下步骤来测试一下。
--bootstrap-server指定的IP替换为自己的服务器地址

指定主题

首先创建一个json文件,指定要进行重新分配的主题名称

[root@localhost kafka]# vim topics-to-move.json
{
 "topics": [
    {"topic": "test"}
 ],
 "version": 1
}

先查看一下当前test主题的分配情况,目前分配在0、1、2三个节点上

[root@localhost kafka]# ./bin/kafka-topics.sh --bootstrap-server IP:9092 --topic test --describe
Topic: test TopicId: 8ZUQdSBFSX6bifqtfkqtfw PartitionCount: 3   ReplicationFactor: 3    Configs: segment.bytes=1073741824
    Topic: test Partition: 0    Leader: 0   Replicas: 2,1,0 Isr: 2,1,0
    Topic: test Partition: 1    Leader: 1   Replicas: 0,2,1 Isr: 1,2,0
    Topic: test Partition: 2    Leader: 2   Replicas: 1,0,2 Isr: 0,2,1

生成新的分配计划

使用kafka-reassign-partitions.sh执行以下命令生成一个新的分配计划,其中--broker-list指定要重新分配到哪些broker上(添加或删除后的节点列表)

[root@localhost kafka]# ./bin/kafka-reassign-partitions.sh --bootstrap-server IP:9092 --topics-to-move-json-file topic-to-move.json --broker-list "0,1,2,3" --generate
Current partition replica assignment
{"version":1,"partitions":[{"topic":"test","partition":0,"replicas":[2,1,0],"log_dirs":["any","any","any"]},{"topic":"test","partition":1,"replicas":[1,0,2],"log_dirs":["any","any","any"]},{"topic":"test","partition":2,"replicas":[0,2,1],"log_dirs":["any","any","any"]}]}

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"test","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"test","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"test","partition":2,"replicas":[2,3,0],"log_dirs":["any","any","any"]}]}

可以看到Current partition replica assignment是当前分配情况,Proposed partition reassignment configuration是生成的新的分配计划,其中多了一个节点3

创建副本存储计划

把上一步生成的新的分配计划存储到json文件

[root@localhost kafka]#  vim increase-replication-factor.json
{"version":1,"partitions":[{"topic":"test","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"test","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"test","partition":2,"replicas":[2,3,0],"log_dirs":["any","any","any"]}]}

执行副本存储计划

[root@localhost kafka]# ./bin/kafka-reassign-partitions.sh --bootstrap-server IP:9092 --reassignment-json-file increase-replication-factor.json --execute
Current partition replica assignment

{"version":1,"partitions":[{"topic":"test","partition":0,"replicas":[2,1,0],"log_dirs":["any","any","any"]},{"topic":"test","partition":1,"replicas":[1,0,2],"log_dirs":["any","any","any"]},{"topic":"test","partition":2,"replicas":[0,2,1],"log_dirs":["any","any","any"]}]}

Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for test-0,test-1,test-2

验证副本存储计划

[root@localhost kafka]# ./bin/kafka-reassign-partitions.sh --bootstrap-server IP:9092 --reassignment-json-file increase-replication-factor.json --verify
Status of partition reassignment:
Reassignment of partition test-0 is complete.
Reassignment of partition test-1 is complete.
Reassignment of partition test-2 is complete.

Clearing broker-level throttles on brokers 0,1,2,3
Clearing topic-level throttles on topic test

重新查看test主题的分配情况,现在0 、1、2、3四个节点都已完成分配

[root@localhost kafka]# ./bin/kafka-topics.sh --bootstrap-server IP:9092 --describe --topic test
Topic: test TopicId: 8ZUQdSBFSX6bifqtfkqtfw PartitionCount: 3   ReplicationFactor: 3    Configs: segment.bytes=1073741824
    Topic: test Partition: 0    Leader: 0   Replicas: 0,1,2 Isr: 2,1,0
    Topic: test Partition: 1    Leader: 1   Replicas: 1,2,3 Isr: 1,2,3
    Topic: test Partition: 2    Leader: 2   Replicas: 2,3,0 Isr: 0,2,3

总结:

  1. 指定要操作的主题
  2. 生成副本存储计划
  3. 执行副本存储计划
  4. 验证副本存储计划
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • kafka的定义:是一个分布式消息系统,由LinkedIn使用Scala编写,用作LinkedIn的活动流(Act...
    时待吾阅读 5,361评论 1 15
  • [TOC]在上一节对副本机制的实现进行了分析,其中提到Broker能够处理来自KafkaController的Le...
    tracy_668阅读 1,926评论 0 3
  • kafka是一个分布式的基于发布/订阅模式的消息队列(Message Queue),主要应用于大数据实时处理领域。...
    dev_winner阅读 548评论 0 2
  • 1 kafka应用系统框架图 作为一款典型的消息中间件产品,kafka系统仍然由producer、broker、c...
    movee阅读 1,078评论 0 0
  • 为什么需要使用kafka 从本质上来讲,是因为互联网发展太快,使用单体架构无疑会是的体量巨大。而微服务架构可以很好...
    wxxhfg阅读 2,447评论 2 20