bbejeck opened a new pull request, #23585: URL: https://github.com/apache/kafka/pull/23585
When a task both consumes a source-topic changelog partition (a KTable that reuses its source topic as its changelog) and produces to that same partition, `StreamTask.checkpointableOffsets()` gave the produced offset precedence over the consumed offset. On a clean suspend, close, or recycle, the store's changelog offset was set to the last offset the task *wrote*, which can be ahead of the last offset actually *applied* to the store. A later restore or standby update starting from that checkpoint silently skips the records in between, leaving those keys stale or missing — with no error or warning. Fix: let `consumedOffsets` win over `recordCollector.offsets()` in `checkpointableOffsets()`. For a source-topic changelog the store is filled by consuming the partition, so the consumed offset is what was applied to the store. This only affects a partition that is both consumed and produced to *and* is a store changelog — i.e. a source-topic changelog. Dedicated changelog, output, and repartition partitions are never in `consumedOffsets` (only input partitions are), so their produced offsets are unchanged. Testing: - `shouldCheckpointConsumedOffsetNotProducedOffsetForSourceChangelogPartition` — new; fails without the fix. - `shouldCheckpointProducedOffsetForDedicatedChangelogPartition` — new; guards that dedicated changelogs keep their produced offset. - `shouldCommitWhileUpdateSnapshotWithTheConsumedOffsetsForSuspendedRunningTask` — updated; it encoded the old produced-wins behavior. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
