Kafka 消费者 Rebalance 风暴排查与调优手册

适用版本：Kafka 2.8.1（KRaft 模式）/ Spring Kafka 3.0.x / Java 11+ 作者：谢若衡-158（8 年中间件研发） 最后更新：2026-10-22

目录（TOC）

1. 什么是 Rebalance 风暴
2. 场景一：max.poll.interval.ms 超时（慢 SQL）
3. 场景二：GC 停顿 / 网络抖动
4. 场景三：Group Coordinator 切换
5. 场景四：扩缩容导致 Partition 重分配
6. 场景五：session.timeout.ms 设置不合理
7. 参数速查表
8. 参考资料

1. 什么是 Rebalance 风暴

把 Kafka Consumer Group 想象成一个会议室：每个 consumer 是一个员工，每个 partition 是一张桌子。Rebalance 就是"重新分配座位"——有人迟到（consumer 掉线）、有人提前走了（consumer 退出）、桌子数量变了（partition 增减），管理员（Group Coordinator）就要重新安排谁坐哪张桌子。

Rebalance 风暴就是：这个"重新排座位"的过程反复发生，每次排到一半就又被触发，导致所有 consumer 反复停止消费（stop-the-world），下游 SLA 被拖垮。

Kafka 2.8 使用 Cooperative Rebalance Protocol（group.protocol=cooperative），相比早期的 Eager 协议，已经大幅减少了 rebalance 的影响范围。但如果触发条件没消除，风暴依然会来。

2. 场景一：max.poll.interval.ms 超时（慢 SQL）

现象

• broker 日志出现 Member ... has failed, removing it from the group
• consumer 日志：The group coordinator ... is not available 或 Rebalance triggered
• 频率：每 5~10 分钟一次，每次停顿 30s~2min
• 业务表现：消息积压，下游 P99 飙升

诊断命令

# 1. 查看 consumer group 状态（是否频繁 Rebalancing）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-consumer-group

# 2. 查看 broker 端 rebalance 日志
grep -i "rebalance\|Member.*has failed\|Generation" \
  /var/log/kafka/server.log | tail -50

# 3. 检查 consumer 端 poll 间隔（Spring Kafka 日志）
grep "poll\|fetch\|commit" /var/log/app/kafka-consumer.log | tail -30

# 4. 确认慢 SQL：JDBC 慢查询日志
grep -E "slow|took.*[0-9]{4,}ms" /var/log/app/sql-slow.log | tail -20

根因

max.poll.interval.ms=300000（5 分钟），但业务 handler 里有 DB 写入 P99 ~8s，max.poll.records=500 一批 500 条 × 8s = 4000s，远超 300s 超时。Coordinator 判定该 consumer "失联"，触发 rebalance。

比喻：你给员工规定了"每 5 分钟必须举手说一次'我还活着'"，但他被一个 8 秒的 SQL 卡住了，500 条消息一条一条处理，根本来不及举手。管理员以为他跑了，就把他的桌子分给别人了。

调优（Before / After）

| 参数 | Before | After | 说明 | |------|--------|-------|------| | max.poll.records | 500 | 50 | 缩小单批，降低单次 poll 耗时 | | max.poll.interval.ms | 300000 | 600000 | 放宽超时，给慢 SQL 留余量 | | session.timeout.ms | 10000 | 45000 | 配合 heartbeat，避免误判 | | heartbeat.interval.ms | 3000（默认） | 15000 | 心跳间隔 < session.timeout/3 |

