[jira] [Commented] (KAFKA-21349) Transaction coordinator loads uncommitted records from __transaction_state and can act on state that is later truncated

["Justine Olshan (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-21349?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18135310#comment-18135310 ] 

Justine Olshan commented on KAFKA-21349:
----------------------------------------

We probably want to cherry-pick this to some older branches, but I need to figure out how far back we want to go.

> Transaction coordinator loads uncommitted records from __transaction_state and can act on state that is later truncated
> -----------------------------------------------------------------------------------------------------------------------
>
>                 Key: KAFKA-21349
>                 URL: https://issues.apache.org/jira/browse/KAFKA-21349
>             Project: Kafka
>          Issue Type: Bug
>          Components: core
>            Reporter: Artem Livshits
>            Assignee: Artem Livshits
>            Priority: Major
>
> When a broker becomes the leader of a {{__transaction_state}} partition, {{TransactionStateManager.loadTransactionMetadata}} reads the log up to the log end offset ({{FetchIsolation.LOG_END}}). A newly elected leader can have records above the high watermark: they were replicated to it but not yet committed. These records can still be truncated if leadership moves again before they are committed. This can happen even with clean elections: another ISR member that doesn't have them can be elected.
> Then the coordinator acts on state that is later rolled back. The most damaging case is a PREPARE_COMMIT or PREPARE_ABORT above the high watermark. After the load, the coordinator sends the markers for it right away. For example:
> # The coordinator appends PREPARE_COMMIT for a transaction. The record reaches follower B but not follower C, and the leader fails before the high watermark advances.
> # B becomes the leader and loads the partition, including the uncommitted PREPARE_COMMIT. It starts sending COMMIT markers, and some data partitions commit.
> # B fails before C replicates the record. C, which is in the ISR, becomes the leader, and the PREPARE_COMMIT is gone. C loads the transaction as ONGOING.
> # The transaction times out, and C aborts it: PREPARE_ABORT, then ABORT markers to the remaining partitions.
> The transaction ends up committed on some partitions and aborted on others.
> Smaller effects of acting on state that is later truncated: client responses or fencing decisions based on state that disappears, and producer epochs being reused after the rollback.
> Since the coordinator only updates its in-memory state after an append with acks=all completes, only the last transition of each transactional id can be uncommitted. So the rollback is at most one step per transactional id.
> Using {{FetchIsolation.HIGH_WATERMARK}} for the load is not enough. A newly elected leader's high watermark can lag behind the true one, so the load would miss committed records. For example, it could miss a committed PREPARE_COMMIT and then abort the transaction on timeout.
> Proposed fix:
> 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. A new leader's high watermark only advances after the ISR followers fetch from it (their logs are prefixes of the leader's log after truncation). In the normal case, the wait is about one follower fetch round trip.
> * In {{loadTransactionsForTxnTopicPartition}}, if the high watermark is below the log end offset, register a {{PartitionListener}} ({{ReplicaManager.maybeAddListener}}). Schedule the load once {{onHighWatermarkUpdated}} reports an offset at or above the log end offset recorded when the wait started.
> * Check the high watermark again after registering. {{Partition.maybeAddListener}} replays the current state before it adds the listener, and high watermark updates from follower fetches can happen between the replay and the insertion.
> * If the leader epoch changes, the partition becomes a follower, or it fails or is deleted before the high watermark catches up, abandon the load: remove the epoch's {{loadingPartitions}} entry. Do this on the load scheduler, since the listener is invoked with partition locks held.
> * {{removeExpiredTransactionalIds()}} should skip partitions that are loading. On a leader epoch bump, the previous epoch's cache entry stays until the new load replaces it. Expiration can write tombstones for it, which would move the log end offset while the load waits.



--
This message was sent by Atlassian Jira
(v10.3.26#10030026)