如何用librdkafka获取过滤中止事务后的Kafka最新稳定偏移量？

假设Kafka日志包含如下消息：

Offset-1 = non-transactional message
Offset-2 = non-transactional message
Offset-3 = transactional message1 for transaction T1
.
.
.
offset-13 = transaction message10 for transaction T1
offset-14 = commit transaction marker
offset-15 = non-transactional message
offset-16 = transactional message1 for transaction T2
.
.
.
offset-27 = abort marker for transaction T2

使用rd_kafka_query_watermark_offsets获取最新稳定偏移量时返回offset-28，需要过滤掉中止事务（T2）后，得到offset-15作为正确的稳定偏移量。

可行的解决方法 #

• 利用消费者事务隔离级别自动过滤
  这是最简便的方案：

  1. 在librdkafka消费者配置中，将isolation.level设置为read_committed，该配置会让消费者自动忽略中止事务的所有消息，只返回已提交的事务消息和非事务消息。
  2. 配置完成后，使用rd_kafka_consumer_watermark_offsets获取消费者视角的高水位偏移量，此时返回的就是过滤掉中止事务后的有效末端偏移量（即offset-15）。
     注意：rd_kafka_query_watermark_offsets是从broker获取物理分区的高水位，不区分事务状态，所以必须用消费者视角的水位接口。

• 手动追踪事务状态计算有效偏移量
  如果需要自定义处理逻辑，可以手动解析事务控制消息：

  1. 读取分区内的消息，识别事务的提交/中止标记（这些属于Kafka内部控制消息，可通过消息属性或__transaction_state主题内容判断）。
  2. 遍历偏移量时，维护事务状态：T1提交则保留其消息，T2中止则跳过其所有关联消息。最终定位到最后一条有效消息的偏移量（offset-15）。

• 事务感知的消费者位置获取
  启动read_committed模式的消费者后，也可以通过rd_kafka_consumer_position获取消费者当前已确认的最后有效偏移量，这个值会自动排除中止事务的消息，直接得到offset-15。

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