Micronaut中KafkaListener运行一段时间后停止工作的问题求助

针对你遇到的这个棘手问题，结合你提供的配置和现象，我整理了几个核心排查方向和解决方案，你可以逐一尝试：

1. 消费者组ID自动生成引发的偏移量异常 #

你提到修改Listener类名就能恢复消费，这大概率和Micronaut Kafka默认的消费者组ID生成逻辑有关——它会自动用Listener类名作为组ID。如果同一个组ID的消费者出现异常退出、偏移量同步紊乱，Kafka Broker会判定该消费者失效，不再分配分区给它。

解决建议：

• 手动指定固定的消费者组ID，摆脱对类名的依赖：

  @KafkaListener(groupId = "specific-task-medical-records-group")
  public class SpecificTaskMedicalRecordsListener {
      // ... 现有代码
  }
  

• 用Kafka命令行工具检查消费者组的偏移量状态，确认是否存在异常：

  kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group specific-task-medical-records-group
  

  如果发现偏移量滞后（LAG）或状态异常，可以尝试重置偏移量（注意：重置会重新消费消息，需结合业务场景评估）：

  kafka-consumer-groups --bootstrap-server localhost:9092 --reset-offsets --to-earliest --topic core-specific-task-save --group specific-task-medical-records-group --execute
  

2. 消费线程因未捕获异常阻塞/终止 #

你的Listener方法里调用了specificTaskService.save(request)，如果这个方法执行超时、阻塞，或者抛出未捕获的异常，会直接导致Micronaut的Kafka消费线程挂起，进而停止消费。

解决建议：

• 在Listener方法中增加全局异常捕获，确保线程不会因为异常终止：

  private static final Logger log = LoggerFactory.getLogger(SpecificTaskMedicalRecordsListener.class);
  
  @Topic("core-specific-task-save")
  public void save(KafkaRequest<SpecificTaskResponse> kafkaRequest) {
      try {
          SpecificTaskRequest request = objectMapper.convertValue(kafkaRequest.getResponse(), SpecificTaskRequest.class);
          manualTenantResolver.setManualTenant(LogUtil.getTenancyFromToken(kafkaRequest.getToken()));
          specificTaskService.save(request);
      } catch (Exception e) {
          // 记录详细日志，必要时可将失败消息转发到死信队列
          log.error("消费消息失败，请求内容: {}", kafkaRequest, e);
      }
  }
  

• 排查specificTaskService.save(request)是否存在慢操作（比如数据库慢查询、远程调用超时），可以通过Micrometer等监控工具跟踪方法执行时间，优化阻塞点。

3. 消费者关键配置缺失导致Broker判定失效 #

你的application.yml只配置了生产者参数，消费者的心跳间隔、会话超时等核心配置都用了默认值，这些默认值可能无法适配你的业务场景，导致Broker误以为消费者已离线。

建议补充的消费者配置：

kafka:
  bootstrap:
    servers: ${KAFKA_HOST:`localhost:9092`}
  enabled: true
  producers:
    default:
      retries: ${KAFKA_RETRIES:5}
      retry.backoff.ms: ${KAFKA_RETRY_BACKOFF:3000}
  consumers:
    default:
      session.timeout.ms: 30000       # 延长会话超时时间，避免误判离线
      heartbeat.interval.ms: 10000     # 心跳间隔设为会话超时的1/3
      auto.commit.interval.ms: 5000    # 自动提交偏移量的间隔
      auto.offset.reset: earliest      # 偏移量不存在时从最开始消费
      max.poll.records: 100            # 单次拉取的最大记录数，根据业务调整
      fetch.max.wait.ms: 500           # 拉取超时时间

4. Docker环境的资源或网络问题 #

Kafka运行在Docker容器中，可能存在资源不足或网络不稳定的情况，导致消费者与Broker的连接中断且无法自动重连。

排查建议：

• 查看Kafka容器日志，检查是否有连接异常、分区分配失败的记录：

  docker logs <kafka-container-id>
  

• 调整Docker容器的CPU、内存配额，避免因资源不足导致Broker无法正常处理请求。
• 验证微服务与Kafka容器之间的网络连通性，排除间歇性断网的可能。

5. 版本兼容性问题 #

你使用的micronaut-kafka:4.5.0是否和你的Micronaut框架版本、Kafka版本兼容？版本不匹配可能引发隐性的线程或连接问题。

解决建议：

• 参考Micronaut官方文档的版本矩阵，确认依赖版本的兼容性。
• 尝试升级到最新的Micronaut Kafka稳定版本，修复已知的Bug。

你可以先从手动指定消费者组ID和添加异常捕获这两个方向入手，这是最容易验证且大概率解决问题的方案。

备注：内容来源于stack exchange，提问作者Aism793