KAFKA-17019 Producer TimeoutException should include root cause (#20159)

### Changes
- Add new Exception class `PotentialCauseException`.
- All `org.apache.kafka.common.errors.TimeoutException` in
`KafkaProducer` has `PotentialCauseException` as  root cause if it
cannot catch any exception.

### Describe
`TimeoutException` can be thrown for various reasons.
However, it is often difficult to identify the root cause,
Because there are so many potential factors that can lead to a
`TimeoutException`.

For example:
1. The `ProducerClient` might be busy, so it may not be able to send the
request in time. As a result, some batches may expire, leading to a
`TimeoutException`.
2. The `broker` might be unavailable due to network issues or internal
failures.
3. A request may be in flight, and although the broker successfully
handles and responds to it, the response might arrive slightly late.

As shown above, there are many possible causes.  In some cases, no
`exception` is caught in the `catch` block, and a `TimeoutException` is
thrown simply by comparing the `elapsed time`.  However, the developer
using `TimeoutException` in `KafkaProducer` likely already knows which
specific reasons could cause it in that context.  Therefore, I think it
would be helpful to include a `PotentialCauseException` that reflects
the likely reason, based on the developer’s knowledge.

Reviewers: TengYao Chi <kitingiao@gmail.com>, Yung
 <yungyung7654321@gmail.com>, Andrew Schofield
 <aschofield@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

Author: chickenchickenlove

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

@@ -248,6 +248,18 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
     public static final String NETWORK_THREAD_PREFIX = "kafka-producer-network-thread";
     public static final String PRODUCER_METRIC_GROUP_NAME = "producer-metrics";
 
+    private static final String INIT_TXN_TIMEOUT_MSG = "InitTransactions timed out — " +
+            "did not complete coordinator discovery or " +
+            "receive the InitProducerId response within max.block.ms.";
+
+    private static final String SEND_OFFSETS_TIMEOUT_MSG =
+            "SendOffsetsToTransaction timed out – did not reach the coordinator or " +
+                    "receive the TxnOffsetCommit/AddOffsetsToTxn response within max.block.ms";
+    private static final String COMMIT_TXN_TIMEOUT_MSG =
+            "CommitTransaction timed out – did not complete EndTxn with the transaction coordinator within max.block.ms";
+    private static final String ABORT_TXN_TIMEOUT_MSG =
+            "AbortTransaction timed out – did not complete EndTxn(abort) with the transaction coordinator within max.block.ms";
+    
     private final String clientId;
     // Visible for testing
     final Metrics metrics;
@@ -672,7 +684,7 @@ public void initTransactions(boolean keepPreparedTxn) {
         long now = time.nanoseconds();
         TransactionalRequestResult result = transactionManager.initializeTransactions(keepPreparedTxn);
         sender.wakeup();
-        result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
+        result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS, INIT_TXN_TIMEOUT_MSG);
         producerMetrics.recordInit(time.nanoseconds() - now);
         transactionManager.maybeUpdateTransactionV2Enabled(true);
     }
@@ -761,7 +773,7 @@ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offs
             long start = time.nanoseconds();
             TransactionalRequestResult result = transactionManager.sendOffsetsToTransaction(offsets, groupMetadata);
             sender.wakeup();
-            result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
+            result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS, SEND_OFFSETS_TIMEOUT_MSG);
             producerMetrics.recordSendOffsets(time.nanoseconds() - start);
         }
     }
@@ -847,7 +859,7 @@ public void commitTransaction() throws ProducerFencedException {
         long commitStart = time.nanoseconds();
         TransactionalRequestResult result = transactionManager.beginCommit();
         sender.wakeup();
-        result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
+        result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS, COMMIT_TXN_TIMEOUT_MSG);
         producerMetrics.recordCommitTxn(time.nanoseconds() - commitStart);
     }
 
@@ -882,7 +894,7 @@ public void abortTransaction() throws ProducerFencedException {
         long abortStart = time.nanoseconds();
         TransactionalRequestResult result = transactionManager.beginAbort();
         sender.wakeup();
-        result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
+        result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS, ABORT_TXN_TIMEOUT_MSG);
         producerMetrics.recordAbortTxn(time.nanoseconds() - abortStart);
     }
 

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

