Kafka Streams 在生成主题时不会将偏移量增加 1
Kafka Streams does not increment offset by 1 when producing to topic
我已经实现了一个简单的 Kafka 死信记录处理器。
它在使用控制台制作者生成的记录时完美运行。
但是我发现我们的 Kafka Streams 应用程序不保证向接收器主题生成记录时，偏移量将针对每条生成的记录递增 1。
死信处理器背景：
我有一个场景，在发布处理记录所需的所有数据之前，可能会收到记录。
当记录与流应用程序的处理不匹配时，它们将移至死信主题，而不是继续向下流。发布新数据时，我们会将死信主题中的最新消息转储回流应用程序的源主题，以便使用新数据进行重新处理。
死信处理器：
- 在运行应用程序开始时记录每个分区的结束偏移量
- 结束偏移量标记停止处理给定死信主题的记录的点，以避免如果重新处理记录 return 到死信主题的无限循环。
- 应用程序从上一个 运行 通过消费者组产生的最后一个 Offsets 恢复。
- 应用程序正在使用事务和
KafkaProducer#sendOffsetsToTransaction 提交最后生成的偏移量。
为了跟踪我的范围内的所有记录何时针对主题的分区进行处理，我的服务将其最后生成的生产者偏移量与消费者保存的结束偏移量映射进行比较。当我们达到结束偏移量时，消费者通过 KafkaConsumer#pause 暂停该分区，当所有分区都暂停时（意味着它们达到保存的结束偏移量）然后调用它退出。
Offsets and Consumer Position
Kafka maintains a numerical offset for each record in a partition. This offset acts as a unique identifier of a record within that partition, and also denotes the position of the consumer in the partition. For example, a consumer which is at position 5 has consumed records with offsets 0 through 4 and will next receive the record with offset 5.
Kafka Producer API 引用下一个偏移量也始终为 +1。
Sends a list of specified offsets to the consumer group coordinator, and also marks those offsets as part of the current transaction. These offsets will be considered committed only if the transaction is committed successfully. The committed offset should be the next message your application will consume, i.e. lastProcessedMessageOffset + 1.
但是你可以在我的调试器中清楚地看到，单个分区消耗的记录不会一次递增 1...
我认为这可能是 Kafka 配置问题，例如 max.message.bytes 但 none 确实有道理。
然后我想也许是因为加入，但没有看到任何方式可以改变制作人的运作方式。
不确定它是否相关，但我们所有的 Kafka 应用程序都在使用 Avro 和 Schema Registry...
无论生产方法如何，偏移量都应该始终增加 1，还是使用 Kafka 流 API 不能提供与普通生产者消费者客户端相同的保证？
我是否完全缺少某些东西？
消息偏移量增加 1 并不是官方的 API 合同，即使 JavaDocs 指出了这一点（似乎应该更新 JavaDocs）。
如果你不使用事务，你要么得到至少一次语义，要么得不到保证（有些人称之为最多一次语义）。对于至少一次，记录可能被写入两次，因此，两个连续消息的偏移量并没有真正增加一个，因为重复写入 "consumes" 两个偏移量。
如果您使用事务，事务的每次提交（或中止）都会将一个提交（或中止）标记写入主题——这些事务标记也 "consume" 一个偏移量（这就是你观察到的）。
因此，一般来说，您不应该依赖连续的偏移量。您得到的唯一保证是，每个偏移量在分区内都是唯一的。
我知道了解消息的偏移量很有用。但是，Kafka 只会保证消息-X 的偏移量大于最后一条消息（X-1）的偏移量。顺便说一句，理想的解决方案不应基于偏移量计算。
在后台，kafka 生产者可能会尝试重新发送消息。此外，如果经纪人倒闭，则可能会发生重新平衡。 Exactly-once-semantics 可能会附加一条额外的消息。因此，如果发生上述任何事件，您的消息的偏移量可能会发生变化。
Kafka 可能会向主题添加其他消息以供内部使用。但是 Kafka 的消费者 API 可能会丢弃这些内部消息。因此，您只能看到您的消息，而消息的偏移量不一定会增加 1。
我已经实现了一个简单的 Kafka 死信记录处理器。
它在使用控制台制作者生成的记录时完美运行。
但是我发现我们的 Kafka Streams 应用程序不保证向接收器主题生成记录时，偏移量将针对每条生成的记录递增 1。
死信处理器背景：
我有一个场景，在发布处理记录所需的所有数据之前，可能会收到记录。 当记录与流应用程序的处理不匹配时，它们将移至死信主题，而不是继续向下流。发布新数据时，我们会将死信主题中的最新消息转储回流应用程序的源主题，以便使用新数据进行重新处理。
死信处理器：
- 在运行应用程序开始时记录每个分区的结束偏移量
- 结束偏移量标记停止处理给定死信主题的记录的点，以避免如果重新处理记录 return 到死信主题的无限循环。
- 应用程序从上一个 运行 通过消费者组产生的最后一个 Offsets 恢复。
- 应用程序正在使用事务和
KafkaProducer#sendOffsetsToTransaction提交最后生成的偏移量。
为了跟踪我的范围内的所有记录何时针对主题的分区进行处理，我的服务将其最后生成的生产者偏移量与消费者保存的结束偏移量映射进行比较。当我们达到结束偏移量时，消费者通过 KafkaConsumer#pause 暂停该分区，当所有分区都暂停时（意味着它们达到保存的结束偏移量）然后调用它退出。
Offsets and Consumer Position Kafka maintains a numerical offset for each record in a partition. This offset acts as a unique identifier of a record within that partition, and also denotes the position of the consumer in the partition. For example, a consumer which is at position 5 has consumed records with offsets 0 through 4 and will next receive the record with offset 5.
Kafka Producer API 引用下一个偏移量也始终为 +1。
Sends a list of specified offsets to the consumer group coordinator, and also marks those offsets as part of the current transaction. These offsets will be considered committed only if the transaction is committed successfully. The committed offset should be the next message your application will consume, i.e. lastProcessedMessageOffset + 1.
但是你可以在我的调试器中清楚地看到，单个分区消耗的记录不会一次递增 1...
我认为这可能是 Kafka 配置问题，例如 max.message.bytes 但 none 确实有道理。
然后我想也许是因为加入，但没有看到任何方式可以改变制作人的运作方式。
不确定它是否相关，但我们所有的 Kafka 应用程序都在使用 Avro 和 Schema Registry...
无论生产方法如何，偏移量都应该始终增加 1，还是使用 Kafka 流 API 不能提供与普通生产者消费者客户端相同的保证？
我是否完全缺少某些东西？
消息偏移量增加 1 并不是官方的 API 合同，即使 JavaDocs 指出了这一点（似乎应该更新 JavaDocs）。
如果你不使用事务，你要么得到至少一次语义，要么得不到保证（有些人称之为最多一次语义）。对于至少一次，记录可能被写入两次，因此，两个连续消息的偏移量并没有真正增加一个，因为重复写入 "consumes" 两个偏移量。
如果您使用事务，事务的每次提交（或中止）都会将一个提交（或中止）标记写入主题——这些事务标记也 "consume" 一个偏移量（这就是你观察到的）。
因此，一般来说，您不应该依赖连续的偏移量。您得到的唯一保证是，每个偏移量在分区内都是唯一的。
我知道了解消息的偏移量很有用。但是，Kafka 只会保证消息-X 的偏移量大于最后一条消息（X-1）的偏移量。顺便说一句，理想的解决方案不应基于偏移量计算。
在后台，kafka 生产者可能会尝试重新发送消息。此外，如果经纪人倒闭，则可能会发生重新平衡。 Exactly-once-semantics 可能会附加一条额外的消息。因此，如果发生上述任何事件，您的消息的偏移量可能会发生变化。
Kafka 可能会向主题添加其他消息以供内部使用。但是 Kafka 的消费者 API 可能会丢弃这些内部消息。因此，您只能看到您的消息，而消息的偏移量不一定会增加 1。