如何更改Kafka主题的副本数？

Question

生产者或管理员创建了Kafka主题后，如何更改该主题的副本数？

Answer 1

在第一步中，我们需要使用副本更改主题

./kafka-topics.sh --describe --zookeeper prod-az-p1-zk01.<domain>.prod:2181 --topic test2

然后在下一步中，我们需要确定需要同步副本的代理列表，并且需要重新平衡主题来执行此操作，创建一个 json 文件并定义所有 ISR 代理和主题

    {"version":1,
    "partitions":[
     {"topic":"test2","partition":0,"replicas":[0,10]},
     {"topic":"test2","partition":1,"replicas":[10,20]}
    ]}

最后我们需要重新平衡分区的主题

./kafka-reassign-partitions.sh --zookeeper prod-az-p1-zk01.<domain>.prod:2181 --reassignment-json-file /tmp/increase-replication-factor.json --execute

验证

[root@prod-az-p2-kafka02 bin]# ./kafka-topics.sh --describe --zookeeper prod-az-p1-zk01.<domain>.prod:2181 --topic test2
Topic: test2    TopicId: -LoL36ztSeyC8rzvnp4YMw PartitionCount: 2   ReplicationFactor: 2    Configs:
    Topic: test2    Partition: 0    Leader: 10  Replicas: 0,10  Isr: 10
    Topic: test2    Partition: 1    Leader: 20  Replicas: 10,20 Isr: 20,10

Answer 2

您也可以为此使用kafkactl：

# first run with --validate-only to see what kafkactl will do
kafkactl alter topic my-topic --replication-factor 2 --validate-only

# then do the replica reassignment
kafkactl alter topic my-topic --replication-factor 2

请注意，kafkactl 用于此目的的 Kafka API 仅适用于 Kafka ≥ 2.4.0。

免责声明：我是这个项目的贡献者

Answer 3

1.将所有主题复制到 json 文件中

#!/bin/bash
topics=`kafka-topics.sh --zookeeper localhost:2181 --list`

while read -r line; do lines+=("$line"); done <<<"$topics"
echo '{"version":1,
 "topics":['
 for t in $topics; do
     echo -e '     { "topic":' \"$t\" '},'
done

echo '  ]
}'

bash alltopics.sh > alltopics.json

2。运行 kafka-reassign-partitions.sh 生成重新平衡的文件

kafka-reassign-partitions.sh --zookeeper localhost:2181 --broker-list "0,1,2" --generate --topics-to-move-json-file alltopics.json > reassign.json

3.清理包含现有值和建议值的 reassign.json 文件

4.运行 kafka-reassign-partitions.sh 重新平衡主题

kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file reassign.json --execute

Answer 4

@Дмитрий-Шепелев 的脚本答案不包括针对具有多个分区的主题的解决方案。此更新版本可以：

#!/bin/bash

brokerids="1,2,3"
topics=`kafka-topics --list --zookeeper zookeeper:2181`

while read -r line; do lines+=("$line"); done <<<"$topics"
echo '{"version":1,
  "partitions":['
for t in $topics; do
    sep=","
    pcount=$(kafka-topics --describe --zookeeper zookeeper:2181 --topic $t | awk '{print $2}' | uniq -c |awk 'NR==2{print $1}')
    for i in $(seq 0 $[pcount - 1]); do
        if [ "${t}" == "${lines[-1]}" ] && [ "$[pcount - 1]" == "$i" ]; then sep=""; fi
        randombrokers=$(echo "$brokerids" | sed -r 's/,/ /g' | tr " " "\n" | shuf | tr  "\n" "," | head -c -1)
        echo "    {\"topic\":\"${t}\",\"partition\":${i},\"replicas\":[${randombrokers}]}$sep"
    done
done

echo '  ]
}'

注意：它还会随机化代理并为每个分区选择两个副本。因此，请确保脚本中的 brokerid 定义正确。

执行如下：

$ ./reassign.sh > reassign.json
$ kafka-reassign-partitions --zookeeper zookeeper:2181 --reassignment-json-file reassign.json --execute

Answer 5

编辑：我被证明是错误的 - 请检查excellent answer from Łukasz Dumiszewski.

为了完整起见，我暂时保留原来的答案。

---

---

我不认为你可以。通常它会像

./kafka-topics.sh --zookeeper localhost:2181 --alter --topic test2 --replication-factor 3

但它说

选项“[replication-factor]”不能与选项“[alter]”一起使用

有趣的是，您可以即时更改分区数量（在运行时执行此操作通常具有巨大的破坏性），但不能增加复制因子，这应该是透明的。但请记住，它是 0.10，而不是 10.0...请参阅此处获取增强请求https://issues.apache.org/jira/browse/KAFKA-1543

Answer 6

如果您想更改所有主题的复制因子，此脚本可能会对您有所帮助：

#!/bin/bash

topics=`kafka-topics --list --zookeeper zookeeper:2181`

