DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
- Add a new attribute "deadLetterQueueRecord" in the DeserializationHandlerResponse, ProductionExceptionHandlerResponse and ProcessExceptionHandlerResponse (KIP-1033) enums.
- Add a new attribute public static final String ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG = "errors.deadletterqueue.topic.name".
- Change the existing exception handler to produce a DeadLetterQueue record if the parameter errors.deadletterqueue.topic.name is set.
- If the DeadLetterQueue record can not be sent to Apache Kafka, the exception would be sent to the Kafka Streams uncaughtExceptionHandler.
Note: to be complete and rely on the ErrorHandlerContext class proposed in KIP-1033, this KIP has a hard dependency on KIP-1033.
Default Dead letter queue record
...
| Code Block | ||
|---|---|---|
| ||
@Override
public ProcessingHandlerResponse handle(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {
return ProcessingHandlerResponse.CONTINUE
.withDeadLetterQueueRecordsandAddToDeadLetterQueue(Collections.singletonList(
new ProducerRecord<>("app-dlq", "Hello".getBytes(StandardCharsets.UTF_8), "World".getBytes(StandardCharsets.UTF_8))
));
} |
...
| Code Block | ||
|---|---|---|
| ||
public interface ProductionExceptionHandler extends Configurable {
...
enum ProductionExceptionHandlerResponse {
. . .
public ProductionExceptionHandlerResponse withDeadLetterQueueRecordsandAddToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
this.deadLetterQueueRecord = deadLetterQueueRecord;
return this;
}
}
}
|
...
| Code Block | ||
|---|---|---|
| ||
public interface DeserializationExceptionHandler extends Configurable {
...
/**
* Enumeration that describes the response from the exception handler.
*/
enum DeserializationHandlerResponse {
. . .
public DeserializationHandlerResponse withDeadLetterQueueRecordsandAddToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
this.deadLetterQueueRecord = deadLetterQueueRecord;
return this;
}
}
}
|
...
| Code Block | ||
|---|---|---|
| ||
public interface DeserializationExceptionHandler extends Configurable {
...
/**
* Enumeration that describes the response from the exception handler.
*/
enum DeserializationHandlerResponse {
. . .
public DeserializationHandlerResponse withDeadLetterQueueRecordsandAddToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
this.deadLetterQueueRecord = deadLetterQueueRecord;
return this;
}
}
}
|
...