帮我写一份 Kafka 分区扩容（12→48）下消费者组 Rebalance 的压测方案与可复现脚本

Kafka 分区扩容（12→48）消费者组 Rebalance 压测方案

一、压测目标与核心风险

分区从 12 扩到 48（×4），核心风险点：

1. Rebalance 风暴：分区数变化触发 consumer group 全量 rebalance，CooperativeStickyAssignor vs. StaticGroupMember 行为差异显著；
2. 消息重复/丢失：rebalance 期间 in-flight 消息的 offset commit 竞态；
3. 消费者吞吐塌陷：rebalance 期间消费者暂停消费，下游积压；
4. 分区重映射：key 的分区归属变化（hash(key) % 12 → hash(key) % 48），同一 key 可能落到不同分区，影响顺序性。

二、环境基线

Kafka: 3.6.1 (KRaft mode, 3 brokers)
JDK:   17.0.9
消费者: 12 实例 (每实例 1 分区，1:1 映射)
分区: 12 → 48
Topic: test-rebalance (replication=3)
Producer 速率: 5000 msg/s, msg size=2KB

三、压测方案（分 3 轮）

Round 1：扩容前基线（12 分区）

# 1. 创建 topic
kafka-topics.sh --create \
  --bootstrap-server broker1:9092,broker2:9092,broker3:9092 \
  --topic test-rebalance --partitions 12 --replication-factor 3

# 2. 启动 12 个消费者 (KafkaConsumer, group-id=stress-g1)
#    配置: max.poll.records=500, session.timeout.ms=45000,
#          heartbeat.interval.ms=15000, max.poll.interval.ms=300000

# 3. 持续生产 5000 msg/s, 30 分钟
kafka-producer-perf-test.sh --topic test-rebalance \
  --num-records 90000000 --record-size 2048 \
  --throughput 5000 --producer-props \
  bootstrap.servers=broker1:9092 \
  acks=all --compression.type=none

# 4. 采集指标 (每 5s):
#    - consumer lag (per partition)
#    - rebalance 事件 (Kafka 日志 grep "Rebalance")
#    - consumer 端 P99 消费延迟
#    - broker 端 request handler 队列深度

基线记录：

• 稳态 consumer lag: < 5000 (0.5s 窗口)
• 无 rebalance 事件
• P99 消费延迟: 2~5ms

Round 2：在线扩容（12→48，消费者不停）

# 1. 触发扩容
kafka-topics.sh --alter \
  --bootstrap-server broker1:9092,broker2:9092,broker3:9092 \
  --topic test-rebalance --partitions 48

# 2. 观察消费者组 (每 2s 采集):
kafka-consumer-groups.sh --describe \
  --group stress-g1 --bootstrap-server broker1:9092

# 3. 关键观察窗口 (扩容后 0~120s):
#    - rebalance 触发时间与持续时间
#    - consumer lag 峰值与恢复时间
#    - 是否有消息重复 (用 key+seq 去重校验)
#    - 消费者实例日志: "Revoking partitions [p0..p11]" → "Adding partitions [p0..p47]"

# 4. 校验顺序性 (对有序 key):
#    扩容前: key "order-0001" → partition hash("order-0001")%12 = 3
#    扩容后: key "order-0001" → partition hash("order-0001")%48 = 15
#    → 同一 key 落不同分区! 需要业务侧确认是否可接受

验收指标： | 指标 | 阈值 | 说明 | |------|------|------| | Rebalance 持续时间 | ≤ 30s | 12→48 分区，CooperativeSticky 应 < 10s | | Consumer lag 峰值 | ≤ 50000 | 即 5s 的积压量 | | Lag 恢复时间 (回到 <5000) | ≤ 60s | | | 消息丢失 | 0 | 必须 | | 消息重复 | ≤ 3% | rebalance 窗口内的可接受重复 | | 顺序性破坏 | 按 key 维度报告 | 需业务确认 |

Round 3：扩容后稳态验证（48 分区）

# 1. 消费者扩到 48 实例 (或保持 12 实例, 每实例 4 分区)
# 2. 持续生产 5000 msg/s, 30 分钟
# 3. 对比 Round 1 基线:
#    - 稳态 lag 应持平或更优 (分区并行度提升)
#    - P99 消费延迟应持平或更优
#    - broker 端请求分布是否均匀 (各 broker 的 request rate)

# 4. 压力叠加: 扩容稳态后, kill 1 个消费者实例
#    验证 rebalance 恢复:
kill -9 <consumer_pid>
#    期望: 剩余消费者 30s 内接管, lag 峰值 ≤ 10000

四、可复现脚本

#!/usr/bin/env bash
# kafka-rebalance-stress.sh
set -euo pipefail

BROKERS="broker1:9092,broker2:9092,broker3:9092"
TOPIC="test-rebalance"
GROUP="stress-g1"
PRODUCER_RATE=5000
DURATION_SEC=1800
RECORD_SIZE=2048

log() { echo "[$(date '+%H:%M:%S')] $*"; }

