Addressing PR review.

Author: chickenchickenlove

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

@@ -51,7 +51,6 @@
 import org.apache.kafka.common.errors.InterruptException;
 import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.InvalidTxnStateException;
-import org.apache.kafka.common.errors.PotentialCauseException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.apache.kafka.common.errors.RecordTooLargeException;
 import org.apache.kafka.common.errors.RetriableException;
@@ -673,7 +672,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 KafkaException("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 +762,9 @@ 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 KafkaException("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 +850,8 @@ 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 KafkaException("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 +887,9 @@ 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 KafkaException("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);
     }
 
@@ -1232,7 +1236,7 @@ private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, long
                 }
                 if (ex.getCause() != null)
                     throw new TimeoutException(errorMessage, ex.getCause());
-                throw new TimeoutException(errorMessage, new PotentialCauseException("Metadata update timed out ― topic missing, auth denied, broker/partition unavailable, or client sender/buffer stalled."));
+                throw new TimeoutException(errorMessage, new KafkaException("Metadata update timed out ― topic missing, auth denied, broker/partition unavailable, or client sender/buffer stalled."));
             }
             cluster = metadata.fetch();
             elapsed = time.milliseconds() - nowMs;
@@ -1241,7 +1245,7 @@ private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, long
                 if (metadata.getError(topic) != null && metadata.getError(topic).exception() instanceof RetriableException) {
                     throw new TimeoutException(errorMessage, metadata.getError(topic).exception());
                 }
-                throw new TimeoutException(errorMessage, new PotentialCauseException("Metadata update timed out ― topic missing, auth denied, broker/partition unavailable, or client sender/buffer stalled."));
+                throw new TimeoutException(errorMessage, new KafkaException("Metadata update timed out ― topic missing, auth denied, broker/partition unavailable, or client sender/buffer stalled."));
             }
             metadata.maybeThrowExceptionForTopic(topic);
             remainingWaitMs = maxWaitMs - elapsed;

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

@@ -27,7 +27,6 @@
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.errors.PotentialCauseException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.header.internals.RecordHeaders;
@@ -425,7 +424,7 @@ public Uuid clientInstanceId(Duration timeout) {
             if (injectTimeoutExceptionCounter > 0) {
                 --injectTimeoutExceptionCounter;
             }
-            throw new TimeoutException(new PotentialCauseException("TimeoutExceptions are successfully injected for test."));
+            throw new TimeoutException(new KafkaException("TimeoutExceptions are successfully injected for test."));
         }
 
         return clientInstanceId;

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

@@ -34,7 +34,6 @@
 import org.apache.kafka.common.errors.FencedLeaderEpochException;
 import org.apache.kafka.common.errors.InvalidMetadataException;
 import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
-import org.apache.kafka.common.errors.PotentialCauseException;
 import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
@@ -416,7 +415,9 @@ private long sendProducerData(long now) {
         for (ProducerBatch expiredBatch : expiredBatches) {
             String errorMessage = "Expiring " + expiredBatch.recordCount + " record(s) for " + expiredBatch.topicPartition
                 + ":" + (now - expiredBatch.createdMs) + " ms has passed since batch creation";
-            failBatch(expiredBatch, new TimeoutException(new PotentialCauseException(errorMessage)), false);
+            KafkaException potentialCause = new KafkaException(
+                    "The broker might be unavailable or responding slowly, or the CPU might be busy.");
+            failBatch(expiredBatch, new TimeoutException(errorMessage, potentialCause), false);
             if (transactionManager != null && expiredBatch.inRetry()) {
                 // This ensures that no new batches are drained until the current in flight batches are fully resolved.
                 transactionManager.markSequenceUnresolved(expiredBatch);

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

@@ -17,8 +17,8 @@
 package org.apache.kafka.clients.producer.internals;
 
 
+import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.errors.InterruptException;
-import org.apache.kafka.common.errors.PotentialCauseException;
 import org.apache.kafka.common.errors.TimeoutException;
 
 import java.util.concurrent.CountDownLatch;
@@ -50,19 +50,19 @@ public void done() {
     }
 
     public void await() {
-        this.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS, () -> new PotentialCauseException("Unknown reason."));
+        this.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS, () -> new KafkaException("Unknown reason."));
     }
 
     public void await(long timeout, TimeUnit unit) {
-        this.await(timeout, unit, () -> new PotentialCauseException("Unknown reason."));
+        this.await(timeout, unit, () -> new KafkaException("Unknown reason."));
     }
 
-    public void await(long timeout, TimeUnit unit, Supplier<PotentialCauseException> potentialCauseException) {
+    public void await(long timeout, TimeUnit unit, Supplier<KafkaException> potentialCauseException) {
         try {
             boolean success = latch.await(timeout, unit);
             if (!success) {
                 throw new TimeoutException("Timeout expired after " + unit.toMillis(timeout) +
-                                           "ms while awaiting " + operation, potentialCauseException.get());
+                    "ms while awaiting " + operation, potentialCauseException.get());
             }
 
             isAcked = true;