Artem Livshits created KAFKA-21349:
--------------------------------------
Summary: 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
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)
