KAFKA-21181: Allow committing an empty transaction in TV2 (#23831)

Follow-up to #23630, addressing
https://github.com/apache/kafka/pull/23630#discussion_r4151983373.
#23630
returns NOT_COORDINATOR once the broker has applied the leadership
change;
this covers a coordinator whose metadata is still behind.

With transaction V2, an EndTxn(COMMIT) at the current epoch on a
transaction
in Empty, CompleteCommit or CompleteAbort state returned
INVALID_TXN_STATE,
which is fatal to the client. A coordinator that has lost leadership but
not
yet applied the metadata change sees the transaction in exactly this
state,
since its partitions were added on the new coordinator, and answered
from its
stale cache.

Commit an empty transaction instead, the same way an empty abort is
handled:
bump the epoch, write PrepareCommit with no partitions and complete it
without
markers.

- A stale coordinator cannot get that write replicated. The transaction
log
is written with acks=-1, and the new leader is in the old leader's ISR
but
stopped fetching from it before the partitions were added. The client
gets
a retriable error (NOT_COORDINATOR or COORDINATOR_NOT_AVAILABLE), looks
up
  the coordinator again and commits on the new one.
- A coordinator that is up to date only sees an empty transaction when
the
client wrote nothing, since a TV2 produce always adds its partition to
the
  transaction before it succeeds. Committing it is a no-op epoch bump.

Retries are unchanged: CompleteCommit + Commit + Retry still returns
NONE and
CompleteAbort + Commit + Retry still returns PRODUCER_FENCED.
Transaction V1
is unchanged. A PrepareCommit with no partitions is loaded and completed
the
same way as the empty PrepareAbort that TV2 already writes.

Tests cover the updated state table, completing without markers, the
retriable error when the write fails, producer ID rotation on an
exhausted
epoch, and loading an empty PrepareCommit from the log.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Reviewers: Justine Olshan <jolshan@confluent.io>, Artem Livshits
<alivshits@confluent.io>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala

@@ -736,13 +736,19 @@ class TransactionCoordinator(txnConfig: TransactionConfig,
     +----------------+-------+---------+-------+---------+
     |                | Retry | Current | Retry | Current |
     +----------------+-------+---------+-------+---------+
-    | Empty          | PF    | EB      | PF    | ITS     |
+    | Empty          | PF    | EB      | PF    | EB      |
     +----------------+-------+---------+-------+---------+
-    | CompleteAbort  | NONE  | EB      | PF    | ITS     |
+    | CompleteAbort  | NONE  | EB      | PF    | EB      |
     +----------------+-------+---------+-------+---------+
-    | CompleteCommit | ITS   | EB      | NONE  | ITS     |
+    | CompleteCommit | ITS   | EB      | NONE  | EB      |
     +----------------+-------+---------+-------+---------+
 
+    Empty, CompleteAbort and CompleteCommit + Commit + Current commits an empty transaction, the same way an abort
+    there aborts one: the epoch is bumped and the transition is written to the log. A coordinator that has lost
+    leadership but has not applied the metadata change yet can only see the transaction as empty, since the
+    partitions were added on the new coordinator. Returning ITS from its cache would be fatal to the client; going
+    through the log instead fails with a retriable error because the write cannot be replicated (KAFKA-21181).
+
     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
@@ -850,6 +856,12 @@ class TransactionCoordinator(txnConfig: TransactionConfig,
               }
             }
 
+            def commitEmptyTransaction(): ApiResult[(Int, TxnTransitMetadata)] = {
+              info(s"TransactionalId: $transactionalId's state is ${txnMetadata.state} with no partitions, committing an " +
+                s"empty transaction for producer $producerId at epoch $producerEpoch.")
+              generateTxnTransitMetadataForTxnCompletion(TransactionState.PREPARE_COMMIT, true)
+            }
+
             if (txnMetadata.pendingTransitionInProgress && txnMetadata.pendingState.get != TransactionState.PREPARE_EPOCH_FENCE) {
               // This check is performed first so that the pending transition can complete before the next checks.
               // With TV2, we may be transitioning over a producer epoch overflow, and the producer may be using the
@@ -872,7 +884,7 @@ class TransactionCoordinator(txnConfig: TransactionConfig,
                   if (isRetry)
                     Left(Errors.NONE)
                   else
-                    logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult)
+                    commitEmptyTransaction()
                 } else {
                   // Abort.
                   if (isRetry)
@@ -896,7 +908,7 @@ class TransactionCoordinator(txnConfig: TransactionConfig,
                       s"timeout while the commit was in flight. Returning ${Errors.PRODUCER_FENCED}.")
                     Left(Errors.PRODUCER_FENCED)
                   } else {
-                    logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult)
+                    commitEmptyTransaction()
                   }
                 }
               case TransactionState.PREPARE_COMMIT =>
