[
https://issues.apache.org/jira/browse/KAFKA-21181?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ritika Reddy updated KAFKA-21181:
---------------------------------
Description:
When a broker loses leadership of a __transaction_state partition, there is a
race between the replica state change and the transaction coordinator's
resignation. BrokerMetadataPublisher.onMetadataUpdate applies the delta to
ReplicaManager first, which makes the broker a follower. It then updates the
group coordinator, and only after that calls
TransactionCoordinator.onResignation, which drops the partition's entry from
the TransactionStateManager cache. Until resignation runs, the broker is a
follower for the partition but still answers coordinator requests from its old
cache. The window is usually short, but its length depends on how much work the
publish does before it reaches the transaction coordinator, and it can stretch
to seconds.
TransactionStateManager.getAndMaybeAddTransactionState only checks
loadingPartitions and transactionMetadataCache; it never checks that the broker
still leads the partition. Answers that require a log write are safe: the
acks=-1 append fails with NOT_LEADER_OR_FOLLOWER, which is mapped to
NOT_COORDINATOR. Answers decided purely from the cache are not, so any request
that hits the old coordinator inside the window can get a wrong answer.
With TV2, one way to lose this race is a fatal producer error:
1. Transaction N commits on coordinator A. A's cache holds COMPLETE_COMMIT at
epoch E, and the client now has epoch E.
2. Leadership of the __transaction_state partition moves to broker B. B loads
the partition, and the partition leaders' implicit AddPartitionsToTxn for
transaction N+1 go to B, which moves the transaction to ONGOING.
3. The client has had no reason to rediscover the coordinator: under TV2,
EndTxn is its first coordinator request in the transaction. So it sends
EndTxn(COMMIT, E) to A.
4. If this request reaches A after A became a follower but before it resigned,
A answers from its cache. COMPLETE_COMMIT + COMMIT at the current epoch is not
a retry, so A returns INVALID_TXN_STATE without writing to the log:
TransactionalId: <id>'s state is COMPLETE_COMMIT, but received transaction
marker result to send: COMMIT
5. The producer treats INVALID_TXN_STATE as fatal. The transaction on B stays
ONGOING until it times out (or the producer re-initializes).
If the EndTxn arrives after A resigns, A returns NOT_COORDINATOR, and the
client rediscovers B and commits normally. The outcome depends only on which
side of that window the request lands. It doesn't need a broker pause or any
fault: an ordinary leadership move (reassignment, controlled shutdown,
preferred leader election) is enough, and producers that commit often are the
most exposed.
Other answers decided purely from the cache race the same way, for example
PRODUCER_FENCED / INVALID_PRODUCER_ID_MAPPING, the verify-only
AddPartitionsToTxn path, and the "partitions already added" shortcut.
was:
When a broker loses leadership of a __transaction_state partition, its
TransactionStateManager keeps the partition's cache entry until
TransactionCoordinator.onResignation runs. In
BrokerMetadataPublisher.onMetadataUpdate, that happens after
replicaManager.applyDelta and after the group coordinator update. So there's a
window in which the broker is already a follower for the partition but still
answers coordinator requests from its old cache.
TransactionStateManager.getAndMaybeAddTransactionState (trunk lines 400-421)
only checks loadingPartitions and transactionMetadataCache; it never checks
that the broker still leads the partition. Answers that require a log write are
safe, because the acks=-1 append fails with NOT_LEADER_OR_FOLLOWER, which is
mapped to NOT_COORDINATOR. Answers decided purely from the cache are not.
With TV2 this surfaces as a fatal error for the producer:
1. Transaction N commits on coordinator A. The cache on A holds COMPLETE_COMMIT
at epoch E, and the client now has epoch E.
2. Leadership of the __transaction_state partition moves to broker B. B loads
the partition, and the partition leaders' implicit AddPartitionsToTxn for
transaction N+1 go to B, which moves the transaction to ONGOING.
3. The client hasn't had any reason to rediscover the coordinator: under TV2
its first coordinator request in the transaction is EndTxn. So it sends
EndTxn(COMMIT, E) to A.
4. A is already a follower but hasn't resigned yet. It answers from the cache:
COMPLETE_COMMIT + COMMIT at the current epoch is not a retry, so it returns
INVALID_TXN_STATE without writing to the log:
TransactionalId: <id>'s state is COMPLETE_COMMIT, but received transaction
marker result to send: COMMIT
5. The producer treats INVALID_TXN_STATE as fatal. The transaction on B stays
ONGOING until it times out.
The correct response is NOT_COORDINATOR, which would make the client rediscover
B and commit. We've seen gaps of several seconds between the broker becoming
follower and resigning, and in one case ~37 s, with no broker pause involved.
Other cache-only answers are exposed the same way, for example PRODUCER_FENCED
/ INVALID_PRODUCER_ID_MAPPING, the verify-only AddPartitionsToTxn path, and the
"partitions already added" shortcut.
> Transaction coordinator could return INVALID_TXN_STATE from stale cache after
> losing __transaction_state leadership
> -------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21181
> URL: https://issues.apache.org/jira/browse/KAFKA-21181
> Project: Kafka
> Issue Type: Bug
> Affects Versions: 4.2.0, 4.3.0
> Reporter: Ritika Reddy
> Assignee: Ritika Reddy
> Priority: Major
> Fix For: 4.2.0, 4.3.0
>
>
> When a broker loses leadership of a __transaction_state partition, there is a
> race between the replica state change and the transaction coordinator's
> resignation. BrokerMetadataPublisher.onMetadataUpdate applies the delta to
> ReplicaManager first, which makes the broker a follower. It then updates the
> group coordinator, and only after that calls
> TransactionCoordinator.onResignation, which drops the partition's entry from
> the TransactionStateManager cache. Until resignation runs, the broker is a
> follower for the partition but still answers coordinator requests from its
> old cache. The window is usually short, but its length depends on how much
> work the publish does before it reaches the transaction coordinator, and it
> can stretch to seconds.
> TransactionStateManager.getAndMaybeAddTransactionState only checks
> loadingPartitions and transactionMetadataCache; it never checks that the
> broker still leads the partition. Answers that require a log write are safe:
> the acks=-1 append fails with NOT_LEADER_OR_FOLLOWER, which is mapped to
> NOT_COORDINATOR. Answers decided purely from the cache are not, so any
> request that hits the old coordinator inside the window can get a wrong
> answer.
> With TV2, one way to lose this race is a fatal producer error:
> 1. Transaction N commits on coordinator A. A's cache holds COMPLETE_COMMIT at
> epoch E, and the client now has epoch E.
> 2. Leadership of the __transaction_state partition moves to broker B. B loads
> the partition, and the partition leaders' implicit AddPartitionsToTxn for
> transaction N+1 go to B, which moves the transaction to ONGOING.
> 3. The client has had no reason to rediscover the coordinator: under TV2,
> EndTxn is its first coordinator request in the transaction. So it sends
> EndTxn(COMMIT, E) to A.
> 4. If this request reaches A after A became a follower but before it
> resigned, A answers from its cache. COMPLETE_COMMIT + COMMIT at the current
> epoch is not a retry, so A returns INVALID_TXN_STATE without writing to the
> log:
> TransactionalId: <id>'s state is COMPLETE_COMMIT, but received transaction
> marker result to send: COMMIT
> 5. The producer treats INVALID_TXN_STATE as fatal. The transaction on B stays
> ONGOING until it times out (or the producer re-initializes).
> If the EndTxn arrives after A resigns, A returns NOT_COORDINATOR, and the
> client rediscovers B and commits normally. The outcome depends only on which
> side of that window the request lands. It doesn't need a broker pause or any
> fault: an ordinary leadership move (reassignment, controlled shutdown,
> preferred leader election) is enough, and producers that commit often are the
> most exposed.
> Other answers decided purely from the cache race the same way, for example
> PRODUCER_FENCED / INVALID_PRODUCER_ID_MAPPING, the verify-only
> AddPartitionsToTxn path, and the "partitions already added" shortcut.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
