CalvinConfluent commented on code in PR #17698:
URL: https://github.com/apache/kafka/pull/17698#discussion_r1854559934
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala:
##########
@@ -564,41 +633,44 @@ class TransactionCoordinator(txnConfig: TransactionConfig,
else
PrepareAbort
- // Maybe allocate new producer ID if we are bumping epoch and
epoch is exhausted
- val nextProducerIdOrErrors =
- if (clientTransactionVersion.supportsEpochBump() &&
!txnMetadata.pendingState.contains(PrepareEpochFence) &&
txnMetadata.isProducerEpochExhausted) {
- try {
- Right(producerIdManager.generateProducerId())
- } catch {
- case e: Exception => Left(Errors.forException(e))
- }
+ generateTxnTransitMetadataForTxnCompletion(nextState)
Review Comment:
I refactored the code a bit. Now we distinguish the normal bump and empty
abort.
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMetadata.scala:
##########
@@ -487,13 +491,17 @@ private[transaction] class TransactionMetadata(val
transactionalId: String,
producerEpoch = transitMetadata.producerEpoch
lastProducerEpoch = transitMetadata.lastProducerEpoch
nextProducerId = transitMetadata.nextProducerId
+ if (toState == PrepareAbort && (state == Empty || state ==
CompleteCommit || state == CompleteAbort)) {
+ // In V2, we allow state transits from Empty, CompleteCommit and
CompleteAbort to PrepareAbort. Let's
+ // use the last update time as the txn start time.
+ txnStartTimestamp = transitMetadata.txnLastUpdateTimestamp
Review Comment:
Refactored.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
