artemlivshits opened a new pull request, #23943: URL: https://github.com/apache/kafka/pull/23943
When a broker becomes the leader of a `__transaction_state` partition, `TransactionStateManager` loads the log up to the log end offset. A newly elected leader can have records above the high watermark that are not committed yet and can still be truncated if leadership moves again, even with clean elections. The coordinator then acts on state that is later rolled back. For example, it can send COMMIT markers for a PREPARE_COMMIT that the next leader doesn't have, and that leader later times out and aborts the transaction, so the transaction is committed on some partitions and aborted on others. Loading with `FetchIsolation.HIGH_WATERMARK` is not enough, because a new leader's high watermark can lag behind the true one, so the load would miss committed records. The coordinator doesn't write to the partition while it is loading (requests get `COORDINATOR_LOAD_IN_PROGRESS`), so the log end offset at the start of the load doesn't change. Once the high watermark reaches it, every record the load reads is committed. This change: - In `loadTransactionsForTxnTopicPartition`, if the high watermark is below the log end offset, registers a `PartitionListener` and schedules the load once the high watermark reaches the log end offset. The high watermark is checked again after the listener is added, since an update between reading it and adding the listener isn't delivered to the listener. - Abandons the load (removes the `loadingPartitions` entry) if the partition is not online, becomes a follower, fails or is deleted before the high watermark catches up. This runs on the scheduler, since listeners are invoked with partition locks held. - Skips partitions that are loading in `removeExpiredTransactionalIds()`. On a leader epoch bump, the previous epoch's cache entry stays until the new load replaces it, and tombstones written for it would move the log end offset while the load waits. - Logs the time spent waiting for the high watermark as part of the load time. In the normal case the wait is about one follower fetch round trip. If an ISR follower is unavailable, the load waits until it is removed from the ISR (`replica.lag.time.max.ms`). This does not cover unclean leader election, where committed records can be lost. Testing: added unit tests to `TransactionStateManagerTest` for waiting for the high watermark, abandoning on each partition transition and when the partition is not online, the high watermark advancing during listener registration, a leader epoch bump while waiting, and expiration of a partition that is being reloaded. 🤖 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]