@@ -913,7 +925,7 @@ class TransactionCoordinator(txnConfig: TransactionConfig,
                 if (txnMarkerResult == TransactionResult.ABORT) {
                   generateTxnTransitMetadataForTxnCompletion(TransactionState.PREPARE_ABORT, true)
                 } else {
-                  logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult)
+                  commitEmptyTransaction()
                 }
               case TransactionState.DEAD | TransactionState.PREPARE_EPOCH_FENCE =>
                 val errorMsg = s"Found transactionalId $transactionalId with state ${txnMetadata.state}. " +

core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorConcurrencyTest.scala

@@ -186,19 +186,24 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
       TransactionVersion.TV_2))
     val txnPartition = txnStateManager.partitionFor(transactionalId)
 
-    def commit(): Errors = {
+    def commit(epoch: Short): Errors = {
       var error: Errors = null
-      transactionCoordinator.handleEndTransaction(transactionalId, producerId, producerEpoch, TransactionResult.COMMIT,
+      transactionCoordinator.handleEndTransaction(transactionalId, producerId, epoch, TransactionResult.COMMIT,
         TransactionVersion.TV_2, (e, _, _) => error = e)
+      // the test replica manager completes a write after a few attempts
+      TestUtils.waitUntilTrue(() => {
+        replicaManager.tryCompleteActions()
+        error != null
+      }, "EndTxn did not complete")
       error
     }
 
-    // committing at the current epoch with no new transaction is invalid
-    assertEquals(Errors.INVALID_TXN_STATE, commit())
+    // committing at the current epoch with no new transaction commits an empty transaction and bumps the epoch
+    assertEquals(Errors.NONE, commit(producerEpoch))
 
     // the replica became a follower but the partition has not been unloaded yet
     loseLeadership(txnPartition)
-    assertEquals(Errors.NOT_COORDINATOR, commit())
+    assertEquals(Errors.NOT_COORDINATOR, commit((producerEpoch + 1).toShort))
   }
 
   @Test

core/src/test/scala/unit/kafka/coordinator/transaction/TransactionCoordinatorTest.scala

@@ -31,7 +31,7 @@ import org.apache.kafka.server.util.MockScheduler
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.{CsvSource, ValueSource}
+import org.junit.jupiter.params.provider.{CsvSource, EnumSource, ValueSource}
 import org.mockito.ArgumentMatchers.{any, anyBoolean, anyInt}
 import org.mockito.Mockito._
 import org.mockito.{ArgumentCaptor, ArgumentMatchers}
@@ -597,8 +597,9 @@ class TransactionCoordinatorTest {
     coordinator.handleEndTransaction(transactionalId, producerId, epoch.toShort, TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback)
     if (isRetry) {
       assertEquals(Errors.NONE, error)
+      verifyNoAppend()
     } else {
-      assertEquals(Errors.INVALID_TXN_STATE, error)
+      verifyEmptyCommitAppended()
     }
     verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId))
   }
@@ -659,21 +660,34 @@ class TransactionCoordinatorTest {
     verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId))
   }
 
