From fb66dcf18b8d88c0cf3001f682efe5e0964d0dc0 Mon Sep 17 00:00:00 2001 From: MayureshGharat Date: Fri, 20 May 2016 11:44:31 -0700 Subject: [PATCH 1/3] Deprecated BufferExhaustedException and also removed its use and the related sensor metric --- .../kafka/clients/producer/BufferExhaustedException.java | 4 +++- .../org/apache/kafka/clients/producer/KafkaProducer.java | 6 ------ .../kafka/clients/producer/internals/RecordAccumulator.java | 4 ---- 3 files changed, 3 insertions(+), 11 deletions(-) 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..b8931f90c629a 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 @@ -19,9 +19,11 @@ import org.apache.kafka.common.KafkaException; /** + * @deprecated This exception class is deprecated and is no longer used. It will be removed in future releases. * 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. + * which data can be sent for long enough for the allocated buffer to be exhausted. */ +@Deprecated public class BufferExhaustedException extends KafkaException { 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..bcf66f610e5b9 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 @@ -492,12 +492,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/RecordAccumulator.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java index fa1e51352626b..3a0edd140f680 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 @@ -136,10 +136,6 @@ 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"); - bufferExhaustedRecordSensor.add(metricName, new Rate()); } /** From ab879b4cbfe026d1f8426be805b194c6298a817e Mon Sep 17 00:00:00 2001 From: MayureshGharat Date: Thu, 9 Jun 2016 18:28:13 -0700 Subject: [PATCH 2/3] Update the buffer-exhausted-records sensor when a timeoutException is thrown on BufferExhaustion --- .../org/apache/kafka/clients/producer/KafkaProducer.java | 6 ++++++ .../kafka/clients/producer/internals/RecordAccumulator.java | 4 ++++ 2 files changed, 10 insertions(+) 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 bcf66f610e5b9..288ba1399c5e8 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 @@ -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 (TimeoutException 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) 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 3a0edd140f680..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 @@ -136,6 +136,10 @@ 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 timed out due to buffer exhaustion"); + bufferExhaustedRecordSensor.add(metricName, new Rate()); } /** From d137aff1b657329bef49f8cf626a45755b03b4b3 Mon Sep 17 00:00:00 2001 From: MayureshGharat Date: Mon, 21 Nov 2016 14:58:59 -0800 Subject: [PATCH 3/3] Made BufferExhaustException a subclass of TimeoutException --- .../clients/producer/BufferExhaustedException.java | 10 ++++------ .../apache/kafka/clients/producer/KafkaProducer.java | 6 +++--- .../kafka/clients/producer/internals/BufferPool.java | 4 ++-- .../clients/producer/internals/BufferPoolTest.java | 8 ++++---- 4 files changed, 13 insertions(+), 15 deletions(-) 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 b8931f90c629a..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,15 +16,13 @@ */ package org.apache.kafka.clients.producer; -import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.errors.TimeoutException; /** - * @deprecated This exception class is deprecated and is no longer used. It will be removed in future releases. - * 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 allocated 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. */ -@Deprecated -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 288ba1399c5e8..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,7 +479,7 @@ 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 (TimeoutException e) { + } catch (BufferExhaustedException e) { this.errors.record(); this.metrics.sensor("buffer-exhausted-records").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/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