@@ -424,7 +424,7 @@ public Uuid clientInstanceId(Duration timeout) {
             if (injectTimeoutExceptionCounter > 0) {
                 --injectTimeoutExceptionCounter;
             }
-            throw new TimeoutException();
+            throw new TimeoutException("TimeoutExceptions are successfully injected for test.");
         }
 
         return clientInstanceId;

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

@@ -414,7 +414,8 @@ private long sendProducerData(long now) {
             log.trace("Expired {} batches in accumulator", expiredBatches.size());
         for (ProducerBatch expiredBatch : expiredBatches) {
             String errorMessage = "Expiring " + expiredBatch.recordCount + " record(s) for " + expiredBatch.topicPartition
-                + ":" + (now - expiredBatch.createdMs) + " ms has passed since batch creation";
+                + ":" + (now - expiredBatch.createdMs) + " ms has passed since batch creation. " 
+                + "The request has not been sent, or no server response has been received yet.";
             failBatch(expiredBatch, new TimeoutException(errorMessage), false);
             if (transactionManager != null && expiredBatch.inRetry()) {
                 // This ensures that no new batches are drained until the current in flight batches are fully resolved.

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

@@ -47,16 +47,12 @@ public void done() {
         this.latch.countDown();
     }
 
-    public void await() {
-        this.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
-    }
-
-    public void await(long timeout, TimeUnit unit) {
+    public void await(long timeout, TimeUnit unit, String expectedTimeoutReason) {
         try {
             boolean success = latch.await(timeout, unit);
             if (!success) {
                 throw new TimeoutException("Timeout expired after " + unit.toMillis(timeout) +
-                    "ms while awaiting " + operation);
+                    "ms while awaiting " + operation + ". " + expectedTimeoutReason);
             }
 
             isAcked = true;

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

@@ -174,6 +174,12 @@
 import static org.mockito.Mockito.when;
 
 public class KafkaProducerTest {
+
+    private static final String INIT_TXN_TIMEOUT_MSG =
+            "InitTransactions timed out — " +
+                    "did not complete coordinator discovery or " +
+                    "receive the InitProducerId response within max.block.ms.";
+    
     private final String topic = "topic";
     private final Collection<Node> nodes = Collections.singletonList(NODE);
     private final Cluster emptyCluster = new Cluster(
@@ -1322,7 +1328,7 @@ public void testInitTransactionsResponseAfterTimeout() throws Exception {
                 "Timed out while waiting for expected `InitProducerId` request to be sent");
 
             time.sleep(maxBlockMs);
-            TestUtils.assertFutureThrows(TimeoutException.class, future);
+            TestUtils.assertFutureThrowsWithMessageContaining(TimeoutException.class, future, INIT_TXN_TIMEOUT_MSG);
 
             client.respond(initProducerIdResponse(1L, (short) 5, Errors.NONE));
 
@@ -1352,7 +1358,8 @@ public void testInitTransactionTimeout() {
                     ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(),
                 FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", NODE));
 
-            assertThrows(TimeoutException.class, producer::initTransactions);
+            var timeoutEx = assertThrows(TimeoutException.class, producer::initTransactions);
+            assertTrue(timeoutEx.getMessage().contains(INIT_TXN_TIMEOUT_MSG));
 
             client.prepareResponse(
                 request -> request instanceof FindCoordinatorRequest &&
@@ -2364,7 +2371,8 @@ public void testOnlyCanExecuteCloseAfterInitTransactionsTimeout() {
 
         Producer<String, String> producer = kafkaProducer(configs, new StringSerializer(), new StringSerializer(),
                 metadata, client, null, time);
-        assertThrows(TimeoutException.class, producer::initTransactions);
+        var timeoutEx1 = assertThrows(TimeoutException.class, producer::initTransactions);
+        assertTrue(timeoutEx1.getMessage().contains(INIT_TXN_TIMEOUT_MSG));
         // other transactional operations should not be allowed if we catch the error after initTransactions failed
         try {
             assertThrows(IllegalStateException.class, producer::beginTransaction);

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

@@ -163,6 +163,7 @@ public class SenderTest {
             TOPIC_NAME, TOPIC_ID,
             "testSplitBatchAndSend", Uuid.fromString("2J9hK8m1wHMKjXfIkQyXx1")
     );
+    private static final String SENDER_TIMEOUT_MSG = "The request has not been sent, or no server response has been received yet.";
     private final TopicPartition tp0 = new TopicPartition(TOPIC_NAME, 0);
     private final TopicPartition tp1 = new TopicPartition(TOPIC_NAME, 1);
     private final TopicPartition tp2 = new TopicPartition(TOPIC_NAME, 2);
@@ -425,6 +426,7 @@ public void setPartition(int partition) {
             @Override
             public void onCompletion(RecordMetadata metadata, Exception exception) {
                 if (exception instanceof TimeoutException) {
+                    assertTrue(exception.getMessage().contains(SENDER_TIMEOUT_MSG));
                     expiryCallbackCount.incrementAndGet();
                     try {
                         accumulator.append(tp1.topic(), tp1.partition(), 0L, key, value,
@@ -2792,7 +2794,7 @@ public void testRecordsFlushedImmediatelyOnTransactionCompletion() throws Except
             runUntil(sender, txnManager::isReady);
 
             assertTrue(commitResult.isSuccessful());
-            commitResult.await();
+            commitResult.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS, "Unexpected Timed out for transaction commit to completed during the test.");
 
             // Finally, we want to assert that the linger time is still effective
             // when the new transaction begins.
@@ -2942,8 +2944,9 @@ public void testForceShutdownWithIncompleteTransaction() {
 
             sender.forceClose();
             sender.run();
-            assertThrows(KafkaException.class, commitResult::await,
-                "The test expected to throw a KafkaException for forcefully closing the sender");
+            assertThrows(KafkaException.class, () -> commitResult.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS,
+                "The test expected to throw a KafkaException for forcefully closing the sender")
+            );
         } finally {
             m.close();
         }
@@ -3154,7 +3157,7 @@ public void testReceiveFailedBatchTwiceWithTransactions() throws Exception {
         assertTrue(txnManager::isReady);
 
         assertTrue(result.isSuccessful());
-        result.await();
+        result.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS, "Unexpected time out during the test.");
 
         txnManager.beginTransaction();
     }
@@ -3193,7 +3196,7 @@ public void testInvalidTxnStateIsAnAbortableError() throws Exception {
         assertTrue(txnManager::isReady);
 
         assertTrue(result.isSuccessful());
-        result.await();
+        result.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS, "Unexpected time out during the test.");
 
         txnManager.beginTransaction();
     }
@@ -3232,7 +3235,7 @@ public void testTransactionAbortableExceptionIsAnAbortableError() throws Excepti
         assertTrue(txnManager::isReady);
 
         assertTrue(result.isSuccessful());
-        result.await();
+        result.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS, "Unexpected time out during the test.");
 
         txnManager.beginTransaction();
     }
@@ -3264,7 +3267,7 @@ public void testAbortableErrorIsConvertedToFatalErrorDuringAbort() throws Except
         TransactionalRequestResult commitResult = transactionManager.beginCommit();
         sender.runOnce();
         try {
-            commitResult.await(1000, TimeUnit.MILLISECONDS);
+            commitResult.await(1000, TimeUnit.MILLISECONDS, "Unexpected time out during the test.");
             fail("Expected abortable error to be thrown for commit");
         } catch (KafkaException e) {
             assertTrue(transactionManager.hasAbortableError());
@@ -3281,7 +3284,7 @@ public void testAbortableErrorIsConvertedToFatalErrorDuringAbort() throws Except
 
         // Verify the error is converted to KafkaException (not TransactionAbortableException)
         try {
-            abortResult.await(1000, TimeUnit.MILLISECONDS);
+            abortResult.await(1000, TimeUnit.MILLISECONDS, "Unexpected time out during the test.");
             fail("Expected KafkaException to be thrown");
         } catch (KafkaException e) {
             // Verify TM is in FATAL_ERROR state
@@ -3884,7 +3887,7 @@ private void doInitTransactions(TransactionManager transactionManager, ProducerI
         prepareInitProducerResponse(Errors.NONE, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch);
         sender.runOnce();
         assertTrue(transactionManager.hasProducerId());
-        result.await();
+        result.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS, "Unexpected time out during the test.");
     }
 
     private void prepareFindCoordinatorResponse(Errors error, String txnid) {