代码修复（Spring Kafka 3.x）

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, OrderEvent> consumerFactory(
            @Value("${kafka.bootstrap-servers}") String servers) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                  new JsonSerializer<>(OrderEvent.class));

        // 核心调优：缩小批次 + 放宽超时
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 600_000);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45_000);
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15_000);

        // 异步提交 offset，避免 commit 阻塞 poll
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
            kafkaListenerContainerFactory(
                    ConsumerFactory<String, OrderEvent> cf) {
        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(cf);
        factory.setConcurrency(6);
        factory.getContainerProperties().setAckMode(
                AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

@Component
public class OrderEventHandler {

    @KafkaListener(topics = "order-events", groupId = "order-consumer-group")
    public void handle(List<OrderEvent> events, Acknowledgment ack) {
        for (OrderEvent event : events) {
            try {
                processOrder(event);
            } catch (Exception e) {
                log.error("Failed to process order {}", event.getId(), e);
            }
        }
        ack.acknowledge();
    }
}

3. 场景二：GC 停顿 / 网络抖动

现象

• rebalance 无明显规律，偶发
• GC log 显示 Full GC 停顿 > 10s
• 网络监控看到 consumer 到 broker 的 RTT 毛刺 > 5s

诊断命令

# 1. 查看 GC 日志（确认停顿时长）
grep -A2 "Pause" /var/log/app/gc.log | grep -E "Pause|Full|G1" | tail -20

# 2. 用 jstack 抓线程栈（看 consumer 线程是否卡在 GC / 网络 IO）
jstack <consumer-pid> | grep -A10 "kafka-consumer\|KafkaConsumer"

# 3. 检查网络：consumer 到 broker 的延迟
tcpdump -i eth0 -nn "port 9092" -w /tmp/kafka.pcap
tshark -r /tmp/kafka.pcap -T fields -e kafka.api_key \
  -e frame.time_delta | awk '{if($2>1) print}' | head -20

# 4. 查看 consumer 心跳是否正常
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-consumer-group --state-only

根因

• GC 停顿：Full GC > 10s 时，consumer 线程被冻结，心跳发不出去，coordinator 判定超时
• 网络抖动：broker 所在节点网络丢包/带宽打满，fetch 请求超时

调优

| 参数 | Before | After | 说明 | |------|--------|-------|------| | JVM | -Xmx4g (CMS) | -Xmx4g (G1, -XX:MaxGCPauseMillis=200) | 换 G1 控制停顿 | | session.timeout.ms | 10000 | 45000 | 容忍 GC 停顿 | | heartbeat.interval.ms | 3000 | 15000 | 心跳线程独立于 poll 线程 |

代码修复

# 推荐 G1 + 停顿目标 200ms
-XX:+UseG1GC
-XX:MaxGCPauseMillis=200
-XX:G1HeapRegionSize=16m
-XX:InitiatingHeapOccupancyPercent=45
-Xlog:gc*:file=/var/log/app/gc.log:time,uptime,level,tags:filecount=5,filesize=50m

// Spring Kafka：不要在 handler 里做 CPU 密集计算（会阻塞 poll 线程）
@KafkaListener(topics = "order-events")
public void handle(OrderEvent event) {
    // ❌ 错误：同步做重计算
    // heavyComputation(event);

    // ✅ 正确：异步提交到独立线程池
    asyncPool.submit(() -> heavyComputation(event));
}

4. 场景三：Group Coordinator 切换

现象

• broker 重启/升级后，所有 consumer 同时触发 rebalance
• 日志：Error while fetching metadata with correlation id ... : LEADER_EPOCH_NOT_AVAILABLE
• 恢复时间 30s~2min

诊断命令

# 1. 查看 coordinator 所在 partition 和 broker
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-consumer-group

# 2. 查看 __consumer_offsets 的 leader 分布
kafka-metadata.sh --snapshot /var/lib/kafka/data/metadata-0/00000000000000000000.log \
  | grep __consumer_offsets | head -10

# 3. 查看 broker 日志中 coordinator 切换记录
grep -i "coordinator\|leader\|follower" /var/log/kafka/server.log | tail -30

# 4. 检查 KRaft 模式下 controller 切换
kafka-metadata.sh --snapshot /var/lib/kafka/data/metadata-0/00000000000000000000.log \
  --describe | grep -i controller

根因

KRaft 单 broker 模式下，coordinator 和 controller 在同一个节点。broker 重启时，所有 consumer group 的 coordinator 需要重新选举，期间 consumer 无法完成 rebalance。

比喻：会议室管理员（coordinator）请假了，新管理员（新 coordinator）还没到岗，所有员工都站在走廊里不知道坐哪张桌子。

调优

| 参数 | Before | After | 说明 | |------|--------|-------|------| | group.protocol | classic | cooperative | 减少 rebalance 影响范围 | | retry.backoff.ms | 100 | 500 | coordinator 不可用时退避重试 | | default.api.timeout.ms | 60000 | 120000 | 给 coordinator 切换留时间 |

代码修复

# application.yml
kafka:
  consumer:
    properties:
      group.protocol: cooperative
      retry.backoff.ms: 500
      default.api.timeout.ms: 120000

5. 场景四：扩缩容导致 Partition 重分配

现象

• 每次滚动发布（K8s Pod 重启）都触发全量 rebalance
• 12 partition / 6 consumer，扩到 8 consumer 或缩到 4 consumer 时必触发
• 日志：Rebalance triggered by member join/leave

诊断命令

# 1. 查看当前 group 成员和 partition 分配
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-consumer-group

# 2. 查看 K8s 滚动发布日志
kubectl logs -l app=order-consumer --tail=100 -f | grep -i "rebalance\|join\|leave"

# 3. 查看 partition 数量
kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic order-events | grep PartitionCount

根因

Consumer 数量变化（Pod 扩缩容、滚动更新）必然触发 rebalance。Eager 协议下，所有 partition 先释放再重新分配（stop-the-world）；Cooperative 协议下，只调整变化的部分。

比喻：会议室从 6 个人变 8 个人，必须重新排座位。Eager 协议 = 所有人先站起来再重新坐；Cooperative 协议 = 只有新来的 2 个人找位子，其他人不动。

调优

| 参数 | Before | After | 说明 | |------|--------|-------|------| | group.protocol | classic（Eager） | cooperative | 增量 rebalance | | K8s 策略 | RollingUpdate (maxUnavailable=1) | maxUnavailable=0, maxSurge=1 | 先起新 Pod 再杀旧 Pod |

代码修复

props.put("group.protocol", "cooperative");

# deployment.yaml
strategy:
  type: RollingUpdate
  rollingUpdate:
    maxUnavailable: 0
    maxSurge: 1

6. 场景五：session.timeout.ms 设置不合理

现象

• consumer 在正常 poll 间隙（GC、慢查询）被 coordinator 踢出
• 日志：Session timeout detected for member ...
• 频率与 GC 停顿或慢查询周期高度相关

诊断命令

# 1. 查看 consumer 心跳与 session 超时
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-consumer-group

# 2. 查看 consumer 端心跳日志
grep "heartbeat\|session\|timeout" /var/log/app/kafka-consumer.log | tail -30

# 3. 对比 GC 停顿与 session.timeout
grep "Pause" /var/log/app/gc.log | awk '{print $NF}' | sort -n | tail -5

根因

session.timeout.ms=10000，而 GC Full 停顿可达 15s。心跳线程虽然独立于 poll 线程，但 JVM 级 Stop-The-World 会冻结所有线程（包括心跳），导致 coordinator 判定 consumer 失联。

比喻：你规定员工每 10 秒要举一次手，但公司突然停电 15 秒（Full GC），所有人都动不了。管理员以为所有人跑了。

调优

| 参数 | Before | After | 说明 | |------|--------|-------|------| | session.timeout.ms | 10000 | 45000 | 容忍 GC 停顿 | | heartbeat.interval.ms | 3000 | 15000 | 心跳间隔 < session.timeout/3 | | max.poll.interval.ms | 300000 | 600000 | 与 session.timeout 解耦 |

代码修复

// 三个参数的关系（Kafka 官方推荐）：
// heartbeat.interval.ms < session.timeout.ms < max.poll.interval.ms
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45_000);
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15_000);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 600_000);

