DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
drop.invalid.large.recordswith a default value of `false` for swallowing too large records.retry.unknown.topic.partition.mswith a default value of `Integer.MAX_VALUE` that performs RETRY for min(`max.block.ms`, `retry.unknown.topic.partition.ms`) encountering the UnknownTopicOrPartitionException.
In the case of an implemented handler for the specified exception, the handler takes precedence.
Always, the most "conservative" thing will hit first. E.g., if there is a retryable error, the producer has a retry timeout, and the handler might have one; whichever timeout hits first will stop the retry loop. If the handler says "SWALLOW", we also break the retry loop (more conservative), and if the handler says FAIL, we fail right away, not even waiting for a timeout to hit
The default behaviour stays as it is, but the user can change the behaviour by implementing the handle() function.
. Obviously, the order of conservativity is FAIL > SWALLOW > RETRY, where FAIL is the most conservative action and RETRY is the least.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
package org.apache.kafka.clients.producer;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import java.io.Closeable;
/**
* Interface that specifies how an the RecordTooLargeException and/or UnknownTopicOrPartitionException should be handled.
* The accepted responses for RecordTooLargeException are FAIL and SWALLOW. Therefore, RETRY will be interpreted and executed as FAIL.
*/
public interface ProducerExceptionHandler extends Configurable, Closeable {
/**
* Determine whether to stop processing, or swallow the error by dropping the record.
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
*/
default NonRetryableResponse handle(final ProducerRecord record, final RecordTooLargeException exception) {
// return the value corresponding to the default behaviour
}
/**
* Determine whether to stop processing, keep retrying internally, or swallow the error by dropping the record.
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
*/
default RetryableResponse handle(final ProducerRecord record, final UnknownTopicOrPartitionException exception) {
// return the value corresponding to the default behaviour
}
enum NonRetryableResponse {
/* stop processing: fail */
FAIL(0, "FAIL"),
/* drop the record and continue */
SWALLOW(1, "SWALLOW");
/**
* an english description of the api--this is for debugging and can change
*/
public final String name;
/**
* the permanent and immutable id of an API--this can't change ever
*/
public final int id;
RecordTooLargeExceptionResponse(final int id, final String name) {
this.id = id;
this.name = name;
}
}
enum RetryableResponse {
/* stop processing: fail */
FAIL(0, "FAIL"),
/* continue: keep retrying */
RETRY(1, "RETRY"),
/* continue: swallow the error */
SWALLOW(2, "SWALLOW");
/**
* an english description of the api--this is for debugging and can change
*/
public final String name;
/**
* the permanent and immutable id of an API--this can't change ever
*/
public final int id;
UnknownTopicOrPartitionExceptionResponse(final int id, final String name) {
this.id = id;
this.name = name;
}
}
} |
...
Compatibility, Deprecation, and Migration Plan
Changed behaviour: The default behaviour stays as it is, but the user can change the behaviour by implementing the handle() function functions as well as setting the two newly introduced config parameters.
Test Plan
Some unit and integration tests will be implemented to ensure that
...