DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
We introduce the ProducerExceptionHandler interface, which can be implemented by the user to manageUnknownTopicOrPartitionException and RecordTooLargeException in the desired manner. It can either stop or continue the processing. Stopping processing is named as FAIL since the transaction or record sending (in non-transactional mode) will fail. In the case of continuing processing, either the record is dropped and the error is ignored (SWALLOW) or sending is retried (RETRY). The accepted responses for RecordTooLargeException are FAIL and SWALLOW. Therefore, RETRY will be interpreted and executed as FAIL.
To configure their own handler, the user must implement the above-introduced interface and add the class name in producer configuration with the key: custom.exception.handler.
More than that, we introduced two more configuration parameters as follows to handle the two exceptions without needing too implement the interface.
drop.invalid.large.recordswith a default value of `false` for swallowing too large records.retry.unknown.topic.partitionwith a default value of `true` that performs RETRY for `max.block.ms` encountering the UnknownTopicOrPartitionException.
| 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
*/
RecordTooLargeExceptionResponse handle(final ProducerRecord record, final RecordTooLargeException exception);
/**
* 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
*/
UnknownTopicOrPartitionExceptionResponse handle(final ProducerRecord record, final UnknownTopicOrPartitionException exception);
enum RecordTooLargeExceptionResponse {
/* 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 UnknownTopicOrPartitionExceptionResponse {
/* 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;
}
}
} |
...