[
https://issues.apache.org/jira/browse/KAFKA-21349?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Justine Olshan resolved KAFKA-21349.
------------------------------------
Resolution: Fixed
I will update the other fix versions when I cherrypick
> 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
> Fix For: 4.5
>
>
> 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)
