spring-kafka使用ConsumerAwareRebalanceListener出现分区丢失异常如何处理

根因说明 #

spring-kafka 2.3及后续版本的默认逻辑中，onPartitionsLost事件触发时，框架会先调用onPartitionsRevokedBeforeCommit回调做兼容处理，此时消费者已经丢失对应分区的持有权，直接操作这些分区提交偏移就会抛出异常。

可落地的优化方案 #

• 回调逻辑增加分区持有校验 #

  在onPartitionsRevokedBeforeCommit方法开头，先通过consumer.assignment()获取当前消费者实际持有的分区列表，和回调传入的待回收分区做交集过滤，只对当前仍持有的分区执行偏移提交逻辑，不存在的分区直接跳过即可。
  示例代码如下：

  @Override
  public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
      // 过滤出当前消费者真实持有、需要处理的分区
      Set<TopicPartition> currentAssigned = consumer.assignment();
      List<TopicPartition> toProcess = partitions.stream()
              .filter(currentAssigned::contains)
              .collect(Collectors.toList());
      if (toProcess.isEmpty()) {
          return;
      }
      // 原有提交已处理偏移的逻辑，替换为仅处理toProcess列表
  }
  

• 分区丢失事件标记跳过逻辑 #

  在监听器中新增一个boolean类型的标记位，当onPartitionsLost触发时将标记位置为true，onPartitionsRevokedBeforeCommit执行前先判断标记位，如果为true直接跳过所有提交逻辑，待重平衡完成后重置标记位即可。该方案适合业务确定分区丢失场景下当前消费者无需再处理对应偏移的场景。

• 从根源减少未提交偏移 #

  调整消费者配置降低重平衡触发概率和未提交偏移量：
  • 将容器的AckMode调整为MANUAL_IMMEDIATE，消息处理完成后立刻调用Acknowledgment.acknowledge()提交偏移，减少待提交偏移的堆积。
  • 根据业务实际消息处理耗时，适当调大max.poll.interval.ms参数，避免因业务处理超时触发不必要的重平衡。

兜底说明 #

你现有记录告警+重新消费的方案已经符合Kafka至少一次消费语义的要求，上述优化仅用于避免不必要的异常抛出和重复消费，不会破坏消费语义的一致性。 #

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