-  @ParameterizedTest
-  @ValueSource(shorts = Array(0, 2))
-  def shouldReturnInvalidTxnRequestOnEndTxnRequestWhenStatusIsCompleteAbortAndResultIsNotAbort(transactionVersion: Short): Unit = {
-    val clientTransactionVersion = TransactionVersion.fromFeatureLevel(transactionVersion)
+  @Test
+  def shouldReturnInvalidTxnRequestOnEndTxnRequestWhenStatusIsCompleteAbortAndResultIsNotAbort(): Unit = {
+    val clientTransactionVersion = TransactionVersion.fromFeatureLevel(0)
     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))))
 
-    // A commit at the current epoch is the next EndTxnRequest, not a retry, so the state transition is invalid.
+    // With transaction V1, CompleteAbort only accepts Abort.
     coordinator.handleEndTransaction(transactionalId, producerId, producerEpoch, TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback)
     assertEquals(Errors.INVALID_TXN_STATE, error)
     verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId))
   }
 
+  @Test
+  def shouldCommitEmptyTransactionOnEndTxnRequestWhenStatusIsCompleteAbortAndResultIsCommitInV2(): 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))))
+
+    // A commit at the current epoch is the next EndTxnRequest, not a retry, so an empty transaction is committed.
+    coordinator.handleEndTransaction(transactionalId, producerId, producerEpoch, TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback)
+    verifyEmptyCommitAppended()
+    verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId))
+  }
+
   @Test
   def shouldReturnProducerFencedOnEndTxnWhenStatusIsCompleteAbortAndCommitAtPreAbortEpochInV2(): Unit = {
     val clientTransactionVersion = TransactionVersion.fromFeatureLevel(2)
@@ -858,14 +872,126 @@ class TransactionCoordinatorTest {
     coordinator.handleEndTransaction(transactionalId, producerId, epoch.toShort, TransactionResult.COMMIT, clientTransactionVersion, endTxnCallback)
     if (isRetry) {
       assertEquals(Errors.PRODUCER_FENCED, error)
+      verifyNoAppend()
     } else {
-      assertEquals(Errors.INVALID_TXN_STATE, error)
+      verifyEmptyCommitAppended()
     }
     verify(transactionManager).getTransactionState(ArgumentMatchers.eq(transactionalId))
   }
 
+  @ParameterizedTest
+  @EnumSource(value = classOf[TransactionState], names = Array("EMPTY", "COMPLETE_COMMIT", "COMPLETE_ABORT"))
+  def shouldPrepareEmptyCommitInV2(state: TransactionState): Unit = {
+    val txnMetadata = new TransactionMetadata(transactionalId, producerId, producerId, RecordBatch.NO_PRODUCER_ID,
+      producerEpoch, (producerEpoch - 1).toShort, 1, state, util.Set.of, 0, time.milliseconds(), TV_2)
+    when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId)))
+      .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata))))
+    when(transactionManager.appendTransactionToLog(
+      ArgumentMatchers.eq(transactionalId),
+      ArgumentMatchers.eq(coordinatorEpoch),
+      capturedTxnTransitMetadata.capture(),
+      capturedErrorsCallback.capture(),
+      any(),
+      any())
+    ).thenAnswer(_ => {
+      txnMetadata.completeTransitionTo(capturedTxnTransitMetadata.getValue)
+      capturedErrorsCallback.getValue.apply(Errors.NONE)
+    })
+
+    coordinator.handleEndTransaction(transactionalId, producerId, producerEpoch, TransactionResult.COMMIT, TV_2, endTxnCallback)
+
+    assertEquals(Errors.NONE, error)
+    assertEquals(producerId, newProducerId)
+    assertEquals(producerEpoch + 1, newEpoch)
+    assertEquals(TransactionState.PREPARE_COMMIT, txnMetadata.state)
+    assertTrue(txnMetadata.topicPartitions.isEmpty)
+    verify(transactionMarkerChannelManager).addTxnMarkersToSend(
+      ArgumentMatchers.eq(coordinatorEpoch),
+      ArgumentMatchers.eq(TransactionResult.COMMIT),
+      ArgumentMatchers.eq(txnMetadata),
+      any[TxnTransitMetadata]())
+  }
+
+  @ParameterizedTest
+  @EnumSource(value = classOf[TransactionState], names = Array("EMPTY", "COMPLETE_COMMIT", "COMPLETE_ABORT"))
+  def shouldGenerateNewProducerIdOnEmptyCommitIfEpochsExhaustedV2(state: TransactionState): Unit = {
+    mockPidGenerator()
+    nextPid = producerId2
+    val txnMetadata = new TransactionMetadata(transactionalId, producerId, producerId, RecordBatch.NO_PRODUCER_ID,
+      (Short.MaxValue - 1).toShort, (Short.MaxValue - 2).toShort, txnTimeoutMs, state, util.Set.of, 0, time.milliseconds(), TV_2)
+    when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId)))
+      .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata))))
+    when(transactionManager.appendTransactionToLog(
+      ArgumentMatchers.eq(transactionalId),
+      ArgumentMatchers.eq(coordinatorEpoch),
+      capturedTxnTransitMetadata.capture(),
+      capturedErrorsCallback.capture(),
+      any(),
+      any())
+    ).thenAnswer(_ => {
+      txnMetadata.completeTransitionTo(capturedTxnTransitMetadata.getValue)
+      capturedErrorsCallback.getValue.apply(Errors.NONE)
+    })
+
+    coordinator.handleEndTransaction(transactionalId, producerId, (Short.MaxValue - 1).toShort, TransactionResult.COMMIT, TV_2, endTxnCallback)
+
+    // The empty commit bumps the epoch to Short.MaxValue, so the producer ID is rotated on completion.
+    assertEquals(Errors.NONE, error)
+    assertEquals(producerId2, newProducerId)
+    assertEquals(0, newEpoch)
+    assertEquals(TransactionState.PREPARE_COMMIT, txnMetadata.state)
+    assertEquals(Short.MaxValue, txnMetadata.producerEpoch)
+    assertEquals(producerId2, capturedTxnTransitMetadata.getValue.nextProducerId)
+    assertTrue(txnMetadata.topicPartitions.isEmpty)
+  }
+
+  @ParameterizedTest
+  @EnumSource(value = classOf[Errors], names = Array("NOT_COORDINATOR", "COORDINATOR_NOT_AVAILABLE"))
+  def shouldReturnRetriableErrorWhenEmptyCommitCannotBeWrittenInV2(appendError: Errors): Unit = {
+    // A stale coordinator sees the transaction as empty since its partitions were added on the new coordinator,
+    // but it cannot replicate the write, so the client gets a retriable error rather than INVALID_TXN_STATE.
+    val txnMetadata = new TransactionMetadata(transactionalId, producerId, producerId, RecordBatch.NO_PRODUCER_ID,
+      producerEpoch, (producerEpoch - 1).toShort, 1, TransactionState.COMPLETE_COMMIT, util.Set.of, 0, time.milliseconds(), TV_2)
+    when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId)))
+      .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata))))
+    when(transactionManager.appendTransactionToLog(
+      ArgumentMatchers.eq(transactionalId),
+      ArgumentMatchers.eq(coordinatorEpoch),
+      any[TxnTransitMetadata],
+      capturedErrorsCallback.capture(),
+      any(),
+      any())
+    ).thenAnswer(_ => capturedErrorsCallback.getValue.apply(appendError))
+
+    coordinator.handleEndTransaction(transactionalId, producerId, producerEpoch, TransactionResult.COMMIT, TV_2, endTxnCallback)
+
+    assertEquals(appendError, error)
+    assertEquals(TransactionState.COMPLETE_COMMIT, txnMetadata.state)
+    assertEquals(producerEpoch, txnMetadata.producerEpoch)
+    verify(transactionMarkerChannelManager, never()).addTxnMarkersToSend(anyInt(), any(), any(), any())
+  }
+
+  private def verifyNoAppend(): Unit =
+    verify(transactionManager, never()).appendTransactionToLog(
+      ArgumentMatchers.eq(transactionalId), any(), any(), any(), any(), any())
+
+  private def verifyEmptyCommitAppended(): Unit = {
+    val newMetadata = ArgumentCaptor.forClass(classOf[TxnTransitMetadata])
+    verify(transactionManager).appendTransactionToLog(
+      ArgumentMatchers.eq(transactionalId),
+      ArgumentMatchers.eq(coordinatorEpoch),
+      newMetadata.capture(),
+      any(),
+      any(),
+      any())
+    assertEquals(TransactionState.PREPARE_COMMIT, newMetadata.getValue.txnState)
+    assertEquals(producerEpoch + 1, newMetadata.getValue.producerEpoch)
+    assertEquals(time.milliseconds(), newMetadata.getValue.txnStartTimestamp)
+    assertTrue(newMetadata.getValue.topicPartitions.isEmpty)
+  }
+
   @Test
