diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala index ccad3b0fe16a4..3665068bd5c9a 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala @@ -738,10 +738,18 @@ class TransactionCoordinator(txnConfig: TransactionConfig, +----------------+-------+---------+-------+---------+ | Empty | PF | EB | PF | ITS | +----------------+-------+---------+-------+---------+ - | CompleteAbort | NONE | EB | ITS | ITS | + | CompleteAbort | NONE | EB | PF | ITS | +----------------+-------+---------+-------+---------+ | CompleteCommit | ITS | EB | NONE | ITS | +----------------+-------+---------+-------+---------+ + + 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 (KAFKA-20785). */ /** @@ -880,7 +888,16 @@ class TransactionCoordinator(txnConfig: TransactionConfig, generateTxnTransitMetadataForTxnCompletion(TransactionState.PREPARE_ABORT, true) } else { // Commit. - logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult) + if (isRetry) { + // 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.PRODUCER_FENCED}.") + Left(Errors.PRODUCER_FENCED) + } 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..90bd5d985c2b9 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,55 @@ 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 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) + 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. 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.PRODUCER_FENCED, 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 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. + 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.PRODUCER_FENCED, error) + verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId)) + } + @Test def shouldReturnInvalidTxnRequestOnEndTxnRequestWhenStatusIsCompleteCommitAndResultIsNotCommit(): Unit = { val clientTransactionVersion = TransactionVersion.fromFeatureLevel(0)