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 manage the exception in the desired manner.
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.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
package org.apache.kafka.common.errors;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.annotation.InterfaceStability;
/**
* Interface that specifies how an exception should be handled.
*/
@InterfaceStability.Evolving
public interface ProducerExceptionHandler extends Configurable {
/**
* 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
*/
ProducerExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
final Exception exception);
enum ProducerExceptionHandlerResponse {
/* 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;
ProducerExceptionHandlerResponse(final int id,
final String name) {
this.id = id;
this.name = name;
}
}
}
|
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
public static final String PRODUCERCUSTOM_EXCEPTION_HANDLER_CLASS_CONFIG = "producercustom.exception.handler"; private static final String PRODUCERCUSTOM_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.common.errors.ProducerExceptionHandler</code> interface.";. . . static { CONFIG = new ConfigDef().define( ..... . . . .define(PRODUCERCUSTOM_EXCEPTION_HANDLER_CLASS_CONFIG, Type.CLASS, null, Importance.MEDIUM, PRODUCER CUSTOM_EXCEPTION_HANDLER_CLASS_DOC); } |
...
Compatibility, Deprecation, and Migration Plan
To configure own handler, the user must implement the above introduced interface and add the class name in producer configuration with the key: producer.exception.handler.
Changed behaviour: The default behaviour stays as it is, but the user can change the behaviour by implementing the handle() function.
...