集群扩容后批量重分配所有Kafka Topic的简便方案咨询

针对1000+Topic的批量重分配需求，直接用Kafka官方工具结合优化后的Shell脚本就能解决，以下是具体步骤：

1. 导出所有非内部Topic列表 #

先过滤掉Kafka内部Topic（以__开头的），避免不必要的操作：

kafka-topics.sh --list --bootstrap-server <你的Kafka集群地址:端口> | grep -v "^__" > all_topics.txt

2. 批量生成重分配计划JSON #

创建一个脚本自动生成完整的重分配配置文件，替换NEW_BROKERS为你的新Broker ID列表，BOOTSTRAP_SERVER为集群地址：

#!/bin/bash
NEW_BROKERS="101,102,103" # 替换为你的新Broker ID，用逗号分隔
BOOTSTRAP_SERVER="kafka-cluster:9092" # 替换为你的Kafka bootstrap地址

# 初始化重分配JSON文件
echo '{"version":1,"partitions":[' > reassign_partitions.json
first_entry=1

# 遍历所有Topic生成分区分配条目
while read topic; do
    # 生成单个Topic的重分配建议
    reassign_result=$(kafka-reassign-partitions.sh --bootstrap-server "$BOOTSTRAP_SERVER" \
        --generate \
        --topics-to-move-json-file <(echo "{\"topics\":[{\"topic\":\"$topic\"}],\"version\":1}") \
        --broker-list "$NEW_BROKERS")
    
    # 提取建议的分区配置部分（跳过前两行提示文本）
    partition_config=$(echo "$reassign_result" | tail -n +3)
    
    # 处理JSON格式的逗号分隔
    if [ "$first_entry" -eq 1 ]; then
        echo "$partition_config" >> reassign_partitions.json
        first_entry=0
    else
        echo ",$partition_config" >> reassign_partitions.json
    fi
done < all_topics.txt

# 闭合JSON文件
echo ']}' >> reassign_partitions.json

给脚本加执行权限后运行：chmod +x generate_reassign.sh && ./generate_reassign.sh

3. 验证并执行重分配 #

• 先验证重分配计划的合理性（可选但推荐）：

kafka-reassign-partitions.sh --bootstrap-server "$BOOTSTRAP_SERVER" --validate --reassignment-json-file reassign_partitions.json

• 执行重分配：

kafka-reassign-partitions.sh --bootstrap-server "$BOOTSTRAP_SERVER" --execute --reassignment-json-file reassign_partitions.json

4. 监控重分配进度 #

实时查看重分配完成情况：

kafka-reassign-partitions.sh --bootstrap-server "$BOOTSTRAP_SERVER" --verify --reassignment-json-file reassign_partitions.json

关键注意事项 #

• 内部Topic处理：__consumer_offsets、__transaction_state等内部Topic，建议单独处理或者让Kafka自动平衡，不要包含在批量操作中
• 分批执行：如果Topic数量过大，可将all_topics.txt分割为多个小文件，分批执行重分配，避免集群压力过高
• 备份配置：执行前建议导出所有Topic的配置和分区信息，用于异常回滚：

kafka-topics.sh --describe --bootstrap-server "$BOOTSTRAP_SERVER" > topic_backup.txt

内容的提问来源于stack exchange，提问作者Harikrishna Gonapuri