Spring-Kafka应用停止前offset提交及历史offset提交问题咨询

你之前的方案失效是因为Spring Kafka的容器销毁优先级高于普通自定义Bean的@PreDestroy方法，日志中KafkaMessageListenerContainer$ListenerConsumer - myManualConsumer: Consumer stopped记录证明消费者实例已经被关闭，此时调用acknowledgment.acknowledge()时，底层Kafka消费者客户端已经停止运行，提交请求无法发送到Kafka集群，所以offset不会实际更新。

可行方案如下：

• 方案1：监听容器停止前置事件
  通过Spring的事件监听机制监听ListenerContainerStoppingEvent事件，该事件会在消费者实例正式停止前触发，此时消费者仍处于存活状态，可以正常执行offset提交、文件刷写等收尾逻辑。示例代码：

  @EventListener
  public void onContainerStopping(ListenerContainerStoppingEvent event) {
      // 只匹配自己的消费者容器，避免其他容器事件触发逻辑
      if ("myManualConsumer".equals(event.getSource().getListenerId())) {
          writeCsv(true);
          acknowledgment.acknowledge();
          LOGGER.info("收尾逻辑执行完成，offset已提交");
      }
  }
  

• 方案2：配置优雅停机超时
  在应用配置中添加优雅停机超时配置，给收尾逻辑留足执行时间，避免容器直接强制终止：

  # 可根据实际收尾逻辑耗时调整，单位支持秒、分钟等
  spring.kafka.listener.graceful-shutdown-timeout=30s
  

• 方案3：自定义容器停止回调
  初始化KafkaMessageListenerContainer时，自定义stopCallback回调，将收尾逻辑绑定到容器停止流程的最前端。

Kafka本身允许提交任意数值的offset，包括比当前已提交offset更早的历史offset，提交操作本身不会抛出语法类异常，但会引发严重的业务逻辑问题：

1. 你描述的场景中，提交offset=101的操作成功后，Kafka服务端记录的该分区已提交offset已经更新为101，代表所有offset≤101的消息都已经处理完成，下次消费会从offset=102开始读取。
2. 后续你再执行offset=100的提交操作，会将该分区的已提交offset回退到100，服务重启后会从offset=101开始重新消费，导致offset=101的消息被重复处理。
3. 隐藏风险更高：如果提交offset=101之后、提交offset=100之前服务宕机，你攒在内存里的offset=100的GOOD事件还没写入CSV，但是Kafka的offset已经更新到101，重启后不会再消费offset=100的消息，直接造成数据丢失。

优化建议 #

不要在过滤逻辑里单独提交offset，可以维护一个单分区下的「当前已处理完成的最大offset」变量，不管是GOOD还是BAD事件，处理完成后都更新这个变量，触发提交条件时直接提交这个最大offset即可，既不会漏提交也不会出现offset回退的问题。

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