[PR] KAFKA-20739: Transaction coordinator accepts InitProducerId v6 keepPreparedTxn recovery [kafka]

["Shekharrajak (via GitHub)" <gi...@apache.org>]

Shekharrajak opened a new pull request, #22704:
URL: https://github.com/apache/kafka/pull/22704

   Ref https://issues.apache.org/jira/browse/KAFKA-20739  
   
   The transaction coordinator rejected every InitProducerId keepPreparedTxn=true with UNSUPPORTED_VERSION (a stub), so a 2PC producer could not recover a prepared transaction after a crash/restart.
    
   
   - TransactionCoordinator.handleInitProducerId: replace the stub — reject only keepPreparedTxn && !enable2Pc (INVALID_REQUEST); otherwise proceed.
   - Add a recovery short-circuit: when an ongoing 2PC transaction exists (ONGOING + isDistributedTwoPhaseCommitTxn), preserve it and return its producerId/epoch as ongoingTxnProducerId/ongoingTxnProducerEpoch 
    
   - KafkaApis.createResponse: populate the v6 ongoing response fields.
   
   
   Client side already sends v6 (KAFKA-20738) and consumes these response fields (TransactionManager → PREPARED_TRANSACTION), so the two complete the recovery loop together.
    
   Tests: TransactionCoordinatorTest — keepPreparedTxn without 2PC → INVALID_REQUEST; with 2PC + no ongoing txn → NONE; with an ongoing 2PC txn → returns its pid/epoch and asserts no  log append.
   


-- 
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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-20739: Transaction coordinator accepts InitProducerId v6 keepPreparedTxn recovery [kafka]

["Shekharrajak (via GitHub)" <gi...@apache.org>]

Shekharrajak commented on code in PR #22704:
URL: https://github.com/apache/kafka/pull/22704#discussion_r3493784606


##########
core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala:
##########
@@ -171,6 +170,23 @@ class TransactionCoordinator(txnConfig: TransactionConfig,
         case Some(epochAndTxnMetadata) => Right(epochAndTxnMetadata)
       }
 
+      // 2PC recovery: if the client wants to keep a prepared transaction and an ongoing 2PC
+      // transaction exists, preserve it and return its producer id/epoch so the client can
+      // finalize it. There is no state change, so respond directly without a log append.
+      val preservedResult = coordinatorEpochAndMetadata.toOption.flatMap { existingEpochAndMetadata =>
+        val txnMetadata = existingEpochAndMetadata.transactionMetadata

Review Comment:
   The epoch fence is skipped on the preserve path —
   the prepared txn's epoch must stay valid for completeTransaction.



-- 
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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org