1 前言

一个问题，一篇文章，一出故事。
笔者部署完Confluent Kafka集群后发现Kafka connect集群由于配置定义的原因，默认的Topic只分配了一个复制因子（ReplicationFactor），即主题是基于单节点运行（不安全），于是想手动修改，集群环境如下，

如何部署Confluent Kafka集群?

2 最佳实践

2.1 查询Topic的状态

kafka-topics --bootstrap-server cfkafka02:9092,cfkafka03:9092,cfkafka04:9092 --describe --topic connect-status

可见如下显示，

Topic: connect-status   TopicId: jVo0WwEPQxSANL3S2i7k5g PartitionCount: 5       ReplicationFactor: 1    Configs: min.insync.replicas=2,cleanup.policy=compact,segment.bytes=1073741824
        Topic: connect-status   Partition: 0    Leader: 4       Replicas: 4     Isr: 4  Offline:
        Topic: connect-status   Partition: 1    Leader: 2       Replicas: 2     Isr: 2  Offline:
        Topic: connect-status   Partition: 2    Leader: 3       Replicas: 3     Isr: 3  Offline:
        Topic: connect-status   Partition: 3    Leader: 4       Replicas: 4     Isr: 4  Offline:
        Topic: connect-status   Partition: 4    Leader: 2       Replicas: 2     Isr: 2  Offline:

需要理解的是，
– connect-status Topic一共有5个Partition，Partition编号是0~4
– connect-status Topic的ReplicationFactor定义为“1”，即当前数据指分配到一个节点
– connect-status Topic的Leader、Replicas、Isr指示Partition分别分配到2或3或4号节点

2.2 修改Topic的ReplicationFactor

2.2.1 定义分配配置

vim update-connect-status.json

通过以上命令创建并从新定义分配，加入如下配置，

{
        "version":1,"partitions":[
        {"topic":"connect-status","partition":0,"replicas":[2,3,4]},
        {"topic":"connect-status","partition":1,"replicas":[2,3,4]},
        {"topic":"connect-status","partition":2,"replicas":[2,3,4]},
        {"topic":"connect-status","partition":3,"replicas":[2,3,4]},
        {"topic":"connect-status","partition":4,"replicas":[2,3,4]}
        ]
}

2.2.2 执行分区更新

kafka-reassign-partitions --bootstrap-server cfkafka02:9092,cfkafka03:9092,cfkafka04:9092 --reassignment-json-file update-connect-status.json --execute

可见如下成功提示，

Current partition replica assignment

{"version":1,"partitions":[{"topic":"connect-status","partition":0,"replicas":[2,3,4],"log_dirs":["any","any","any"]},{"topic":"connect-status","partition":1,"replicas":[2,3,4],"log_dirs":["any","any","any"]},{"topic":"connect-status","partition":2,"replicas":[2,3,4],"log_dirs":["any","any","any"]},{"topic":"connect-status","partition":3,"replicas":[2,3,4],"log_dirs":["any","any","any"]},{"topic":"connect-status","partition":4,"replicas":[2,3,4],"log_dirs":["any","any","any"]}]}

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

2.2.2 验证分区更新

kafka-reassign-partitions --bootstrap-server cfkafka02:9092,cfkafka03:9092,cfkafka04:9092 --reassignment-json-file update-connect-status.json --verify

可见如下提示，

Status of partition reassignment:
Reassignment of partition connect-status-0 is complete.
Reassignment of partition connect-status-1 is complete.
Reassignment of partition connect-status-2 is complete.
Reassignment of partition connect-status-3 is complete.
Reassignment of partition connect-status-4 is complete.

Clearing broker-level throttles on brokers 2,3,4
Clearing topic-level throttles on topic connect-status

2.2 修改其他Topic的ReplicationFactor

kafka-topics --bootstrap-server cfkafka02:9092,cfkafka03:9092,cfkafka04:9092 --describe --topic connect-configs
kafka-topics --bootstrap-server cfkafka02:9092,cfkafka03:9092,cfkafka04:9092 --describe --topic connect-offsets

注：请使用以上命令先查询后再按步骤修改，这里不再详述

参阅文档
===================
https://kafka.apache.org/documentation/#basic_ops_modify_topic