Repository: apache/kafka
File: core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala (Scala)

Code before:
              val nextProducerIdOrErrors =
                if (!isEpochFence && txnMetadata.isProducerEpochExhausted) {
                  try {
                    Right(producerIdManager.generateProducerId())
                  } catch {
                    case e: Exception => Left(Errors.forException(e))
                  }
                } else {
                  Right(RecordBatch.NO_PRODUCER_ID)
                }

              if (nextState == TransactionState.PREPARE_ABORT && isEpochFence) {
                // We should clear the pending state to make way for the transition to PrepareAbort and also bump
                // the epoch in the transaction metadata we are about to append.
                txnMetadata.pendingState = None
                txnMetadata.producerEpoch = producerEpoch
                txnMetadata.lastProducerEpoch = RecordBatch.NO_PRODUCER_EPOCH
              }

              nextProducerIdOrErrors.flatMap {
                nextProducerId =>
                  // For epoch fence case, we don't want to bump the epoch again in prepareAbortOrCommit
                  // since it was already bumped during the fencing phase to avoid double increment
                  val transactionVersionForCall = if (isEpochFence) {
                    // Use TV_0 to prevent epoch bump since it was already bumped during fencing
                    TransactionVersion.TV_0
                  } else {
                    clientTransactionVersion
                  }
                  
                  Right(coordinatorEpoch, txnMetadata.prepareAbortOrCommit(nextState, transactionVersionForCall, nextProducerId.asInstanceOf[Long], time.milliseconds(), noPartitionAdded))
              }
            }

            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
              // new producer ID that is still only in pending state.
              Left(Errors.CONCURRENT_TRANSACTIONS)
            } else if (txnMetadata.producerId != producerId && !retryOnOverflow)
              Left(Errors.INVALID_PRODUCER_ID_MAPPING)
            else if (!isValidEpoch)
              Left(Errors.PRODUCER_FENCED)
            else txnMetadata.state match {
              case TransactionState.ONGOING =>
                val nextState = if (txnMarkerResult == TransactionResult.COMMIT)
                  TransactionState.PREPARE_COMMIT
                else
                  TransactionState.PREPARE_ABORT

                generateTxnTransitMetadataForTxnCompletion(nextState, false)

Review comment (bug):
Instead of doing this, let's just do something at line 825 -- the TV will move forward with KIP-939 and using old version could lead to subtle bugs over time.

Code after:
              val nextProducerIdOrErrors =
                if (!isEpochFence && txnMetadata.isProducerEpochExhausted) {
                  try {
                    Right(producerIdManager.generateProducerId())
                  } catch {
                    case e: Exception => Left(Errors.forException(e))
                  }
                } else {
                  Right(RecordBatch.NO_PRODUCER_ID)
                }

              if (nextState == TransactionState.PREPARE_ABORT && isEpochFence) {
                // We should clear the pending state to make way for the transition to PrepareAbort
                txnMetadata.pendingState = None
                // For TV2+, don't manually set the epoch - let prepareAbortOrCommit handle it naturally.
              }

              nextProducerIdOrErrors.flatMap {
                nextProducerId =>
                  Right(coordinatorEpoch, txnMetadata.prepareAbortOrCommit(nextState, clientTransactionVersion, nextProducerId.asInstanceOf[Long], time.milliseconds(), noPartitionAdded))
              }
            }

            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
              // new producer ID that is still only in pending state.
              Left(Errors.CONCURRENT_TRANSACTIONS)
            } else if (txnMetadata.producerId != producerId && !retryOnOverflow)
              Left(Errors.INVALID_PRODUCER_ID_MAPPING)
            else if (!isValidEpoch)
              Left(Errors.PRODUCER_FENCED)
            else txnMetadata.state match {
              case TransactionState.ONGOING =>
                val nextState = if (txnMarkerResult == TransactionResult.COMMIT)
                  TransactionState.PREPARE_COMMIT
                else
                  TransactionState.PREPARE_ABORT

                generateTxnTransitMetadataForTxnCompletion(nextState, false)
              case TransactionState.COMPLETE_COMMIT =>
                if (txnMarkerResult == TransactionResult.COMMIT) {
                  if (isRetry)
                    Left(Errors.NONE)
                  else
                    logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult)
                } else {
                  // Abort.
                  if (isRetry)
                    logInvalidStateTransitionAndReturnError(transactionalId, txnMetadata.state, txnMarkerResult)
                  else