DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
- UnknownTopicOrPartitionException: For this case, the producer handles this exception internally and only issues a WARN log about missing metadata and retries internally. Later, when the producer hits the "
deliverdelivery.timeout.ms",it throws a TimeoutException and the user can only blindly retry, resulting in an infinite retry loop. The thrown TimeoutException "cuts" the connection to the underlying root cause of missing metadata (which could indeed be a transient error but is persistent for a non-existing topic). Thus, there is no programmatic way to break the infinite retry loop. Kafka Streams also blindly retries for this case, and the application gets stuck.
...
Always, the most "conservative" thing will hit first. E.g., if there is a Retriable retriable 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. 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 NonRetriableResponse 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 RetriableResponse handle(final ProducerRecord record, final UnknownTopicOrPartitionException exception) {
// return the value corresponding to the default behaviour
}
enum NonRetriableResponse {
/* 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 RetriableResponse {
/* 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;
}
}
} |
...