From b5a31b6b0eb40e5aac5da54987843c39f93ef55f Mon Sep 17 00:00:00 2001 From: Gavin Wang Date: Wed, 19 Aug 2026 22:41:12 -0400 Subject: [PATCH 1/4] KAFKA-20785: Return INVALID_PRODUCER_EPOCH when EndTxn commit races with coordinator-side abort Under transactions V2, when the transaction coordinator aborts an open transaction on its own (e.g. after transaction.timeout.ms elapses), the abort bumps the producer epoch. A commit that was already in flight when the abort completed arrives with the pre-abort epoch, which matches the V2 EndTxn retry condition (stored epoch == request epoch + 1), so it passes the epoch check and fails the state check instead: the coordinator returns INVALID_TXN_STATE, which producer clients treat as unconditionally fatal. Kafka Streams cannot recover from it and the affected StreamThread dies, even though the transaction was fully rolled back server-side and the commit never took effect. Under transactions V1 the same race fails the strict epoch check and returns PRODUCER_FENCED, which applications such as Kafka Streams handle by rebalancing; the fatal outcome is V2-only. Return the recoverable INVALID_PRODUCER_EPOCH instead of INVALID_TXN_STATE for a commit that arrives in COMPLETE_ABORT at a retry epoch (epoch-bump or producer-id-overflow variant). Only a coordinator-initiated abort can produce this combination, since a client retrying its own EndTxn always carries the operation it originally sent. Producer clients map INVALID_PRODUCER_EPOCH on the EndTxn response to ProducerFencedException, which Kafka Streams recovers from by rebalancing, restoring the V1 behavior for this race. A commit in COMPLETE_ABORT at the current epoch still returns INVALID_TXN_STATE, as that indicates a client-side bug rather than this race. This is the EndTxn-path analogue of KAFKA-19690, which made the same correction on the Produce path. --- .../transaction/TransactionCoordinator.scala | 24 +++++++++- .../TransactionCoordinatorTest.scala | 45 ++++++++++++++++++- 2 files changed, 66 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index ccad3b0fe16a4..2af7ee27e8fe0 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -716,6 +716,7 @@ class TransactionCoordinator(txnConfig: TransactionConfig, Note: PF = PRODUCER_FENCED ITS = INVALID_TXN_STATE + IPE = INVALID_PRODUCER_EPOCH NONE = No error and no epoch bump EB = No error and epoch bump @@ -738,10 +739,17 @@ class TransactionCoordinator(txnConfig: TransactionConfig, +----------------+-------+---------+-------+---------+ | Empty | PF | EB | PF | ITS | +----------------+-------+---------+-------+---------+ - | CompleteAbort | NONE | EB | ITS | ITS | + | CompleteAbort | NONE | EB | IPE | ITS | +----------------+-------+---------+-------+---------+ | CompleteCommit | ITS | EB | NONE | ITS | +----------------+-------+---------+-------+---------+ + + CompleteAbort + Commit + Retry returns IPE rather than ITS because the coordinator may abort an open + transaction on its own (e.g. when it exceeds transaction.timeout.ms), bumping the epoch without the + producer's knowledge. A commit that was already in flight when such an abort completed arrives with the + pre-abort epoch and is indistinguishable from a retry. The commit is guaranteed not to have taken effect, + so the recoverable INVALID_PRODUCER_EPOCH is returned, matching the transaction V1 behavior for this race, + instead of the fatal INVALID_TXN_STATE (KAFKA-20785). */ /** @@ -880,7 +888,19 @@ class TransactionCoordinator(txnConfig: TransactionConfig, generateTxnTransitMetadataForTxnCompletion(TransactionState.PREPARE_ABORT, true) } else { // Commit. - logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult) + if (isRetry) { + // The commit raced with an abort the producer did not request (e.g. the coordinator aborted the + // transaction after transaction.timeout.ms elapsed) and lost: the abort bumped the epoch, so the + // commit arrives with the pre-abort epoch. The commit is guaranteed not to have taken effect, so + // return the recoverable INVALID_PRODUCER_EPOCH, matching the transaction V1 behavior for this + // race, rather than the fatal INVALID_TXN_STATE (KAFKA-20785). + info(s"TransactionalId: $transactionalId's state is ${txnMetadata.state}, but received a COMMIT at " + + s"the pre-abort epoch $producerEpoch. The transaction was likely aborted by the coordinator on " + + s"timeout while the commit was in flight. Returning ${Errors.INVALID_PRODUCER_EPOCH}.") + Left(Errors.INVALID_PRODUCER_EPOCH) + } else { + logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult) + } } case TransactionState.PREPARE_COMMIT => if (txnMarkerResult == TransactionResult.COMMIT) diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala index 06c2709724526..8690d23886fc3 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala @@ -668,11 +668,54 @@ class TransactionCoordinatorTest { when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata)))) - coordinator.handleEndTransaction(transactionalId, producerId, requestEpoch(clientTransactionVersion), TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback) + // A commit at the current epoch is the next EndTxnRequest, not a retry, so the state transition is invalid. + coordinator.handleEndTransaction(transactionalId, producerId, producerEpoch, TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback) assertEquals(Errors.INVALID_TXN_STATE, error) verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId)) } + @Test + def shouldReturnInvalidProducerEpochOnEndTxnWhenStatusIsCompleteAbortAndCommitAtPreAbortEpochInV2(): Unit = { + val clientTransactionVersion = TransactionVersion.fromFeatureLevel(2) + val txnMetadata = new TransactionMetadata(transactionalId, producerId, producerId, RecordBatch.NO_PRODUCER_ID, + producerEpoch, (producerEpoch - 1).toShort, 1, TransactionState.COMPLETE_ABORT, util.Set.of, 0, time.milliseconds(), clientTransactionVersion) + when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) + .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata)))) + + // The coordinator aborted the transaction (e.g. on timeout) and bumped the epoch while the commit was in + // flight, so the commit arrives with the pre-abort epoch. This must not be the fatal INVALID_TXN_STATE: + // the commit did not take effect, and the producer can recover by aborting (KAFKA-20785). + coordinator.handleEndTransaction(transactionalId, producerId, (producerEpoch - 1).toShort, TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback) + assertEquals(Errors.INVALID_PRODUCER_EPOCH, error) + verify(transactionManager, never()).appendTransactionToLog( + ArgumentMatchers.eq(transactionalId), + ArgumentMatchers.any(), + ArgumentMatchers.any(), + ArgumentMatchers.any(), + ArgumentMatchers.any(), + ArgumentMatchers.any() + ) + verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId)) + } + + @Test + def shouldReturnInvalidProducerEpochOnEndTxnWhenStatusIsCompleteAbortAndCommitOnRetryOverflowInV2(): Unit = { + val clientTransactionVersion = TransactionVersion.fromFeatureLevel(2) + // The coordinator-side abort exhausted the epoch, rotating to a new producer ID with epoch 0 and recording + // the old producer ID in prevProducerId. + val newProducerId = producerId + 1 + val txnMetadata = new TransactionMetadata(transactionalId, newProducerId, producerId, RecordBatch.NO_PRODUCER_ID, + 0.toShort, RecordBatch.NO_PRODUCER_EPOCH, 1, TransactionState.COMPLETE_ABORT, util.Set.of, 0, time.milliseconds(), clientTransactionVersion) + when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId))) + .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata)))) + + // Same race as above, but the pre-abort epoch was Short.MaxValue - 1, so the stale commit matches the + // retry-on-overflow condition instead of the epoch-bump one. + coordinator.handleEndTransaction(transactionalId, producerId, (Short.MaxValue - 1).toShort, TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback) + assertEquals(Errors.INVALID_PRODUCER_EPOCH, error) + verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId)) + } + @Test def shouldReturnInvalidTxnRequestOnEndTxnRequestWhenStatusIsCompleteCommitAndResultIsNotCommit(): Unit = { val clientTransactionVersion = TransactionVersion.fromFeatureLevel(0) From dd20b3d9b318ee846a9f26958f7b3e9ad391fb31 Mon Sep 17 00:00:00 2001 From: Gavin Wang Date: Fri, 28 Aug 2026 14:36:09 -0400 Subject: [PATCH 2/4] KAFKA-20785: Clarify why INVALID_PRODUCER_EPOCH is returned for the commit/abort race The previous comments said the error choice matches transaction V1 behavior, which misread as V1 returning INVALID_PRODUCER_EPOCH here. V1 actually fails this race at its strict epoch check with PRODUCER_FENCED; what the fix restores is the recoverable outcome, not the error code. State that explicitly in the state-table note, along with why INVALID_PRODUCER_EPOCH is used instead of PRODUCER_FENCED: no newer producer exists (the epoch is merely stale), it mirrors KAFKA-19690's produce-path fix in UnifiedLog, and producer clients treat the two errors identically on EndTxn responses. The branch comment now defers to the state-table note instead of repeating it. --- .../transaction/TransactionCoordinator.scala | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index 2af7ee27e8fe0..746d762cc5fca 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -747,9 +747,13 @@ class TransactionCoordinator(txnConfig: TransactionConfig, CompleteAbort + Commit + Retry returns IPE rather than ITS because the coordinator may abort an open transaction on its own (e.g. when it exceeds transaction.timeout.ms), bumping the epoch without the producer's knowledge. A commit that was already in flight when such an abort completed arrives with the - pre-abort epoch and is indistinguishable from a retry. The commit is guaranteed not to have taken effect, - so the recoverable INVALID_PRODUCER_EPOCH is returned, matching the transaction V1 behavior for this race, - instead of the fatal INVALID_TXN_STATE (KAFKA-20785). + pre-abort epoch and is indistinguishable from a retry. The commit is guaranteed not to have taken effect. + Under transaction V1 this race fails the strict epoch check above and returns the recoverable + PRODUCER_FENCED; V2's retry-tolerant epoch check accepts the request instead, so the recoverable outcome + is restored here at the state check. INVALID_PRODUCER_EPOCH is used rather than PRODUCER_FENCED because + no newer producer exists (the epoch is merely stale), mirroring the produce path's fix for the same race + (KAFKA-19690, UnifiedLog); the producer client treats the two errors identically on EndTxn responses + (KAFKA-20785). */ /** @@ -889,11 +893,8 @@ class TransactionCoordinator(txnConfig: TransactionConfig, } else { // Commit. if (isRetry) { - // The commit raced with an abort the producer did not request (e.g. the coordinator aborted the - // transaction after transaction.timeout.ms elapsed) and lost: the abort bumped the epoch, so the - // commit arrives with the pre-abort epoch. The commit is guaranteed not to have taken effect, so - // return the recoverable INVALID_PRODUCER_EPOCH, matching the transaction V1 behavior for this - // race, rather than the fatal INVALID_TXN_STATE (KAFKA-20785). + // The commit raced with a coordinator-side abort (e.g. on transaction.timeout.ms) and is guaranteed + // not to have taken effect; see the CompleteAbort + Commit + Retry note under the state table above (KAFKA-20785). info(s"TransactionalId: $transactionalId's state is ${txnMetadata.state}, but received a COMMIT at " + s"the pre-abort epoch $producerEpoch. The transaction was likely aborted by the coordinator on " + s"timeout while the commit was in flight. Returning ${Errors.INVALID_PRODUCER_EPOCH}.") From 2fd9f83130f18b612ddf3c79ca6eece5ae5d0127 Mon Sep 17 00:00:00 2001 From: Gavin Wang Date: Fri, 28 Aug 2026 14:59:54 -0400 Subject: [PATCH 3/4] KAFKA-20785: Switch the EndTxn commit/abort race error to PRODUCER_FENCED Transactional requests conventionally return PRODUCER_FENCED on epoch mismatches, and PRODUCER_FENCED is also what transactions V1's strict epoch check returns for this same race; INVALID_PRODUCER_EPOCH is the produce-path convention (KAFKA-19690). Client behavior is unchanged: the producer client already folds INVALID_PRODUCER_EPOCH into ProducerFencedException on EndTxn responses, so the two codes are handled identically. --- .../transaction/TransactionCoordinator.scala | 19 ++++++++++--------- .../TransactionCoordinatorTest.scala | 11 ++++++----- 2 files changed, 16 insertions(+), 14 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index 746d762cc5fca..0fcca70514852 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -716,7 +716,6 @@ class TransactionCoordinator(txnConfig: TransactionConfig, Note: PF = PRODUCER_FENCED ITS = INVALID_TXN_STATE - IPE = INVALID_PRODUCER_EPOCH NONE = No error and no epoch bump EB = No error and epoch bump @@ -739,21 +738,23 @@ class TransactionCoordinator(txnConfig: TransactionConfig, +----------------+-------+---------+-------+---------+ | Empty | PF | EB | PF | ITS | +----------------+-------+---------+-------+---------+ - | CompleteAbort | NONE | EB | IPE | ITS | + | CompleteAbort | NONE | EB | PF | ITS | +----------------+-------+---------+-------+---------+ | CompleteCommit | ITS | EB | NONE | ITS | +----------------+-------+---------+-------+---------+ - CompleteAbort + Commit + Retry returns IPE rather than ITS because the coordinator may abort an open + CompleteAbort + Commit + Retry returns PF rather than ITS because the coordinator may abort an open transaction on its own (e.g. when it exceeds transaction.timeout.ms), bumping the epoch without the producer's knowledge. A commit that was already in flight when such an abort completed arrives with the pre-abort epoch and is indistinguishable from a retry. The commit is guaranteed not to have taken effect. Under transaction V1 this race fails the strict epoch check above and returns the recoverable PRODUCER_FENCED; V2's retry-tolerant epoch check accepts the request instead, so the recoverable outcome - is restored here at the state check. INVALID_PRODUCER_EPOCH is used rather than PRODUCER_FENCED because - no newer producer exists (the epoch is merely stale), mirroring the produce path's fix for the same race - (KAFKA-19690, UnifiedLog); the producer client treats the two errors identically on EndTxn responses - (KAFKA-20785). + is restored here at the state check (KAFKA-20785). Strictly speaking no newer producer took over the + transactional.id — the epoch is merely stale after the coordinator-side bump — but transactional requests + conventionally return PRODUCER_FENCED for epoch mismatches, and the producer client folds + INVALID_PRODUCER_EPOCH into ProducerFencedException on EndTxn responses anyway, so the two codes behave + identically. The produce path's fix for the same race returns INVALID_PRODUCER_EPOCH, following the + produce-path convention (KAFKA-19690, UnifiedLog). */ /** @@ -897,8 +898,8 @@ class TransactionCoordinator(txnConfig: TransactionConfig, // not to have taken effect; see the CompleteAbort + Commit + Retry note under the state table above (KAFKA-20785). info(s"TransactionalId: $transactionalId's state is ${txnMetadata.state}, but received a COMMIT at " + s"the pre-abort epoch $producerEpoch. The transaction was likely aborted by the coordinator on " + - s"timeout while the commit was in flight. Returning ${Errors.INVALID_PRODUCER_EPOCH}.") - Left(Errors.INVALID_PRODUCER_EPOCH) + s"timeout while the commit was in flight. Returning ${Errors.PRODUCER_FENCED}.") + Left(Errors.PRODUCER_FENCED) } else { logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult) } diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala index 8690d23886fc3..90bd5d985c2b9 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala @@ -675,7 +675,7 @@ class TransactionCoordinatorTest { } @Test - def shouldReturnInvalidProducerEpochOnEndTxnWhenStatusIsCompleteAbortAndCommitAtPreAbortEpochInV2(): Unit = { + def shouldReturnProducerFencedOnEndTxnWhenStatusIsCompleteAbortAndCommitAtPreAbortEpochInV2(): Unit = { val clientTransactionVersion = TransactionVersion.fromFeatureLevel(2) val txnMetadata = new TransactionMetadata(transactionalId, producerId, producerId, RecordBatch.NO_PRODUCER_ID, producerEpoch, (producerEpoch - 1).toShort, 1, TransactionState.COMPLETE_ABORT, util.Set.of, 0, time.milliseconds(), clientTransactionVersion) @@ -684,9 +684,10 @@ class TransactionCoordinatorTest { // The coordinator aborted the transaction (e.g. on timeout) and bumped the epoch while the commit was in // flight, so the commit arrives with the pre-abort epoch. This must not be the fatal INVALID_TXN_STATE: - // the commit did not take effect, and the producer can recover by aborting (KAFKA-20785). + // the commit did not take effect, and the producer can recover by aborting. PRODUCER_FENCED follows the + // transactional-request convention and matches what V1's strict epoch check returns for this race (KAFKA-20785). coordinator.handleEndTransaction(transactionalId, producerId, (producerEpoch - 1).toShort, TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback) - assertEquals(Errors.INVALID_PRODUCER_EPOCH, error) + assertEquals(Errors.PRODUCER_FENCED, error) verify(transactionManager, never()).appendTransactionToLog( ArgumentMatchers.eq(transactionalId), ArgumentMatchers.any(), @@ -699,7 +700,7 @@ class TransactionCoordinatorTest { } @Test - def shouldReturnInvalidProducerEpochOnEndTxnWhenStatusIsCompleteAbortAndCommitOnRetryOverflowInV2(): Unit = { + def shouldReturnProducerFencedOnEndTxnWhenStatusIsCompleteAbortAndCommitOnRetryOverflowInV2(): Unit = { val clientTransactionVersion = TransactionVersion.fromFeatureLevel(2) // The coordinator-side abort exhausted the epoch, rotating to a new producer ID with epoch 0 and recording // the old producer ID in prevProducerId. @@ -712,7 +713,7 @@ class TransactionCoordinatorTest { // Same race as above, but the pre-abort epoch was Short.MaxValue - 1, so the stale commit matches the // retry-on-overflow condition instead of the epoch-bump one. coordinator.handleEndTransaction(transactionalId, producerId, (Short.MaxValue - 1).toShort, TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback) - assertEquals(Errors.INVALID_PRODUCER_EPOCH, error) + assertEquals(Errors.PRODUCER_FENCED, error) verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId)) } From 4786396286f5c59649753a2aaf6338afd7c760b8 Mon Sep 17 00:00:00 2001 From: Gavin Wang Date: Fri, 28 Aug 2026 15:10:03 -0400 Subject: [PATCH 4/4] update comment --- .../coordinator/transaction/TransactionCoordinator.scala | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index 0fcca70514852..3665068bd5c9a 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -749,12 +749,7 @@ class TransactionCoordinator(txnConfig: TransactionConfig, pre-abort epoch and is indistinguishable from a retry. The commit is guaranteed not to have taken effect. Under transaction V1 this race fails the strict epoch check above and returns the recoverable PRODUCER_FENCED; V2's retry-tolerant epoch check accepts the request instead, so the recoverable outcome - is restored here at the state check (KAFKA-20785). Strictly speaking no newer producer took over the - transactional.id — the epoch is merely stale after the coordinator-side bump — but transactional requests - conventionally return PRODUCER_FENCED for epoch mismatches, and the producer client folds - INVALID_PRODUCER_EPOCH into ProducerFencedException on EndTxn responses anyway, so the two codes behave - identically. The produce path's fix for the same race returns INVALID_PRODUCER_EPOCH, following the - produce-path convention (KAFKA-19690, UnifiedLog). + is restored here at the state check (KAFKA-20785). */ /**