相同group-id的Kafka消费者始终卡在读取同一分区问题求助

问题根因 #

• 同组消费者重复分配坏分区的原因：你的第一个消费者遇到反序列化异常后直接崩溃退出消费者组，Kafka会触发消费者组重平衡，将故障消费者负责的坏分区重新分配给组内其他存活的消费者，因此第二个消费者也会读取到该分区的坏消息。
• 循环卡顿的核心原因：你开启了自动offset提交，自动提交的触发前提是消息被成功拉取并完成处理，但反序列化异常是在消息拉取后、业务逻辑处理前的序列化阶段抛出的，此时还未达到自动提交的触发条件，offset始终无法提交。配合RestartSource.onFailuresWithBackoff的重启逻辑，消费者每次重启都会从上次未提交的offset位置重新拉取，始终卡在坏消息位置无限循环。

排查步骤 #

• 确认当前消费者组的offset提交情况：执行kafka-consumer-groups.sh --bootstrap-server <kafka地址> --describe --group <你的group-id>，查看坏分区对应的CURRENT-OFFSET值，确认offset是否卡在固定位置不动。
• 验证坏消息存在：执行kafka-console-consumer.sh --bootstrap-server <kafka地址> --topic <topic名> --partition <坏分区号> --offset <卡住的offset值> --max-messages 1，拉取对应位置的消息，确认是否为格式异常的Avro消息。

解决方案 #

1. 实现容错反序列化器
   不要直接使用原生KafkaAvroDeserializer，封装一层异常捕获逻辑，避免反序列化失败直接抛出异常终止流：

class FaultTolerantAvroDeserializer[T <: SpecificRecordBase] extends Deserializer[T] {
  private val inner = new KafkaAvroDeserializer()

  override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = 
    inner.configure(configs, isKey)

  override def deserialize(topic: String, data: Array[Byte]): T = {
    try {
      inner.deserialize(topic, data).asInstanceOf[T]
    } catch {
      case e: Exception =>
        // 此处可记录错误日志，上报坏消息的topic、分区、offset信息
        null.asInstanceOf[T]
    }
  }

  override def close(): Unit = inner.close()
}

将你的消费者配置中的反序列化器替换为上述自定义实现即可。

2. 改为手动提交offset
   关闭自动提交配置，使用Alpakka Kafka提供的手动提交能力，仅在消息处理完成（或判定为坏消息需要丢弃）时提交offset，避免offset卡滞：

// 修改消费者配置，关闭自动提交
ConsumerSettings(actorSystem, new StringDeserializer, deserializer)
  .withBootstrapServers(bootstrapServer)
  .withGroupId(groupId)
  .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
  .withProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false") // 关闭自动提交

业务逻辑中对反序列化返回null的坏消息直接跳过，正常处理完成后再通过Committer.sink提交offset。

3. 调整重启策略
   修改RestartSource的异常判定逻辑，仅对网络波动、Kafka集群不可用等可恢复的系统异常触发重启，单条消息的反序列化异常属于不可恢复的业务错误，不需要触发流重启。

4. 紧急恢复方法
   如果当前业务已经卡滞，可以先手动跳过坏消息：

kafka-consumer-groups.sh --bootstrap-server <kafka地址> --group <你的group-id> --topic <topic名>:<分区号> --reset-offsets --to-offset <坏消息offset+1> --execute

执行后重启消费者即可恢复正常消费。

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