# === Round 1: 基线 ===
log "Round 1: 创建 12 分区 topic"
kafka-topics.sh --create --bootstrap-server "$BROKERS" \
  --topic "$TOPIC" --partitions 12 --replication-factor 3 2>/dev/null || true

log "Round 1: 生产 ${PRODUCER_RATE} msg/s × ${DURATION_SEC}s"
kafka-producer-perf-test.sh --topic "$TOPIC" \
  --num-records $((PRODUCER_RATE * DURATION_SEC)) \
  --record-size "$RECORD_SIZE" \
  --throughput "$PRODUCER_RATE" \
  --producer-props bootstrap.servers="$BROKERS" acks=all &
PRODUCER_PID=$!

# 采集 lag (每 5s)
for i in $(seq 1 $((DURATION_SEC / 5))); do
  sleep 5
  kafka-consumer-groups.sh --describe --group "$GROUP" \
    --bootstrap-server "$BROKERS" 2>/dev/null | grep -v "^GROUP" \
    | awk '{print $1, $5, $6}' | tee -a lag_round1.csv
done
wait $PRODUCER_PID 2>/dev/null || true
log "Round 1 完成"

# === Round 2: 在线扩容 ===
log "Round 2: 触发分区扩容 12 → 48"
EXPAND_TS=$(date +%s)
kafka-topics.sh --alter --bootstrap-server "$BROKERS" \
  --topic "$TOPIC" --partitions 48
log "扩容命令已发出, 观察 120s..."

for i in $(seq 1 24); do  # 24 × 5s = 120s
  sleep 5
  ELAPSED=$(( $(date +%s) - EXPAND_TS ))
  LAG=$(kafka-consumer-groups.sh --describe --group "$GROUP" \
    --bootstrap-server "$BROKERS" 2>/dev/null | awk -F'\\t' 'NR>1{sum+=$5} END{print sum+0}')
  echo "$ELAPSED,${LAG}" | tee -a lag_round2.csv
done

# 校验 rebalance 日志
log "Rebalance 事件:"
grep -h "Rebalance\|Revoking\|Adding partitions" consumer-*.log | tail -20

# === Round 3: 稳态 ===
log "Round 3: 48 分区稳态验证 ${DURATION_SEC}s"
kafka-producer-perf-test.sh --topic "$TOPIC" \
  --num-records $((PRODUCER_RATE * DURATION_SEC)) \
  --record-size "$RECORD_SIZE" \
  --throughput "$PRODUCER_RATE" \
  --producer-props bootstrap.servers="$BROKERS" acks=all &
PRODUCER_PID=$!
for i in $(seq 1 $((DURATION_SEC / 5))); do
  sleep 5
  LAG=$(kafka-consumer-groups.sh --describe --group "$GROUP" \
    --bootstrap-server "$BROKERS" 2>/dev/null | awk -F'\\t' 'NR>1{sum+=$5} END{print sum+0}')
  echo "$((i*5)),${LAG}" | tee -a lag_round3.csv
done
wait $PRODUCER_PID 2>/dev/null || true
log "Round 3 完成"

# === 汇总 ===
log "=== 汇总 ==="
log "Round2 lag 峰值: $(awk -F, '{if($2>m)m=$2}END{print m}' lag_round2.csv)"
log "Round2 lag 恢复: $(awk -F, '$2<5000{print $1; exit}' lag_round2.csv)s"
log "Round3 稳态 lag (10min 后): $(awk -F, 'NR>120{sum+=$2;n++}END{print sum/n}' lag_round3.csv)"

五、关键配置建议

# 消费者侧 (推荐)
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
session.timeout.ms=45000
heartbeat.interval.ms=15000
max.poll.interval.ms=300000
max.poll.records=500
enable.auto.commit=false
# 手动 commit: 每批消费后同步 commitSync()

六、风险清单与缓解

| 风险 | 影响 | 缓解 | |------|------|------| | 扩容瞬间 rebalance 全组暂停 | 消费中断 ≤30s | 用 CooperativeSticky 减少重分配范围 | | 分区重映射破坏顺序 | 同一 key 落不同分区 | 业务侧用 partition key 绑定逻辑 或 接受乱序 | | 消费者数 < 分区数 | 部分实例负载 ×4 | 扩容后同步扩消费者实例 | | Broker 元数据同步延迟 | 扩容后 5~10s 内 producer 可能写旧分区数 | 扩容后等待 kafka-topics.sh --describe 确认再放量 |

七、验收 Checklist

• [ ] Round 1 基线数据已记录 (lag, P99, rebalance 次数=0)
• [ ] Round 2 rebalance 持续时间 ≤ 30s
• [ ] Round 2 consumer lag 峰值 ≤ 50000 且 60s 内恢复
• [ ] 消息丢失 = 0 (producer acks=all + consumer offset 对账)
• [ ] Round 3 稳态 lag 与 Round 1 持平或更优
• [ ] Kill 1 实例后 30s 内恢复
• [ ] 顺序性报告已出具 (按 key 维度)