KAFKA-20785: Return PRODUCER_FENCED when EndTxn commit races with coordinator-side abort

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)

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)