Addressing PR review.

Author: chickenchickenlove

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

@@ -673,7 +673,7 @@ public void initTransactions(boolean keepPreparedTxn) {
         TransactionalRequestResult result = transactionManager.initializeTransactions(keepPreparedTxn);
         sender.wakeup();
         result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS,
-                     new PotentialCauseException("InitTransactions timed out – could not discover the transaction coordinator or receive the InitProducerId response within max.block.ms (broker unavailable, network lag, or ACL denial)."));
+                     () -> new PotentialCauseException("InitTransactions timed out – could not discover the transaction coordinator or receive the InitProducerId response within max.block.ms (broker unavailable, network lag, or ACL denial)."));
         producerMetrics.recordInit(time.nanoseconds() - now);
         transactionManager.maybeUpdateTransactionV2Enabled(true);
     }
@@ -763,7 +763,7 @@ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offs
             TransactionalRequestResult result = transactionManager.sendOffsetsToTransaction(offsets, groupMetadata);
             sender.wakeup();
             result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS,
-                         new PotentialCauseException("SendOffsetsToTransaction timed out – unable to reach the consumer-group or transaction coordinator or to receive the TxnOffsetCommit/AddOffsetsToTxn response within max.block.ms (coordinator unavailable, rebalance in progress, network/ACL issue)."));
+                         () -> new PotentialCauseException("SendOffsetsToTransaction timed out – unable to reach the consumer-group or transaction coordinator or to receive the TxnOffsetCommit/AddOffsetsToTxn response within max.block.ms (coordinator unavailable, rebalance in progress, network/ACL issue)."));
             producerMetrics.recordSendOffsets(time.nanoseconds() - start);
         }
     }
@@ -849,7 +849,7 @@ public void commitTransaction() throws ProducerFencedException {
         TransactionalRequestResult result = transactionManager.beginCommit();
         sender.wakeup();
         result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS,
-                     new PotentialCauseException("CommitTransaction timed out – failed to complete EndTxn with the transaction coordinator within max.block.ms (coordinator unavailable, network lag, ACL/rebalance)."));
+                     () -> new PotentialCauseException("CommitTransaction timed out – failed to complete EndTxn with the transaction coordinator within max.block.ms (coordinator unavailable, network lag, ACL/rebalance)."));
         producerMetrics.recordCommitTxn(time.nanoseconds() - commitStart);
     }
 
@@ -885,7 +885,7 @@ public void abortTransaction() throws ProducerFencedException {
         TransactionalRequestResult result = transactionManager.beginAbort();
         sender.wakeup();
         result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS,
-                     new PotentialCauseException("AbortTransaction timed out – could not complete EndTxn(abort) with the transaction coordinator within max.block.ms (coordinator unavailable/rebalancing, network lag, or ACL denial)."));
+                     () -> new PotentialCauseException("AbortTransaction timed out – could not complete EndTxn(abort) with the transaction coordinator within max.block.ms (coordinator unavailable/rebalancing, network lag, or ACL denial)."));
         producerMetrics.recordAbortTxn(time.nanoseconds() - abortStart);
     }
 

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionalRequestResult.java

@@ -23,6 +23,7 @@
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 
 public final class TransactionalRequestResult {
     private final CountDownLatch latch;
@@ -56,16 +57,16 @@ public void await(long timeout, TimeUnit unit) {
         this.await(timeout, unit, null);
     }
 
-    public void await(long timeout, TimeUnit unit, PotentialCauseException potentialCauseException) {
+    public void await(long timeout, TimeUnit unit, Supplier<PotentialCauseException> potentialCauseException) {
         try {
             boolean success = latch.await(timeout, unit);
             if (!success) {
                 if (potentialCauseException == null) {
                     throw new TimeoutException("Timeout expired after " + unit.toMillis(timeout) +
-                                               "ms while awaiting " + operation);
+                                               "ms while awaiting " + operation, new PotentialCauseException("Unknown reason."));
                 } else {
                     throw new TimeoutException("Timeout expired after " + unit.toMillis(timeout) +
-                                               "ms while awaiting " + operation, potentialCauseException);
+                                               "ms while awaiting " + operation, potentialCauseException.get());
                 }
             }