jolshan commented on code in PR #19751:
URL: https://github.com/apache/kafka/pull/19751#discussion_r2114881725


##########
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:
   To clarify for myself:
   This method would not be called from keepPrepared=True initProducerId calls, 
and thus we should update the ongoing transaction's producer epoch and not the 
client (in some cases nextProducerEpoch) epoch.
   
   If we call this method, how does that affect nextProducerEpoch/Id? I could 
imagine we bump the epoch here such that it matches the nextProducerEpoch. 



-- 
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

Reply via email to