reactor-kafka消费者无限暂停问题咨询：原因及规避方案

一、避免消费者无限暂停的方案 #

先拆解下你现有代码里的几个潜在问题，再给出具体修复动作：

1. 调整消息确认时机
   你现在在delayUntil调用前就执行了acknowledge()，这会让Kafka误以为消息已经处理完，但实际业务逻辑dispatch还没执行。如果dispatch处理慢或者阻塞，后续消息会被不断拉取却堆在内存里，背压越来越严重，最后直接停死。把确认操作移到delayUntil之后，确保业务处理完再告诉Kafka：

partitionFlux.publishOn(defaultScheduler())
    .delayUntil(this::dispatch)
    .doOnNext(kvReceiverRecord -> kvReceiverRecord.receiverOffset().acknowledge())
    .checkpoint("Dispatching the message to handler")

2. 别用无限制的重试和重复
   当前代码里的retry()和repeat()是无限循环的，要是碰到持续的处理阻塞或者无法恢复的异常，消费者会一直重试拉取，把资源耗光后彻底暂停。给重试加个次数限制，再针对无法恢复的错误做兜底处理：

.retry(3) // 限制最多重试3次
.onErrorResume(e -> {
    log.error("消息处理失败，无法恢复", e);
    return Mono.empty();
})
// 删掉repeat()，因为receiver::receive本身就是持续拉取消息的流，不需要额外重复

3. 优化背压相关配置

• 调整publishOn的线程池，确保有足够的线程处理消息，别让线程耗尽导致背压：

  Scheduler defaultScheduler = Schedulers.newBoundedElastic(10, 100, "kafka-handler"); // 根据你的业务量调整参数
  

• 给消费者设置max.poll.records，减少每次拉取的消息数，降低单批次处理的压力：

  consumerOptions.consumerProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "50");
  

4. 处理dispatch的阻塞和异常
   确保dispatch不会长时间阻塞，如果是IO密集型操作，一定要用非阻塞的Mono/Flux，或者把阻塞逻辑放到专用线程池里。同时加错误处理，别让异常静默消失：

private Mono<Void> dispatch(KafkaReceiverRecord<K, V> record) {
    return Mono.fromRunnable(() -> {
        // 你的业务处理逻辑
    })
    .subscribeOn(Schedulers.boundedElastic()) // 阻塞操作扔到boundedElastic线程池
    .onErrorMap(e -> new RuntimeException("消息分发失败", e)); // 包装异常方便排查
}

5. 去掉重复的publishOn
   你在groupBy前后都用了publishOn，这会增加线程切换的开销，还可能让背压信号传递混乱。保留groupBy之后的那个就行，或者统一规划线程池的使用。

二、Reactor-Kafka消费者暂停与恢复的触发条件 #

触发暂停的情况 #

• 背压过载：当下游处理速度跟不上消息拉取速度（比如dispatch慢、线程池满），Reactor-Kafka会自动暂停对应分区的拉取，直到下游有能力处理更多消息。
• 手动调用暂停API：通过ReceiverOffset.pause()或者KafkaReceiver.pause(TopicPartition...)手动暂停指定分区。
• 重平衡触发：重平衡开始前，消费者会自动暂停所有分区的拉取，防止重平衡过程中处理消息导致偏移量混乱。
• 未处理的异常：如果下游流出现未处理的异常，且没有合适的错误恢复策略，流可能会进入暂停状态；要是用了无限制重试但问题一直没解决，也会触发持续暂停。

触发恢复的情况 #

• 背压缓解：当下游处理完积压的消息，释放了线程资源，Reactor-Kafka会自动恢复对应分区的拉取。
• 手动调用恢复API：通过ReceiverOffset.resume()或者KafkaReceiver.resume(TopicPartition...)手动恢复指定分区。
• 重平衡完成：重平衡结束后，消费者会恢复对新分配到的分区的拉取。
• 错误恢复：当重试策略成功解决异常，流重新开始处理消息时，对应的分区拉取会恢复。

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