Reactive Kafka消费者持续重平衡问题排查求助

核心问题分析 #

1. 自动提交与手动ACK冲突：同时开启ENABLE_AUTO_COMMIT_CONFIG=true和手动调用acknowledge()，导致Offset提交逻辑混乱，Rebalance过程中提交请求失败触发异常，进而引发循环Rebalance。
2. 消费者数量超过Partition数：70个消费者对应60个Partition，有10个消费者始终无法分配到Partition，这类空闲消费者会持续参与Group协调，一旦出现心跳波动就触发Rebalance，且Rebalance后仍有空闲消费者，形成恶性循环。
3. 调度器线程数不足：仅配置2个处理线程，若业务逻辑偶发延迟，会导致消息处理堆积，间接影响消费者的poll()调用频率，增加Rebalance风险。
4. 请求超时与会话超时配置不合理：REQUEST_TIMEOUT_MS_CONFIG(300500ms)小于SESSION_TIMEOUT_MS_CONFIG(360000ms)，可能导致心跳请求超时，被Coordinator判定为消费者死亡，触发Rebalance。
5. 异常处理逻辑不当：onErrorContinue忽略包括RebalanceInProgressException在内的所有异常，导致Rebalance过程中仍尝试提交Offset，加剧循环；retryWhen+repeat()的组合可能在Rebalance未完成时重复发起消费请求，引发持续Rebalance。
6. 重复Client ID：所有消费者使用相同的CLIENT_ID_CONFIG，可能导致Kafka服务器端识别混乱，干扰Group协调逻辑。

具体解决方案 #

1. 修复Offset提交逻辑冲突 #

关闭自动提交，完全使用手动ACK控制Offset提交：

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

reactor-kafka中，手动调用kafkaEvent.receiverOffset().acknowledge()后，会根据配置的commitBatchSize和commitInterval批量提交Offset，避免自动提交与手动ACK的冲突。

2. 匹配消费者数量与Partition数 #

• 方案一：将消费者实例数调整为≤60个，确保每个消费者都能分配到至少1个Partition，消除空闲消费者。
• 方案二：将Topic的Partition数扩容至≥70个，让每个消费者都能分配到Partition，避免Group内存在无任务的空闲节点。

3. 调整调度器线程数 #

增加处理线程数，避免业务处理线程成为瓶颈：

// 根据消费能力调整，例如设为10个线程
Scheduler schedulers = Schedulers.newBoundedElastic(10, 10, "UThread");

线程数建议参考Partition数或实际消息处理吞吐量，确保消息能及时处理，不影响消费者的poll()调度。

4. 修正超时配置 #

确保REQUEST_TIMEOUT_MS_CONFIG大于SESSION_TIMEOUT_MS_CONFIG，避免心跳请求超时被误判：

props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 400000); // 大于360000ms的会话超时

5. 优化异常处理逻辑 #

• 调整onErrorContinue，仅忽略业务相关异常，保留Rebalance异常的处理：

.onErrorContinue((throwable, o) -> {
    if (!(throwable instanceof RebalanceInProgressException)) {
        LOGGER.error("Error while consuming record: {}", o, throwable);
    }
})

• 调整retryWhen逻辑，避免在Rebalance期间重复重试：

.retryWhen(Retry.backoff(3, Duration.ofSeconds(3))
        .transientErrors(true)
        .filter(throwable -> !(throwable instanceof RebalanceInProgressException)))

• 移除repeat()，reactor-kafka的receive()本身会持续消费，无需额外重复订阅，否则可能在Rebalance后重复初始化消费逻辑。

6. 使用唯一Client ID #

为每个消费者实例生成唯一的Client ID，例如结合UUID：

props.put(ConsumerConfig.CLIENT_ID_CONFIG, "client-" + UUID.randomUUID());

避免多个消费者使用相同Client ID导致Kafka服务器端的连接识别问题。

7. 验证Rebalance配置 #

确保所有消费者实例使用相同的PARTITION_ASSIGNMENT_STRATEGY_CONFIG（即CooperativeStickyAssignor），避免因分配策略不一致引发强制Rebalance。

额外排查点 #

• 检查Kafka集群状态：确认Group Coordinator节点稳定，无网络波动或节点故障。
• 监控消费者指标：观察consumer_rebalance_total、heartbeat_send_rate等指标，定位Rebalance触发的具体时机和频率。

内容的提问来源于stack exchange，提问作者Vijay Manohar