while read -r line; do lines+=("$line"); done <<<"$topics"
echo '{"version":1,
  "partitions":[' > tmp.json
for t in $topics; do 
    if [ "${t}" == "${lines[-1]}" ]; then
        echo "    {\"topic\":\"${t}\",\"partition\":0,\"replicas\":[0,1,2]}" >> tmp.json
    else
        echo "    {\"topic\":\"${t}\",\"partition\":0,\"replicas\":[0,1,2]}," >> tmp.json
    fi
done

echo '  ]
}' >> tmp.json

kafka-reassign-partitions --zookeeper zookeeper:2181 --reassignment-json-file tmp.json --execute

Answer 7

Łukasz Dumiszewski 的answer 是正确的，但手动生成该文件有点困难。 幸运的是，有一些简单方法可以实现@Łukasz Dumiszewski 所说的。

• 如果您使用的是kafka-manager tool，从版本2.0.0.2 开始，您可以在主题视图的Generate Partition Assignment 部分更改复制因子。然后你应该点击Reassign Partitions来应用生成的分区分配（如果你选择不同的复制因子，你会得到一个警告，但你可以在之后点击Force Reassign）。

• 如果你安装了 ruby​​，你可以使用这个helper script

• 如果您更喜欢 nodejs，您也可以使用this gist 生成文件。

Answer 8

要增加给定主题的副本数量，您必须：

1.使用以下命令为现有主题指定额外的分区（假设从 2 增加到 3）

bin/kafktopics.sh --zookeeper localhost:2181 --alter --topic topic-to-increase --partitions 3

2。在自定义重新分配 json 文件中指定额外的副本

例如，您可以创建 increase-replication-factor.json 并将此内容放入其中：

{"version":1,
  "partitions":[
     {"topic":"topic-to-increase","partition":0,"replicas":[0,1,2]},
     {"topic":"topic-to-increase","partition":1,"replicas":[0,1,2]},
     {"topic":"topic-to-increase","partition":2,"replicas":[0,1,2]}
]}

3.使用带有 kafka-reassign-partitions 工具的 --execute 选项的文件

bin/kafka-reassign-partitions --zookeeper localhost:2181 --reassignment-json-file increase-replication-factor.json --execute

4.使用 kafka-topics 工具验证复制因子

bin/kafka-topics --zookeeper localhost:2181 --topic topic-to-increase --describe

Answer 9

如果您有很多分区，使用kafka-reassign-partitions 生成 Łukasz Dumiszewski 的答案（和官方文档）所需的 json 文件可以节省时间。下面是一个将 64 个分区主题从 1 台服务器复制到 2 台服务器而无需指定所有分区的示例：

expand_topic=TestTopic
current_server=111
new_servers=111,222
echo '{"topics": [{"topic":"'${expand_topic}'"}], "version":1}' > /tmp/topics-to-expand.json
/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file /tmp/topics-to-expand.json --broker-list "${current_server}" --generate | tail -1 | sed s/\\[${current_server}\\]/\[${new_servers}\]/g | tee /tmp/topic-expand-plan.json
/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file /tmp/topic-expand-plan.json --execute
/bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic ${expand_topic}

输出：

Topic:TestTopic PartitionCount:64   ReplicationFactor:2 Configs:retention.ms=6048000
    Topic: TestTopic    Partition: 0    Leader: 111 Replicas: 111,222   Isr: 111,222
    Topic: TestTopic    Partition: 1    Leader: 111 Replicas: 111,222   Isr: 111,222
    ....

Answer 10

要增加给定主题的副本数量，您必须：

1。在自定义重新分配 json 文件中指定额外的副本

例如，您可以创建 increase-replication-factor.json 并将此内容放入其中：

{"version":1,
  "partitions":[
     {"topic":"signals","partition":0,"replicas":[0,1,2]},
     {"topic":"signals","partition":1,"replicas":[0,1,2]},
     {"topic":"signals","partition":2,"replicas":[0,1,2]}
]}

2。将该文件与 kafka-reassign-partitions 工具的 --execute 选项一起使用

[或 kafka-reassign-partitions.sh - 取决于 kafka 包]

例如：

$ kafka-reassign-partitions --zookeeper localhost:2181 --reassignment-json-file increase-replication-factor.json --execute

3。使用 kafka-topics 工具验证复制因子

[或 kafka-topics.sh - 取决于 kafka 包]

 $ kafka-topics --zookeeper localhost:2181 --topic signals --describe

Topic:signals   PartitionCount:3    ReplicationFactor:3 Configs:retention.ms=1000000000
Topic: signals  Partition: 0    Leader: 2   Replicas: 0,1,2 Isr: 2,0,1
Topic: signals  Partition: 1    Leader: 2   Replicas: 0,1,2 Isr: 2,0,1
Topic: signals  Partition: 2    Leader: 2   Replicas: 0,1,2 Isr: 2,0,1

另请参阅：the part of the official documentation that describes how to increase the replication factor。