DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
| Code Block | ||
|---|---|---|
| ||
/**
* Interface that specifies how an exception when attempting to produce a result to
* Kafka should be handled.
*/
public interface ProductionExceptionHandler extends Configurable {
/**
* Inspect a record that we attempted to produce, and the exception that resulted
* from attempting to produce it and determine whether or not to continue processing.
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
* @deprecated Please use the ProductionExceptionHandlerResponse.handle(record, exception, context)
*/
@Deprecated
ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
final Exception exception);
/**
* Inspect a record that we attempted to produce, and the exception that resulted
* from attempting to produce it and determine whether or not to continue processing.
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
* @param context Processing context
*/
@SuppressWarnings("deprecation")
default ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
final Exception exception,
final ProcessingContext context) {
return handle(record, exception);
}
/**
* Handles serialization exception and determine if the process should continue. The default implementation is to
* fail the process.
*
* @param record the record that failed to serialize
* @param exception the exception that occurred during serialization
* @deprecated Please use the handleSerializationException(record, exception, context)
*/
@Deprecated
default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
final Exception exception) {
return ProductionExceptionHandlerResponse.FAIL;
}
/**
* Handles serialization exception and determine if the process should continue. The default implementation is to
* fail the process.
*
* @param record the record that failed to serialize
* @param exception the exception that occurred during serialization
* @param context Processing context
*/
@SuppressWarnings("deprecation")
default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
final Exception exception,
final ProcessingContext context
) {
return handleSerializationException(record, exception);
}
enum ProductionExceptionHandlerResponse {
/*. continue. processing. */
public ProductionExceptionHandlerResponse CONTINUEwithDeadLetterQueueRecord(0, "CONTINUE"),
ProducerRecord<byte[], byte[]> deadLetterQueueRecord, String deadLetterQueueTopicName) {
/* fail processing */
this.deadLetterQueueRecord = FAIL(1, "FAIL");
deadLetterQueueRecord;
/**
* an english description of the api--this is for debugging and can change return this;
*/}
public final String name;
/**
* the permanent and immutable id of an API--this can't change ever
*/
public final int id;
public ProducerRecord<byte[], byte[]> deadLetterQueueRecord;
ProductionExceptionHandlerResponse(final int id,
final String name) {
this.id = id;
this.name = name;
}
public ProductionExceptionHandlerResponse withDeadLetterQueueRecord(ProducerRecord<byte[], byte[]> deadLetterQueueRecord) {
this.deadLetterQueueRecord = deadLetterQueueRecord;
return this;
}
}
}
|
DeserializationExceptionHandler.java
Changes:
- Adding the public ProducerRecord<byte[], byte[]> deadLetterQueueRecord; attribute in the ProductionExceptionHandlerResponse
}
}
|
DeserializationExceptionHandler.java
Changes:
- Adding the public ProducerRecord<byte[], byte[]> deadLetterQueueRecord; attribute in the ProductionExceptionHandlerResponse
| Code Block | ||
|---|---|---|
| ||
public interface DeserializationExceptionHandler extends Configurable {
/**
* Inspect a record and the exception received.
* <p>
* Note, that the passed in {@link ProcessorContext} only allows to access metadata like the task ID.
* However, it cannot be used to emit records via {@link ProcessorContext#forward(Object, Object)};
* calling {@code forward()} (and some other methods) would result in a runtime exception.
*
* @param context processor context
* @param record record that failed deserialization
* @param exception the actual exception
*/
@SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
DeserializationHandlerResponse handle(final ProcessorContext context, | ||
| Code Block | ||
| ||
public interface DeserializationExceptionHandler extends Configurable { /** * Inspect a record and the exception received. * <p> * Note, that the passed in {@link ProcessorContext} only allows to access metadata like the task ID. * However, it cannot be used to emit records via {@link ProcessorContext#forward(Object, Object)}; * calling {@code forward()} (and some other methods) would result in a runtime exception. * * @param context processor context * @param record record that failed deserialization * @param exception the actual exception */ @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated. DeserializationHandlerResponse handle(final ProcessorContext context, final ConsumerRecord<byte[], byte[]> record, final Exception exception); /** * Enumeration that describes the response from the exception handler. */ enum DeserializationHandlerResponse { /* continue with processing */ CONTINUE(0, "CONTINUE"), /* fail the processing and stop */ FAIL(1, "FAIL"); /** an english description of the api--this is for debugging and can change */ public final String name; final /** the permanent and immutable id of an API--this can't change ever */ ConsumerRecord<byte[], byte[]> record, public final int id; public ProducerRecord<byte[], byte[]> deadLetterQueueRecord; DeserializationHandlerResponse(final int id, final StringException name) {exception); /** * Enumeration that describes the response from this.idthe =exception id;handler. */ enum this.name = name; DeserializationHandlerResponse { . . } . public DeserializationHandlerResponse withDeadLetterQueueRecord(ProducerRecord<byte[], byte[]> deadLetterQueueRecord, String deadLetterQueueTopicName) { this.deadLetterQueueRecord = deadLetterQueueRecord; return this; } } } |
...
With KIP-1033, a similar behavior would be added to the potential new ProcessingExceptionHandler: adding a public ProducerRecord<byte[], byte[]> deadLetterQueueRecord; attribute in the ProcessingExceptionHandlerResponse the ProcessingExceptionHandlerResponse.
The goal is to support, regardless of the timeline, DLQ in the ProcessingExceptionHandler if KIP-1033 is approved. The change will either be part of KIP-1034 or KIP-1033 if this KIP is implemented first.
Compatibility, Deprecation, and Migration Plan
...