7. 参数速查表

| # | 参数 | 默认值 | 推荐值（本场景） | 调优风险 | |---|------|--------|-----------------|---------| | 1 | session.timeout.ms | 45000 | 45000 | 太小→误判失联；太大→故障检测慢 | | 2 | heartbeat.interval.ms | 3000 | 15000 | 太小→网络压力大；太大→心跳粒度粗 | | 3 | max.poll.interval.ms | 300000 | 600000 | 太小→慢 handler 超时；太大→故障恢复慢 | | 4 | max.poll.records | 500 | 50 | 太小→吞吐下降；太大→单次 poll 耗时长 | | 5 | fetch.min.bytes | 1 | 65536 | 太大→延迟增加；太小→请求频繁 | | 6 | fetch.max.wait.ms | 500 | 200 | 太大→延迟增加；太小→空轮询多 | | 7 | retry.backoff.ms | 100 | 500 | 太小→重试风暴；太大→恢复慢 | | 8 | default.api.timeout.ms | 60000 | 120000 | 太大→故障恢复慢；太小→coordinator 切换超时 | | 9 | group.protocol | classic | cooperative | 需 broker 2.4+；旧版本不支持 | | 10 | auto.offset.reset | latest | latest（按业务） | earliest→消费历史慢；latest→可能丢消息 |

8. 参考资料

1. Apache Kafka Consumer Configs（官方文档）
2. Apache Kafka: Consumer Rebalance（官方博客）
3. Spring Kafka Documentation - Consumer
4. Kafka Cooperative Rebalance Protocol（KIP-429）
5. Confluent: Tuning Kafka Consumers

一句话总结：Rebalance 风暴 90% 的根因是「consumer 来不及举手（poll 超时）」或「管理员换人了（coordinator 切换）」。把 max.poll.records 缩小、session.timeout.ms 放宽、group.protocol 切到 cooperative，三步就能消除 80% 的风暴。剩下的 20% 靠 GC 调优和网络保障。