Quarkus Kafka消费者长时间重试后停止消费新消息求助

针对你遇到的「长时间重试后消费者停止消费新消息」的问题，结合Quarkus SmallRye Kafka的运行机制，以下是具体的排查方向和解决方案：

1. 调整Kafka消费者核心超时参数（最可能解决问题） #

你的重试时长超过2小时，而Kafka消费者默认的max.poll.interval.ms（两次拉取请求的最大间隔）通常仅为5分钟。当消费者长时间卡在重试逻辑中，未发起新的poll请求时，Kafka集群会判定该消费者失效，将其踢出消费组并重新分配分区。但本地消费者实例处理完当前消息后，已失去对应分区的消费权限，导致无法拉取新消息，最终引发堆积。

修改配置： #

在你的YAML配置中新增以下参数：

mp:
  messaging:
    incoming:
      dedicated-channel-19:
        # 保留原有配置，新增以下参数
        max.poll.interval.ms: 21600000  # 设置为6小时，大于你的最长重试时长
        session.timeout.ms: 1800000     # 30分钟，需大于3倍心跳间隔
        heartbeat.interval.ms: 60000    # 1分钟，确保心跳持续发送

2. 避免工作线程池阻塞 #

@Blocking注解默认使用Quarkus全局工作线程池，长时间的无限重试会占用所有可用线程，导致没有线程处理新的消息拉取任务。

解决方案： #

• 为该消费者配置独立线程池：

quarkus.thread-pool.dedicated-channel-19.core-threads: 5
quarkus.thread-pool.dedicated-channel-19.max-threads: 10

• 在方法上指定使用该线程池：

@Blocking("dedicated-channel-19")

• 优化重试策略，添加指数退避避免高频重试：

@Retry(
    delay = 10, delayUnit = ChronoUnit.SECONDS,
    maxRetries = -1, maxDuration = 0,
    retryOn = KafkaRetryableException.class,
    backoff = @Backoff(maxDelay = 60, delayUnit = ChronoUnit.SECONDS) // 最大重试间隔60秒
)

3. 检查手动ACK与重试的交互逻辑 #

使用MANUAL ACK策略时，若consumeEntry方法中的ACK逻辑存在异常，可能导致消费者内部的消息状态跟踪混乱，处理完当前消息后无法触发下一次拉取。

验证步骤： #

• 暂时将ACK策略改为POST_PROCESSING，让框架自动处理ACK：

@Acknowledgment(Acknowledgment.Strategy.POST_PROCESSING)

如果问题不再复现，说明你的手动ACK逻辑存在问题，需检查：

• 是否仅在处理成功时调用message.ack()
• 抛出KafkaRetryableException时是否未调用message.nack()（@Retry会自动处理重试，无需手动nack）

4. 启用详细日志定位根因 #

添加Kafka消费者的DEBUG日志，跟踪心跳、分区分配、偏移量提交等关键行为：

quarkus.log.category."io.smallrye.reactive.messaging.kafka".level=DEBUG
quarkus.log.category."org.apache.kafka.clients.consumer".level=DEBUG

重点关注以下日志：

• 消费者心跳发送记录
• 分区分配/撤销事件
• 偏移量提交结果

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