设置read_committed后Kafka消费者无法读取已提交事务消息

以下是针对你的问题的具体排查步骤和解决方案：

• 检查消费者组偏移量状态
  先用Kafka自带工具查看消费者组的当前偏移量：

  kafka-consumer-groups.sh --bootstrap-server xxx:9092 --describe --group inc.59803.test2
  

  如果该组已经记录了偏移量（比如之前消费到了11551503），auto.offset.reset: 'earliest'不会生效，消费者会从已记录的偏移量开始拉取，自然读不到目标区间的消息。此时可以重置组偏移量：

  kafka-consumer-groups.sh --bootstrap-server xxx:9092 --reset-offsets --to-earliest --topic owfe.history --group inc.59803.test2 --execute
  

  重置后重新启动消费者测试。

• 验证客户端与集群版本兼容性
  Confluent 6.0.0对应的confluent-kafka Python客户端推荐版本为1.6.x（与Confluent平台版本保持一致）。版本不匹配可能导致事务协议解析异常。用以下命令查看当前客户端版本：

  pip show confluent-kafka
  

  若版本不匹配，升级或降级到1.6.x版本后重试。

• 尝试手动指定分区与起始偏移量
  自动订阅（subscribe）可能存在分区分配延迟，改为手动分配分区并指定起始偏移量：

  conf = {'bootstrap.servers': "xxx:9092",
          'group.id': "inc.59803.test2",
          'auto.offset.reset': 'earliest',
          'isolation.level': 'read_committed',
          }
  
  consumer = Consumer(conf)
  # 手动分配分区并设置起始偏移量
  tp = TopicPartition('owfe.history', 0, 11551495)
  consumer.assign([tp])
  consumer.seek(tp)
  
  print(consumer.get_watermark_offsets(tp))
  
  for i in range(200):
    msg = consumer.poll(timeout=5.0)
    if msg is not None:
      print(msg.topic(), msg.offset(), msg.key())
  consumer.close()
  

  运行修改后的代码，查看是否能读取到目标消息。

• 排查集群版本的已知bug
  Confluent 6.0.0存在部分事务处理的已知问题，例如当事务包含大量消息时，read_committed消费者可能无法正确识别COMMIT标记。可以临时将isolation.level改为read_uncommitted验证是否能读取消息：如果能读取，则大概率是集群版本的问题，建议升级到Confluent 6.1.0及以上版本（该版本修复了多个事务相关bug）。

• 确认事务COMMIT标记的位置
  用kafka-dump-log.sh详细查看日志中的事务标记：

  kafka-dump-log.sh --files /path/to/kafka/data/owfe.history-0/000000000011551495.log --print-data-log
  

  确认COMMIT标记的偏移量是否在目标消息区间之后。如果COMMIT消息的偏移量超过当前watermark上限（11551495），说明消费者暂时拉取不到COMMIT消息，无法解锁事务内的消息。这种情况可能是集群副本同步异常，需要检查分区的ISR状态和Leader副本是否正常运行。

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