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) enumenums.
- Add the processingMetadata attribute in the ProductionExceptionHandlerResponse.handle method and ensure backward compatibility with previous implementations of the interface.
- Add a new attribute public static final String DEFAULT_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.
...
| Code Block | ||
|---|---|---|
| ||
public interface ProductionExceptionHandler extends Configurable {
...
enum ProductionExceptionHandlerResponse {
. . .
public ProductionExceptionHandlerResponse withDeadLetterQueueRecord(withDeadLetterQueueRecords(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]> deadLetterQueueRecord, String deadLetterQueueTopicName>> 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 withDeadLetterQueueRecord(withDeadLetterQueueRecords(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]> deadLetterQueueRecord, String deadLetterQueueTopicName>> deadLetterQueueRecords) {
this.deadLetterQueueRecord = deadLetterQueueRecord;
return this;
}
}
}
|
...