artemlivshits commented on code in PR #19751:
URL: https://github.com/apache/kafka/pull/19751#discussion_r2094234325
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala:
##########
@@ -772,135 +793,159 @@ class TransactionCoordinator(txnConfig:
TransactionConfig,
val coordinatorEpoch = epochAndTxnMetadata.coordinatorEpoch
txnMetadata.inLock {
- producerIdCopy = txnMetadata.producerId
- producerEpochCopy = txnMetadata.producerEpoch
- // PrepareEpochFence has slightly different epoch bumping logic so
don't include it here.
- // Note that, it can only happen when the current state is Ongoing.
- isEpochFence =
txnMetadata.pendingState.contains(TransactionState.PREPARE_EPOCH_FENCE)
- // True if the client retried a request that had overflowed the
epoch, and a new producer ID is stored in the txnMetadata
- val retryOnOverflow = !isEpochFence && txnMetadata.prevProducerId
== producerId &&
- producerEpoch == Short.MaxValue - 1 && txnMetadata.producerEpoch
== 0
- // True if the client retried an endTxn request, and the bumped
producer epoch is stored in the txnMetadata.
- val retryOnEpochBump = !isEpochFence && txnMetadata.producerEpoch
== producerEpoch + 1
-
- val isValidEpoch = {
- if (!isEpochFence) {
- // With transactions V2, state + same epoch is not sufficient
to determine if a retry transition is valid. If the epoch is the
- // same it actually indicates the next endTransaction call.
Instead, we want to check the epoch matches with the epoch in the retry
conditions.
- // Return producer fenced even in the cases where the epoch is
higher and could indicate an invalid state transition.
- // Use the following criteria to determine if a v2 retry is
valid:
- txnMetadata.state match {
- case TransactionState.ONGOING | TransactionState.EMPTY |
TransactionState.DEAD | TransactionState.PREPARE_EPOCH_FENCE =>
- producerEpoch == txnMetadata.producerEpoch
- case TransactionState.PREPARE_COMMIT |
TransactionState.PREPARE_ABORT =>
- retryOnEpochBump
- case TransactionState.COMPLETE_COMMIT |
TransactionState.COMPLETE_ABORT =>
- retryOnEpochBump || retryOnOverflow || producerEpoch ==
txnMetadata.producerEpoch
+ // These checks are performed first so that we don't have to deal
with intermediate states
+ // (including PrepareCommit and PrepareAbort) in already complex
state machine.
+ if (txnMetadata.pendingTransitionInProgress &&
!txnMetadata.pendingState.contains(TransactionState.PREPARE_EPOCH_FENCE)) {
+ Left(Errors.CONCURRENT_TRANSACTIONS)
+ } else if (txnMetadata.state == TransactionState.PREPARE_COMMIT ||
txnMetadata.state == TransactionState.PREPARE_ABORT) {
Review Comment:
Here we eagerly reject requests if the transaction is in PrepareCommit /
PrepareAbort state. Previously the code tried to validate the state and return
fenced error on mismatch, but in the end if the checks were successful the
request got rejected with concurrent transactions error. The checks grew
extremely complex with the new nextProduceEpoch state, so we avoid this
complexity altogether by just returning concurrent transactions error for this
transient state.
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -109,7 +125,7 @@ private[transaction] class TransactionMetadata(val
transactionalId: String,
def prepareIncrementProducerEpoch(newTxnTimeoutMs: Int,
expectedProducerEpoch: Option[Short],
updateTimestamp: Long): Either[Errors,
TxnTransitMetadata] = {
- if (isProducerEpochExhausted)
+ if (TransactionMetadata.isEpochExhausted(producerEpoch))
Review Comment:
This changed because `isProducerEpochExhausted` used `producerEpoch`, but
now it uses `clientProducerEpoch` (which sometimes could be
`nextProducerEpoch`), so in the place we want to continue checking
`producerEpoch`, thus this is an explicit call.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
