diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/BufferExhaustedException.java b/clients/src/main/java/org/apache/kafka/clients/producer/BufferExhaustedException.java index f375f8f715bf3..c2bf577df779a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/BufferExhaustedException.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/BufferExhaustedException.java @@ -16,13 +16,13 @@ */ package org.apache.kafka.clients.producer; -import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.errors.TimeoutException; /** - * This exception is thrown if the producer is in non-blocking mode and the rate of data production exceeds the rate at - * which data can be sent for long enough for the alloted buffer to be exhausted. + * This exception is thrown if the rate of data production exceeds the rate at which data can be sent + * for long enough for the allotted buffer to be exhausted. */ -public class BufferExhaustedException extends KafkaException { +public class BufferExhaustedException extends TimeoutException { private static final long serialVersionUID = 1L; diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java index 54a5474b5aef7..ef1e6eeab32f1 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java @@ -424,9 +424,9 @@ public Future send(ProducerRecord record) { * * @throws InterruptException If the thread is interrupted while blocked * @throws SerializationException If the key or value are not valid objects given the configured serializers - * @throws TimeoutException If the time taken for fetching metadata or allocating memory for the record has surpassed max.block.ms. * @throws KafkaException If a Kafka related error occurs that does not belong to the public API exceptions. - * + * @throws TimeoutException if the time taken for fetching metadata for the topic has surpassed max.block.ms. + * @throws BufferExhaustedException if the time taken for allocating memory for the record has surpassed max.block.ms. */ @Override public Future send(ProducerRecord record, Callback callback) { @@ -479,6 +479,12 @@ private Future doSend(ProducerRecord record, Callback call // handling exceptions and record the errors; // for API exceptions return them in the future, // for other exceptions throw directly + } catch (BufferExhaustedException e) { + this.errors.record(); + this.metrics.sensor("buffer-exhausted-records").record(); + if (this.interceptors != null) + this.interceptors.onSendError(record, tp, e); + throw e; } catch (ApiException e) { log.debug("Exception occurred during message send:", e); if (callback != null) @@ -492,12 +498,6 @@ private Future doSend(ProducerRecord record, Callback call if (this.interceptors != null) this.interceptors.onSendError(record, tp, e); throw new InterruptException(e); - } catch (BufferExhaustedException e) { - this.errors.record(); - this.metrics.sensor("buffer-exhausted-records").record(); - if (this.interceptors != null) - this.interceptors.onSendError(record, tp, e); - throw e; } catch (KafkaException e) { this.errors.record(); if (this.interceptors != null) diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java index b42b0ec010fce..8d4cda74c3af8 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java @@ -23,8 +23,8 @@ import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; +import org.apache.kafka.clients.producer.BufferExhaustedException; import org.apache.kafka.common.MetricName; -import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Rate; @@ -138,7 +138,7 @@ public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedEx if (waitingTimeElapsed) { this.waiters.remove(moreMemory); - throw new TimeoutException("Failed to allocate memory within the configured max blocking time " + maxTimeToBlockMs + " ms."); + throw new BufferExhaustedException("Failed to allocate memory within the configured max blocking time " + maxTimeToBlockMs + " ms."); } remainingTimeToBlockNs -= timeNs; diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java index fa1e51352626b..68dfcfc19ad15 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java @@ -138,7 +138,7 @@ public double measure(MetricConfig config, long now) { metrics.addMetric(metricName, availableBytes); Sensor bufferExhaustedRecordSensor = metrics.sensor("buffer-exhausted-records"); - metricName = metrics.metricName("buffer-exhausted-rate", metricGrpName, "The average per-second number of record sends that are dropped due to buffer exhaustion"); + metricName = metrics.metricName("buffer-exhausted-rate", metricGrpName, "The average per-second number of record sends that are timed out due to buffer exhaustion"); bufferExhaustedRecordSensor.add(metricName, new Rate()); } diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java index 3756d8adb6288..00d07f67bf87f 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java @@ -16,7 +16,7 @@ */ package org.apache.kafka.clients.producer.internals; -import org.apache.kafka.common.errors.TimeoutException; +import org.apache.kafka.clients.producer.BufferExhaustedException; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.SystemTime; @@ -168,7 +168,7 @@ public void testBlockTimeout() throws Exception { try { pool.allocate(10, maxBlockTimeMs); fail("The buffer allocated more memory than its maximum value 10"); - } catch (TimeoutException e) { + } catch (BufferExhaustedException e) { // this is good } long endTimeMs = systemTime.milliseconds(); @@ -185,7 +185,7 @@ public void testCleanupMemoryAvailabilityWaiterOnBlockTimeout() throws Exception try { pool.allocate(2, maxBlockTimeMs); fail("The buffer allocated more memory than its maximum value 2"); - } catch (TimeoutException e) { + } catch (BufferExhaustedException e) { // this is good } assertTrue(pool.queued() == 0); @@ -239,7 +239,7 @@ public void run() { try { pool.allocate(2, maxBlockTimeMs); fail("The buffer allocated more memory than its maximum value 2"); - } catch (TimeoutException e) { + } catch (BufferExhaustedException e) { // this is good } catch (InterruptedException e) { // this can be neglected