-  def shouldReturnInvalidTxnRequestOnEndTxnV2IfNotEndTxnV2Retry(): Unit = {
+  def shouldTreatEndTxnV2AtCurrentEpochAsNextRequestNotRetry(): Unit = {
     when(transactionManager.getTransactionState(ArgumentMatchers.eq(transactionalId)))
       .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, new TransactionMetadata(transactionalId, producerId, producerId,
         RecordBatch.NO_PRODUCER_ID, producerEpoch, RecordBatch.NO_PRODUCER_EPOCH, 1, TransactionState.PREPARE_COMMIT, util.Set.of, 0, time.milliseconds(), TV_2)))))
@@ -879,9 +1005,9 @@ class TransactionCoordinatorTest {
       .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, new TransactionMetadata(transactionalId, producerId, producerId,
         RecordBatch.NO_PRODUCER_ID, producerEpoch, RecordBatch.NO_PRODUCER_EPOCH, 1, TransactionState.COMPLETE_COMMIT, util.Set.of, 0, time.milliseconds(), TV_2)))))
 
-    // If producerEpoch is the same, this is not a retry of the EndTxnRequest, but the next EndTxnRequest. Return INVALID_TXN_STATE.
+    // If producerEpoch is the same, this is not a retry of the EndTxnRequest, but the next EndTxnRequest. Commit an empty transaction.
     coordinator.handleEndTransaction(transactionalId, producerId, producerEpoch, TransactionResult.COMMIT, TV_2, endTxnCallback)
