Kafka在提交偏移之前设置消费者偏移

Question

我正在运行一个程序，该程序从主题中的消息开始，使用它，处理它，提交下一个偏移量，并将新消息发布到同一主题，所有这些都以事务方式进行。我有以下（简化的）跟踪：

Fetch READ_COMMITTED at offset 20 for partition test-topic-0
processing message at offset 20
Committed offset 21 for partition test-topic-0
Sending PRODUCE
COMMITTING_TRANSACTION
Fetch READ_COMMITTED at offset 22 for partition test-topic-0
processing message at offset 22 <==== first time
...rebalance...
Setting offset for partition test-topic-0 to the committed offset FetchPosition{offset=21
Committed offset 23 for partition test-topic-0
Sending PRODUCE
COMMITTING_TRANSACTION
Fetch READ_COMMITTED at offset 24 for partition test-topic-0
stale fetch response for partition test-topic-0 since its offset 24 does not match the expected offset FetchPosition{offset=21
Fetch READ_COMMITTED at offset 21 for partition test-topic-0
processing message at offset 22 <==== second time

因此，我处理了消息“22”两次。是否预计 kafka 只是将消费者偏移量回退到在提交的偏移量之前？日志的顺序看起来正确吗？如有必要，我可以使用完整日志更新问题，但我认为那里没有任何用处。

Answer 1

看起来在生产者完成交易之前发生了重新平衡。查看您正在使用的代码/配置/Kafka 版本会很有帮助。

事务性consume-process-produce 要求生产者做几件不同的事情。处理一批记录时：

• producer.beginTransaction() - 此方法保证从调用它到事务中止/提交期间产生的所有内容都成为单个事务的一部分。

• producer.send(producerRecord) - 对于您在批处理中处理的每条消息。

• producer.sendOffsetsToTransaction( Map<TopicPartition, OffsetAndMetadata> offsetsToCommit, consumer.groupMetadata() ) - 一旦你完成了批处理，它将作为事务的一部分提交偏移量。请注意，以任何其他方式提交偏移量都不会提供事务保证。

一旦生成了批次中的所有记录并且您将偏移量作为交易的一部分提交，您最终会提交交易并完成交易 - producer.commitTransaction()

话虽如此，这应该可以解释为什么它拒绝了消息 24 并重新处理了消息 22。我相信消息 23 没有到达最后一个生产者步骤，但需要查看代码才能确定。来自 Kafka 权威指南：

为了保证消息按顺序阅读， read_committed 模式不会返回产生的消息 在第一个仍处于打开状态的事务开始之后（称为 最后稳定偏移量，或 LSO）。这些消息将被保留，直到 该事务由生产者提交或中止，或直到 到达 transaction.timeout.ms（默认 15 分钟）并被 经纪人。

和

两个主要错误（对于事务）是假设只有一次保证适用于除了生产到 Kafka 之外的操作，并且消费者总是读取整个事务并拥有有关事务边界的信息。