kafka broker迁移

此脚本可实现旧 Broker 15 → 新 Broker 69 的健康分区批量迁移真正支持中断恢复可以按 10/20 个 Partition 一批执行直到旧 Broker 的ReplicaCount全部降到 0脚本可实现禁止在已有 reassignment 时提交新批次cd /mnt/allin_data/allin/middleware/software/kafka_2.13-2.7.2 cat kafka_migrate_healthy_partitions.sh SCRIPT #!/bin/bash set -uo pipefail # # Kafka Healthy Partition Migration # # Old brokers : 1,2,3,4,5 # New brokers : 6,7,8,9 # Target RF : 3 # # 每个 Partition 直接 # # old replicas # ↓ # 6,7,8 / 7,8,9 / 8,9,6 / 9,6,7 # # 前提 # URP 0 # Unavailable 0 # # VERSION1.0.0 KAFKA_HOME$(cd $(dirname $0) pwd) TOPICS${KAFKA_HOME}/bin/kafka-topics.sh REASSIGN${KAFKA_HOME}/bin/kafka-reassign-partitions.sh ZKSHELL${KAFKA_HOME}/bin/zookeeper-shell.sh BOOTSTRAP10.40.0.143:9092 ZKcementindustrynode16:2181 BATCH_SIZE20 MAX_BATCHES1 WAIT_INTERVAL10 WAIT_TIMEOUT7200 STATE_DIR/tmp/kafka_healthy_migrate_20260820 EXECUTE0 RESUME0 RUN_ALL0 log() { echo [$(date %F %T)] $* } die() { log ERROR: $* exit 1 } usage() { cat EOF Usage: $0 [OPTIONS] Options: --batch-size N 每批 Partition 数 default: 20 --max-batches N 单次最多执行多少批 default: 1 --execute 正式执行 默认仅生成计划 --resume 继续已有 checkpoint --all 一直处理到完成 --state-dir DIR checkpoint目录 --wait-timeout SEC 单批最长等待时间 default: 7200 Examples: # dry-run $0 --batch-size 10 # 第一批10个 $0 --batch-size 10 --execute # 下一批20个 $0 --batch-size 20 --execute --resume EOF } while [[ $# -gt 0 ]]; do case $1 in --batch-size) BATCH_SIZE$2 shift 2 ;; --max-batches) MAX_BATCHES$2 shift 2 ;; --execute) EXECUTE1 shift ;; --resume) RESUME1 shift ;; --all) RUN_ALL1 shift ;; --state-dir) STATE_DIR$2 shift 2 ;; --wait-timeout) WAIT_TIMEOUT$2 shift 2 ;; -h|--help) usage exit 0 ;; *) die Unknown option $1 ;; esac done mkdir -p $STATE_DIR PLAN${STATE_DIR}/plan.tsv COMPLETED${STATE_DIR}/completed.tsv HISTORY${STATE_DIR}/history.log SNAPSHOT${STATE_DIR}/cluster.txt touch $COMPLETED $HISTORY # # helpers # active_reassignment_count() { $REASSIGN \ --bootstrap-server $BOOTSTRAP \ --list \ 2/dev/null \ | awk /Current partition reassignments/ {next} /No partition reassignments/ {next} /[A-Za-z0-9._-]-[0-9]/ {n} END {print n0} } normalize_csv() { echo $1 | tr , \n | sed /^$/d | sort -n | paste -sd, - } csv_count() { if [[ -z $1 ]]; then echo 0 else echo $1 | awk -F, {print NF} fi } snapshot() { $TOPICS \ --bootstrap-server $BOOTSTRAP \ --describe \ ${SNAPSHOT}.tmp 2${STATE_DIR}/snapshot.err if [[ $? -ne 0 ]]; then cat ${STATE_DIR}/snapshot.err return 1 fi mv ${SNAPSHOT}.tmp $SNAPSHOT } urp_count() { $TOPICS \ --bootstrap-server $BOOTSTRAP \ --describe \ --under-replicated-partitions \ 2/dev/null | awk /Topic:/ {n} END {print n0} } unavailable_count() { $TOPICS \ --bootstrap-server $BOOTSTRAP \ --describe \ --unavailable-partitions \ 2/dev/null | awk /Topic:/ {n} END {print n0} } get_state() { local t$1 local p$2 awk \ -v T$t \ -v P$p { topic part leader replicas isr for(i1;iNF;i) { if($iTopic:) topic$(i1) if($iPartition:) part$(i1) if($iLeader:) leader$(i1) if($iReplicas:) replicas$(i1) if($iIsr:) isr$(i1) } if(topicT partP) { print topic \t part \t leader \t replicas \t isr exit } } $SNAPSHOT } is_completed() { local topic$1 local part$2 awk -F\t \ -v t$topic \ -v p$part $1t $2p {found1} END {exit !found} $COMPLETED } mark_completed() { local topic$1 local part$2 local target$3 if ! is_completed $topic $part; then printf %s\t%s\t%s\t%s\n \ $topic \ $part \ $target \ $(date %F %T) \ $COMPLETED fi } # # Broker health # check_brokers() { local ids ids$( $ZKSHELL $ZK ls /brokers/ids 2/dev/null | grep ^\[ | tail -1 ) log Brokers: $ids for id in 6 7 8 9; do echo $ids | grep -Eq (^|[^0-9])${id}([^0-9]|$) || die New Broker $id not online done } # # 生成迁移计划 # # 仅选择仍然包含旧 Broker 1~5 的 Partition # # 最终目标轮转 # 6,7,8 # 7,8,9 # 8,9,6 # 9,6,7 # generate_plan() { snapshot || die snapshot failed $PLAN local idx0 while IFS$\t read -r topic part leader replicas isr do [[ -z $topic ]] continue local has_old0 IFS, read -ra R $replicas for r in ${R[]}; do if (( r 1 r 5 )); then has_old1 fi done [[ $has_old -eq 0 ]] continue local target case $((idx % 4)) in 0) target6,7,8 ;; 1) target7,8,9 ;; 2) target8,9,6 ;; 3) target9,6,7 ;; esac printf %d\t%s\t%s\t%s\t%s\n \ $idx \ $topic \ $part \ $replicas \ $target \ $PLAN idx$((idx1)) done ( awk { t p l r s for(i1;iNF;i) { if($iTopic:) t$(i1) if($iPartition:) p$(i1) if($iLeader:) l$(i1) if($iReplicas:) r$(i1) if($iIsr:) s$(i1) } if(t!) print t \t p \t l \t r \t s } $SNAPSHOT ) log Migration plan partitions: $(wc -l $PLAN) } ACTIVE$(active_reassignment_count) if [[ $ACTIVE -ne 0 ]]; then log Kafka currently has $ACTIVE active partition reassignments. $REASSIGN \ --bootstrap-server $BOOTSTRAP \ --list die Wait for current reassignment to complete before starting another batch. fi # # 构建下一批 # build_batch() { local outfile$1 $outfile snapshot || die snapshot failed while IFS$\t read -r idx topic part original target do [[ -z $topic ]] continue if is_completed $topic $part; then continue fi local state state$(get_state $topic $part) [[ -z $state ]] die Partition disappeared: $topic-$part local ct cp leader replicas isr IFS$\t read -r \ ct cp leader replicas isr $state local norm_target local norm_rep local norm_isr norm_target$(normalize_csv $target) norm_rep$(normalize_csv $replicas) norm_isr$(normalize_csv $isr) # 已经完全迁完 if [[ $norm_rep $norm_target $norm_isr $norm_target $(csv_count $isr) -eq 3 ]]; then mark_completed $topic $part $target continue fi # 已经执行 reassignment但新副本还在追数据 if [[ $norm_rep $norm_target ]]; then log Resume in-progress: $topic-$part Replicas$replicas ISR$isr else # 正常健康状态才能开始新的 reassignment if [[ $(normalize_csv $replicas) ! $(normalize_csv $isr) ]]; then die Partition not healthy before migration: $topic-$part Replicas$replicas ISR$isr fi log Candidate $topic-$part $replicas - $target fi printf %s\t%s\t%s\n \ $topic \ $part \ $target \ $outfile if [[ $(wc -l $outfile) -ge $BATCH_SIZE ]]; then break fi done $PLAN } # # JSON # generate_json() { local input$1 local output$2 awk -F\t BEGIN { print { print \version\: 1, print \partitions\: [ first1 } { split($3,r,,) if(!first) print , printf {\topic\:\%s\,\partition\:%s,\replicas\:[%s,%s,%s],\log_dirs\:[\any\,\any\,\any\]}, $1,$2,r[1],r[2],r[3] first0 } END { print print ] print } } $input $output } # # 等待本批完成 # wait_ready() { local batch$1 local start start$(date %s) while true; do snapshot || die snapshot failed local pending0 local total0 while IFS$\t read -r topic part target do total$((total1)) local state state$(get_state $topic $part) local ct cp leader replicas isr IFS$\t read -r \ ct cp leader replicas isr $state local nt local nr local ni nt$(normalize_csv $target) nr$(normalize_csv $replicas) ni$(normalize_csv $isr) if [[ $nr $nt $ni $nt $(csv_count $isr) -eq 3 ]]; then log READY $topic-$part Replicas$replicas ISR$isr else log WAIT $topic-$part Replicas$replicas ISR$isr Target$target pending$((pending1)) fi done $batch local unavailable unavailable$(unavailable_count) local urp urp$(urp_count) log Batch status: pending$pending/$total URP$urp Unavailable$unavailable if [[ $unavailable -ne 0 ]]; then die Unavailable partitions detected: $unavailable fi if [[ $pending -eq 0 ]]; then return 0 fi if (( $(date %s) - start WAIT_TIMEOUT )); then die Timeout waiting for batch fi sleep $WAIT_INTERVAL done } # # execute one batch # execute_batch() { local num$1 local batch$2 local json${STATE_DIR}/batch_${num}.json generate_json $batch $json log log BATCH $num log cat $json if [[ $EXECUTE -eq 0 ]]; then log DRY-RUN - not executing return 2 fi # 先判断哪些已经处于目标 replicas # resume情况下避免重复execute snapshot || die snapshot failed local need_execute0 while IFS$\t read -r topic part target do state$(get_state $topic $part) IFS$\t read -r \ ct cp leader replicas isr $state if [[ $(normalize_csv $replicas) ! $(normalize_csv $target) ]]; then need_execute1 fi done $batch if [[ $need_execute -eq 1 ]]; then $REASSIGN \ --bootstrap-server $BOOTSTRAP \ --reassignment-json-file $json \ --execute \ 21 | tee -a $HISTORY rc${PIPESTATUS[0]} [[ $rc -eq 0 ]] || die Reassignment execute failed else log Batch already assigned; continuing wait fi wait_ready $batch # verify 清 throttle $REASSIGN \ --bootstrap-server $BOOTSTRAP \ --reassignment-json-file $json \ --verify \ 21 | tee -a $HISTORY || true while IFS$\t read -r topic part target do mark_completed $topic $part $target done $batch log BATCH $num SUCCESS } # # MAIN # log Kafka Healthy Migration v$VERSION log Bootstrap : $BOOTSTRAP log BatchSize : $BATCH_SIZE log StateDir : $STATE_DIR log Execute : $EXECUTE log Resume : $RESUME check_brokers U$(unavailable_count) R$(urp_count) # # 禁止在已有 reassignment 时提交新批次 # ACTIVE_REASSIGN$( $REASSIGN \ --bootstrap-server $BOOTSTRAP \ --list \ 2/dev/null ) if echo $ACTIVE_REASSIGN \ | grep -qE ^[[:space:]]*[A-Za-z0-9._-]-[0-9][[:space:]]; then log Existing Kafka partition reassignment detected: echo $ACTIVE_REASSIGN die Existing reassignment is still running. Wait until it finishes, then --resume. fi log Initial URP$R Unavailable$U [[ $U -eq 0 ]] || die Unavailable ! 0 [[ $R -eq 0 ]] || die URP ! 0 before migration if [[ $RESUME -eq 1 -s $PLAN ]]; then log Using existing plan else if [[ $RESUME -eq 0 -s $PLAN ]]; then die State exists. Use --resume or another --state-dir fi generate_plan fi batch_num1 executed0 while true; do BATCH${STATE_DIR}/current_batch.tsv build_batch $BATCH count$(wc -l $BATCH) if [[ $count -eq 0 ]]; then log No old-broker partitions remaining in plan. break fi log Next batch partitions$count column -t -s $\t $BATCH 2/dev/null || cat $BATCH execute_batch $batch_num $BATCH rc$? if [[ $rc -eq 2 ]]; then exit 0 fi executed$((executed1)) batch_num$((batch_num1)) if [[ $RUN_ALL -ne 1 $executed -ge $MAX_BATCHES ]]; then break fi done snapshot remaining$( awk { replicas for(i1;iNF;i) if($iReplicas:) replicas$(i1) nsplit(replicas,a,,) for(i1;in;i) if(a[i]1 a[i]5) { count break } } END { print count0 } $SNAPSHOT ) log log SUMMARY log log Completed checkpoints : $(wc -l $COMPLETED) log Remaining old partitions: $remaining log URP : $(urp_count) log Unavailable : $(unavailable_count) SCRIPT chmod x kafka_migrate_healthy_partitions.sh bash -n kafka_migrate_healthy_partitions.shbash -n无输出才继续第一批仍建议先 10 个先 dry-run./kafka_migrate_healthy_partitions.sh \ --batch-size 10 \ --state-dir /tmp/kafka_healthy_migrate_20260820你应该看到Migration plan partitions: 2470Next batch partitions10topicA 0 6,7,8topicB 1 7,8,9topicC 2 8,9,6...确认没有问题以后正式./kafka_migrate_healthy_partitions.sh \ --batch-size 10 \ --state-dir /tmp/kafka_healthy_migrate_20260820 \ --execute \ --resume第一批完成以后bin/kafka-topics.sh \ --bootstrap-server 10.40.0.143:9092 \ --describe \ --under-replicated-partitions | wc -lbin/kafka-topics.sh \ --bootstrap-server 10.40.0.143:9092 \ --describe \ --unavailable-partitions | wc -l最终都应恢复为URP0Unavailable0然后下一批可以提高到 20./kafka_migrate_healthy_partitions.sh \ --batch-size 20 \ --state-dir /tmp/kafka_healthy_migrate_20260820 \ --execute \ --resume建议暂时不要用--all。你现在底层 SmartPQI 故障还没有物理处理掉2470 个 Partition 一次连续复制风险过高。建议人工一批一批跑先10 → 20 → 20...。等 active reassignment0 后再重新启动脚本