-    assertEquals(Errors.INVALID_TXN_STATE, error)
+    verifyEmptyCommitAppended()
     verify(transactionManager, times(2)).getTransactionState(ArgumentMatchers.eq(transactionalId))
   }
 

core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala

@@ -37,7 +37,7 @@ import org.apache.kafka.server.util.{InterBrokerSendThread, RequestAndCompletion
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.ValueSource
+import org.junit.jupiter.params.provider.{EnumSource, ValueSource}
 import org.mockito.ArgumentMatchers.any
 import org.mockito.{ArgumentCaptor, ArgumentMatchers}
 import org.mockito.Mockito.{clearInvocations, mock, mockConstruction, times, verify, verifyNoMoreInteractions, when}
@@ -495,6 +495,48 @@ class TransactionMarkerChannelManagerTest {
     assertEquals(TransactionState.COMPLETE_COMMIT, txnMetadata2.state)
   }
 
+  @ParameterizedTest
+  @EnumSource(value = classOf[TransactionResult], names = Array("COMMIT", "ABORT"))
+  def shouldCompleteTxnWithoutSendingMarkersWhenNoPartitions(result: TransactionResult): Unit = {
+    // With transaction V2, ending a transaction with no partitions writes PrepareCommit or PrepareAbort with no partitions.
+    val (prepareState, completeState) =
+      if (result == TransactionResult.COMMIT) (TransactionState.PREPARE_COMMIT, TransactionState.COMPLETE_COMMIT)
+      else (TransactionState.PREPARE_ABORT, TransactionState.COMPLETE_ABORT)
+    val txnMetadata = new TransactionMetadata(transactionalId1, producerId1, producerId1, RecordBatch.NO_PRODUCER_ID,
+      producerEpoch, lastProducerEpoch, txnTimeoutMs, prepareState, util.Set.of, 0L, 0L, TransactionVersion.TV_2)
+    when(txnStateManager.getTransactionState(ArgumentMatchers.eq(transactionalId1)))
+      .thenReturn(Right(Some(new CoordinatorEpochAndTxnMetadata(coordinatorEpoch, txnMetadata))))
+
+    val txnTransitionMetadata = txnMetadata.prepareComplete(time.milliseconds())
+
+    when(txnStateManager.appendTransactionToLog(
+      ArgumentMatchers.eq(transactionalId1),
+      ArgumentMatchers.eq(coordinatorEpoch),
+      ArgumentMatchers.eq(txnTransitionMetadata),
+      capturedErrorsCallback.capture(),
+      any(),
+      any()))
+      .thenAnswer(_ => {
+        txnMetadata.completeTransitionTo(txnTransitionMetadata)
+        capturedErrorsCallback.getValue.apply(Errors.NONE)
+      })
+
+    channelManager.addTxnMarkersToSend(coordinatorEpoch, result, txnMetadata, txnTransitionMetadata)
+
+    // The completion is written right away since there are no markers to wait for.
+    verify(txnStateManager).appendTransactionToLog(
+      ArgumentMatchers.eq(transactionalId1),
+      ArgumentMatchers.eq(coordinatorEpoch),
+      ArgumentMatchers.eq(txnTransitionMetadata),
+      any(),
+      any(),
+      any())
+    assertTrue(channelManager.generateRequests().isEmpty)
+    assertEquals(0, channelManager.numTxnsWithPendingMarkers)
+    assertEquals(Optional.empty(), txnMetadata.pendingState)
+    assertEquals(completeState, txnMetadata.state)
+  }
+
   @Test
   def shouldAbortAppendToLogOnEndTxnWhenNotCoordinatorError(): Unit = {
     mockCache()

core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala

@@ -735,6 +735,42 @@ class TransactionStateManagerTest {
     verifyWritesTxnMarkersInPrepareState(TransactionState.PREPARE_ABORT)
   }
 
+  @ParameterizedTest
+  @EnumSource(value = classOf[TransactionState], names = Array("PREPARE_COMMIT", "PREPARE_ABORT"))
+  def shouldCompleteEmptyPreparedTransactionOnLoad(state: TransactionState): Unit = {
+    // With transaction V2, ending a transaction with no partitions writes PrepareCommit or PrepareAbort with no
+    // partitions. On load, it must still be completed.
+    val (expectedResult, expectedCompleteState) =
+      if (state == TransactionState.PREPARE_COMMIT) (TransactionResult.COMMIT, TransactionState.COMPLETE_COMMIT)
+      else (TransactionResult.ABORT, TransactionState.COMPLETE_ABORT)
+    val txnMetadata = new TransactionMetadata(transactionalId1, producerIds(transactionalId1), RecordBatch.NO_PRODUCER_ID,
+      RecordBatch.NO_PRODUCER_ID, 1.toShort, RecordBatch.NO_PRODUCER_EPOCH, transactionTimeoutMs, state,
+      util.Set.of, time.milliseconds(), time.milliseconds(), TV_2)
+
+    txnRecords += new SimpleRecord(txnMessageKeyBytes1, TransactionLog.valueToBytes(txnMetadata.prepareNoTransit(), TV_2))
+    prepareTxnLog(topicPartition, 0, MemoryRecords.withRecords(0L, Compression.NONE, txnRecords.toArray: _*))
+
+    var txnResult: TransactionResult = null
+    var loadedMetadata: TransactionMetadata = null
+    var completeMetadata: TxnTransitMetadata = null
+    transactionManager.loadTransactionsForTxnTopicPartition(partitionId, 0, (_, result, metadata, newMetadata) => {
+      txnResult = result
+      loadedMetadata = metadata
+      completeMetadata = newMetadata
+    })
+    scheduler.tick()
+
+    assertEquals(expectedResult, txnResult)
+    assertEquals(state, loadedMetadata.state)
+    assertEquals(1.toShort, loadedMetadata.producerEpoch)
+    assertTrue(loadedMetadata.topicPartitions.isEmpty)
+    assertEquals(expectedCompleteState, completeMetadata.txnState)
+
+    loadedMetadata.completeTransitionTo(completeMetadata)
+    assertEquals(expectedCompleteState, loadedMetadata.state)
+    assertEquals(1.toShort, loadedMetadata.producerEpoch)
+  }
+
   @Test
   def shouldRemoveCompleteCommitExpiredTransactionalIds(): Unit = {
     setupAndRunTransactionalIdExpiration(Errors.NONE, TransactionState.COMPLETE_COMMIT)

transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/TransactionMetadata.java

@@ -253,7 +253,7 @@ public class TransactionMetadata {
             data.lastProducerEpoch = lastProducerEpoch;
         }
 
-        // With transaction V2, it is allowed to abort the transaction without adding any partitions. Then, the transaction
+        // With transaction V2, it is allowed to abort or commit the transaction without adding any partitions. Then, the transaction
         // start time is uncertain but it is still required. So we can use the update time as the transaction start time.
         data.txnStartTimestamp = noPartitionAdded ? updateTimestamp : txnStartTimestamp;
         data.nextProducerId = nextProducerId;
@@ -387,11 +387,11 @@ public class TransactionMetadata {
 
             case PREPARE_ABORT: // from endTxn
             case PREPARE_COMMIT:
-                // In V2, we allow state transits from Empty, CompleteCommit and CompleteAbort to PrepareAbort. It is possible
-                // their updated start time is not equal to the current start time.
-                boolean allowedEmptyAbort = toState == TransactionState.PREPARE_ABORT && transitMetadata.clientTransactionVersion().supportsEpochBump() &&
+                // In V2, we allow state transits from Empty, CompleteCommit and CompleteAbort to PrepareAbort or PrepareCommit.
+                // It is possible their updated start time is not equal to the current start time.
+                boolean allowedEmptyCompletion = transitMetadata.clientTransactionVersion().supportsEpochBump() &&
                     (state == TransactionState.EMPTY || state == TransactionState.COMPLETE_COMMIT || state == TransactionState.COMPLETE_ABORT);
-                boolean validTimestamp = txnStartTimestamp == transitMetadata.txnStartTimestamp() || allowedEmptyAbort;
+                boolean validTimestamp = txnStartTimestamp == transitMetadata.txnStartTimestamp() || allowedEmptyCompletion;
 
                 if (!validProducerEpoch(transitMetadata) ||
                     !topicPartitions.equals(transitMetadata.topicPartitions()) ||

transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/TransactionState.java

@@ -34,6 +34,7 @@ public enum TransactionState {
      * <p>
      * transition: received AddPartitionsToTxnRequest => Ongoing
      *             received AddOffsetsToTxnRequest => Ongoing
+     *             received EndTxnRequest with commit and TransactionV2 enabled => PrepareCommit
      *             received EndTxnRequest with abort and TransactionV2 enabled => PrepareAbort
      */
     EMPTY((byte) 0, org.apache.kafka.clients.admin.TransactionState.EMPTY.toString(), true),
@@ -48,7 +49,13 @@ public enum TransactionState {
     ONGOING((byte) 1, org.apache.kafka.clients.admin.TransactionState.ONGOING.toString(), false),
     /**
      * Group is preparing to commit
+     * <p>
      * transition: received acks from all partitions => CompleteCommit
+     * <p>
+     * Note, In transaction v2, we allow Empty, CompleteCommit, CompleteAbort to transition to PrepareCommit. A
+     * coordinator that has lost leadership but not yet applied the metadata change sees the transaction as having no
+     * partitions, so it commits an empty transaction, which fails with a retriable error since the write cannot be
+     * replicated, rather than returning a fatal error to the client.
      */
     PREPARE_COMMIT((byte) 2, org.apache.kafka.clients.admin.TransactionState.PREPARE_COMMIT.toString(), false),
     /**
@@ -94,7 +101,7 @@ public enum TransactionState {
     public static final Map<TransactionState, Set<TransactionState>> VALID_PREVIOUS_STATES = Map.of(
         EMPTY, Set.of(EMPTY, COMPLETE_COMMIT, COMPLETE_ABORT),
         ONGOING, Set.of(ONGOING, EMPTY, COMPLETE_COMMIT, COMPLETE_ABORT),
-        PREPARE_COMMIT, Set.of(ONGOING),
+        PREPARE_COMMIT, Set.of(ONGOING, EMPTY, COMPLETE_COMMIT, COMPLETE_ABORT),
         PREPARE_ABORT, Set.of(ONGOING, PREPARE_EPOCH_FENCE, EMPTY, COMPLETE_COMMIT, COMPLETE_ABORT),
         COMPLETE_COMMIT, Set.of(PREPARE_COMMIT),
         COMPLETE_ABORT, Set.of(PREPARE_ABORT),

transaction-coordinator/src/test/java/org/apache/kafka/coordinator/transaction/TransactionMetadataTest.java

@@ -24,6 +24,7 @@ import org.apache.kafka.server.common.TransactionVersion;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.util.HashSet;
@@ -198,6 +199,40 @@ public class TransactionMetadataTest {
         assertEquals(time.milliseconds() + 1, txnMetadata.txnStartTimestamp());
     }
 
+    @ParameterizedTest
+    @EnumSource(value = TransactionState.class, names = {"EMPTY", "COMPLETE_COMMIT", "COMPLETE_ABORT"})
+    public void testTransitFromNoOngoingTransactionToPrepareCommitAndCompleteCommitInV2(TransactionState state) {
+        short producerEpoch = 735;
+
+        TransactionMetadata txnMetadata = new TransactionMetadata(
+            transactionalId,
+            producerId,
+            RecordBatch.NO_PRODUCER_ID,
+            RecordBatch.NO_PRODUCER_ID,
+            producerEpoch,
+            RecordBatch.NO_PRODUCER_EPOCH,
+            30000,
+            state,
+            Set.of(),
+            state == TransactionState.EMPTY ? -1 : time.milliseconds() - 1,
+            time.milliseconds(),
+            TV_2
+        );
+
+        TxnTransitMetadata transitMetadata = txnMetadata.prepareAbortOrCommit(TransactionState.PREPARE_COMMIT, TV_2, RecordBatch.NO_PRODUCER_ID, time.milliseconds() + 1, true);
+        txnMetadata.completeTransitionTo(transitMetadata);
+        assertEquals(TransactionState.PREPARE_COMMIT, txnMetadata.state());
+        assertEquals(producerId, txnMetadata.producerId());
+        assertEquals(producerEpoch + 1, txnMetadata.producerEpoch());
+        assertEquals(time.milliseconds() + 1, txnMetadata.txnStartTimestamp());
+
+        transitMetadata = txnMetadata.prepareComplete(time.milliseconds() + 2);
+        txnMetadata.completeTransitionTo(transitMetadata);
+        assertEquals(TransactionState.COMPLETE_COMMIT, txnMetadata.state());
+        assertEquals(producerEpoch + 1, txnMetadata.producerEpoch());
+        assertTrue(txnMetadata.topicPartitions().isEmpty());
+    }
+
     @Test
     public void testTolerateUpdateTimeShiftDuringEpochBump() {
         short producerEpoch = 1;