bbejeck opened a new pull request, #24002: URL: https://github.com/apache/kafka/pull/24002
Under exactly-once, a corrupted task (for example after a commit times out) is closed dirty, which drops the records it fetched but has not processed yet. It is then revived with its input partitions marked to be reset to the committed offsets. That reset only runs once the revived task is initialized and completes restoration. With the classic protocol, StreamThread enforces a rebalance right after the corruption, and any assignment that arrives before the revived task is initialized has the same effect: handleTasksPendingInitialization closes the revived task and a new task is created for the same task id. The new task starts with no pending reset, so it processes from the current consumer position, and the next commit moves the input offset past the dropped records. Those records never reach the state store or the changelog. This change keeps the input partitions that a task closed while pending initialization still had to reset, and hands them to the new active task created in the same assignment, limited to that task's input partitions that are still assigned to the consumer. ReadOnlyTask throws on the new Task#partitionsForOffsetReset, like the other mutating and non-read-only methods. CommitTimeoutOffsetResetIntegrationTest reproduces the loss by delaying the EndTxn response past max.block.ms, and new unit tests in TaskManagerTest and StreamTaskTest cover the hand-over. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
