Skip to content

KAFKA-3720 : Deprecated BufferExhaustedException and also removed its use and the related sensor metric #1417

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
MayureshGharat wants to merge 3 commits into from
Closed

KAFKA-3720 : Deprecated BufferExhaustedException and also removed its use and the related sensor metric #1417

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fb66dcf
Deprecated BufferExhaustedException and also removed its use and the …
MayureshGharat May 20, 2016
ab879b4
Update the buffer-exhausted-records sensor when a timeoutException is…
MayureshGharat Jun 10, 2016
d137aff
Made BufferExhaustException a subclass of TimeoutException
MayureshGharat Nov 21, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions clients/src/main/java/org/apache/kafka/clients/producer/BufferExhaustedException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
16 changes: 8 additions & 8 deletions clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -424,9 +424,9 @@ public Future<RecordMetadata> send(ProducerRecord<K, V> 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 <code>max.block.ms</code>.
* @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 <code>max.block.ms</code>.
* @throws BufferExhaustedException if the time taken for allocating memory for the record has surpassed <code>max.block.ms</code>.
*/
@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
Expand Down Expand Up @@ -479,6 +479,12 @@ private Future<RecordMetadata> doSend(ProducerRecord<K, V> 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;

@kawamuray kawamuray Nov 22, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is gonna be a slight spec change. Before this, a TimeoutException thrown either by waiting metadata or by waiting buffer allocation were caught by the following clause for ApiException(since TimeoutException extends RetriableException which is an ApiException), so a FutureFailure returned instead of throwing, and the callback triggered too.

As the result of this change, two TimeoutException cases are treated differently:

  • timeout occurred while waiting metadata update => callback called, FutureFailure returned instead of throw
  • timeout occurred while waiting buffer allocation => callback not called, throw exception

This sounds confusing and inconsistent. I think we can either take 1. always return FailureFuture for TimeoutException or 2. always throw for timeout exception.
IMO, by design of KafkaProducer(method blocking is part of interface), the latter one makes more sense to me.
WDYT?

@ijuma ijuma Nov 22, 2016

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a good point. The current implementation is a bit misleading with regards to the javadoc which states:

* @throws TimeoutException If the time taken for fetching metadata or allocating memory for the record has surpassed <code>max.block.ms</code>.

We don't actually throw that exception unless you do Future.get. It seems to me that the change in this PR actually fixes the implementation to match the specified contract.

Another option is to change the javadoc to match the implementation (probably less likely to break users) and then we would simply have a check in the ApiException catch block to record the metric for BufferExhaustedException. This seems safer.

Thoughts?

@kawamuray kawamuray Nov 22, 2016 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right, it's indeed an option which is much safer.
Still I think we better throw here by following reasons:

  • It sounds semantically correct more. Callback and Future should be used for providing result of asynchronous(background) processing. However, these two TimeoutException occurrs while a KafkaProducer is still doing synchronous(foreground) processing and that(kafka producer has to do some foreground processing before it appends record to the accumulator) is the reason why a caller of producer#send is forced to wait until the result turns out, so the caller should receive the result of that call in a synchronous way.
  • Assuming this is going to be shipped with 0.10.2.0, making a breaking change on behavior isn't preferred but allowed if we leave a correct note on "breaking changes" section.
  • We can expect this breaking change is relatively less-harm with expect to most users who uses producer in sensitive situation would already using it like below. Users may see some new error logs but it still doesn't break the whole processing(maybe I'm biased, feel free to leave objection if any :D)
try {
    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
          // logging
        }
     });
} catch (RuntimeException/* or maybe KafkaException, TimeoutException whatever */ e) {
     // logging
}

} catch (ApiException e) {
log.debug("Exception occurred during message send:", e);
if (callback != null)
Expand All @@ -492,12 +498,6 @@ private Future<RecordMetadata> doSend(ProducerRecord<K, V> 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)
Expand Down
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}

Expand Down
8 changes: 4 additions & 4 deletions clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand All @@ -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);
Expand Down Expand Up @@ -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
Expand Down