address comments

Author: rreddy-22

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -647,27 +647,20 @@ private TransactionManager configureTransactionState(ProducerConfig config,
      * @throws InterruptException if the thread is interrupted while blocked
      */
     public void initTransactions() {
-        throwIfNoTransactionManager();
-        throwIfProducerClosed();
-        long now = time.nanoseconds();
-        TransactionalRequestResult result = transactionManager.initializeTransactions();
-        sender.wakeup();
-        result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
-        producerMetrics.recordInit(time.nanoseconds() - now);
-        transactionManager.maybeUpdateTransactionV2Enabled(true);
+        initTransactions(false);
     }
 
     /**
      * Initialize the transactional state for this producer, similar to {@link #initTransactions()} but
-     * with additional handling for two-phase commit (2PC). Must be called before any send operations
-     * that require a {@code transactionalId}.
+     * with additional capabilities to keep a previously prepared transaction.
+     * Must be called before any send operations that require a {@code transactionalId}.
      * <p>
      * Unlike the standard {@link #initTransactions()}, when {@code keepPreparedTxn} is set to
-     * {@code true}, the producer does <em>not</em> automatically abort existing transactions
-     * in the “prepare” phase. Instead, it enters a recovery mode allowing only finalization
-     * of those previously prepared transactions. This behavior is crucial for 2PC scenarios,
-     * where transactions should remain intact until the external transaction manager decides
-     * whether to commit or abort.
+     * {@code true}, the producer does <em>not</em> automatically abort existing transactions.
+     * Instead, it enters a recovery mode allowing only finalization of those previously prepared transactions.
+     *
+     * This behavior is especially crucial for 2PC scenarios, where transactions should remain intact
+     * until the external transaction manager decides whether to commit or abort.
      * <p>
      * When {@code keepPreparedTxn} is {@code false}, this behaves like the normal transactional
      * initialization, aborting any unfinished transactions and resetting the producer for

clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java

@@ -141,23 +141,6 @@ public MockProducer() {
         this(Cluster.empty(), false, null, null, null);
     }
 
-    @Override
-    public void initTransactions() {
-        verifyNotClosed();
-        verifyNotFenced();
-        if (this.transactionInitialized) {
-            throw new IllegalStateException("MockProducer has already been initialized for transactions.");
-        }
-        if (this.initTransactionException != null) {
-            throw this.initTransactionException;
-        }
-        this.transactionInitialized = true;
-        this.transactionInFlight = false;
-        this.transactionCommitted = false;
-        this.transactionAborted = false;
-        this.sentOffsets = false;
-    }
-
     @Override
     public void initTransactions(boolean keepPreparedTxn) {
         verifyNotClosed();

clients/src/main/java/org/apache/kafka/clients/producer/Producer.java

@@ -42,7 +42,9 @@ public interface Producer<K, V> extends Closeable {
     /**
      * See {@link KafkaProducer#initTransactions()}
      */
-    void initTransactions();
+    default void initTransactions() {
+        initTransactions(false);
+    }
 
     /**
      * See {@link KafkaProducer#initTransactions(boolean)}