...
- Add a new getter and setter to configure dead letter queue records in the DeserializationHandlerResponseDeserializationExceptionHandler, ProductionExceptionHandlerResponse ProductionExceptionHandler and ProcessExceptionHandlerResponse enumsProcessExceptionHandler.
- 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.
...
| Code Block |
|---|
|
public static final String ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG = "errors.deadletterqueue.topic.name";
.define(ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG,
Type.STRING,
null, /* default */
Importance.HIGH,
ERRORS_DEADLETTERQUEUE_TOPIC_NAME_DOC) |
If the user implement implements a custom exception handler, it is up to the custom handler to build DLQ records to send, in this case, the errors.deadletterqueue.topic.name configuration has no impact.
| Code Block |
|---|
|
@Override
public ProcessingHandlerResponse handle(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {
return ProcessingHandlerResponse.CONTINUE addRecordsToDeadLetterQueue(
.andAddToDeadLetterQueue(Collections.singletonList(
new ProducerRecord<>("app-dlq", "Hello".getBytes(StandardCharsets.UTF_8), "World".getBytes(StandardCharsets.UTF_8)))
);
return ))ProcessingHandlerResponse.CONTINUE;
} |
ProductionExceptionHandler.java
...
- Adding the public getter ProducerRecord<byte[], byte[]> deadLetterQueueRecord() in the ProductionExceptionHandlerResponse as default implementation
- Adding the method addRecordsToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) to request records to be produced as default implementation
- As the ErrorHandlerContext does not provide the sourceKey/Value in the handle method to limit the memory impact, the Dead Letter Queue record would only contains metadata. handleSerializationException is not impacted.
| Code Block |
|---|
|
public interface ProductionExceptionHandler extends Configurable {
...
enum/** ProductionExceptionHandlerResponse {
. . .
public Iterable<org.apache.kafka.clients.producer.ProducerRecord<bytea list of Kafka records to publish, e.g. in a Dead Letter Queue topic */
Queue<ProducerRecord<byte[], byte[]> deadLetterQueueRecords>> deadLetterQueueRecordsQueue = new ConcurrentLinkedQueue<>();
default void public ProductionExceptionHandlerResponse andAddToDeadLetterQueue(Iterable<org.addRecordsToDeadLetterQueue(final Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords); {
}
}
|
DeserializationExceptionHandler.java
Changes:
- Adding the public getter ProducerRecord<byte[], byte[]> deadLetterQueueRecord() in the DeserializationExceptionHandlerResponse
| Code Block |
|---|
|
public interface DeserializationExceptionHandler extends Configurable {
if (deadLetterQueueRecords ...
/**== null) {
* Enumeration that describes the response from the exception handler. return;
*/
enum DeserializationHandlerResponse {}
. . .
public Iterable<org.apache.kafka.clients.producer.for (final ProducerRecord<byte[], byte[]> deadLetterQueueRecord : deadLetterQueueRecords) {
this.deadLetterQueueRecordsQueue.add(deadLetterQueueRecord);
}
}
public DeserializationHandlerResponse andAddToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<bytedefault List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords(); {
}
}
|
ProcessingExceptionHandler.java
Changes:
...
final LinkedList<ProducerRecord<byte[], byte[] |
...
>> deadLetterQueueRecords = new LinkedList<>();
while (true) {
final ProducerRecord<byte[], byte[] |
...
> record = this.deadLetterQueueRecordsQueue.poll();
if (record == null) {
break;
}
deadLetterQueueRecords.add(record);
}
return deadLetterQueueRecords;
}
... }
|
DeserializationExceptionHandler.java
Changes:
- Adding the public getter ProducerRecord<byte[], byte[]> deadLetterQueueRecord() in the DeserializationExceptionHandlerResponse as default implementation
- Adding the method addRecordsToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) to request records to be produced as default implementation
| Code Block |
|---|
|
public interface DeserializationExceptionHandler extends Configurable {
...
/** a list of Kafka records to publish, e.g. in a Dead Letter Queue topic */
Queue<ProducerRecord<byte[], byte[]>> deadLetterQueueRecordsQueue = new ConcurrentLinkedQueue<>();
default void addRecordsToDeadLetterQueue(final Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
if (deadLetterQueueRecords == null) {
return;
}
for (final ProducerRecord<byte[], byte[]> deadLetterQueueRecord : deadLetterQueueRecords) {
this.deadLetterQueueRecordsQueue.add(deadLetterQueueRecord);
}
}
default List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
final LinkedList<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords = new LinkedList<>();
while (true) {
final ProducerRecord<byte[], byte[]> record = this.deadLetterQueueRecordsQueue.poll();
if (record == null) {
break;
}
deadLetterQueueRecords.add(record);
}
return deadLetterQueueRecords;
}
...
}
|
ProcessingExceptionHandler.java
Changes:
- Adding the public getter ProducerRecord<byte[], byte[]> deadLetterQueueRecord() in the ProcessingExceptionHandler as default implementation
- Adding the method addRecordsToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) to request records to be produced as default implementation
| Code Block |
|---|
|
public interface ProcessingExceptionHandler extends Configurable {
...
Queue<ProducerRecord<byte[], byte[]>> deadLetterQueueRecordsQueue = new ConcurrentLinkedQueue<>();
default void addRecordsToDeadLetterQueue(Iterable<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
if (deadLetterQueueRecords == null) {
return;
}
for (ProducerRecord<byte[], byte[]> record : deadLetterQueueRecords) {
deadLetterQueueRecordsQueue.add(record);
}
}
default List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords = new LinkedList<>();
while (true) {
ProducerRecord<byte[], byte[]> record = deadLetterQueueRecordsQueue.poll();
if (record == null) {
break;
}
deadLetterQueueRecords.add(record);
}
return deadLetterQueueRecords;
}
...
}
|
RecordContext.java
Changes:
- Adding the public byte[] sourceRawKey and byte[] sourceRawValue in the RecordContext pointing to the source record data
| Code Block |
|---|
| language | java |
|---|
| title | ErrorHandlerContext.java |
|---|
|
/**
* RecordContext interface
*/
public interface RecordContext {
. . . /**
* Return the non-deserialized byte[] of the input message key if the context has been triggered by a message.
*
* <p> If this method is invoked within a {@link Punctuator#punctuate(long)
* punctuation callback}, or while processing a record that was forwarded by a punctuation
* callback, it will return null.
*
* <p> If this method is invoked in a sub-topology due to a repartition, the returned key would be one sent
* to the repartition topic.
*
* <p> Always returns null if this method is invoked within a
* ProductionExceptionHandler.handle(ErrorHandlerContext, ProducerRecord, Exception)
*
* @return the raw byte of the key of the source message
*/
byte[] sourceRawKey();
/**
* Return the non-deserialized byte[] of the input message value if the context has been triggered by a message.
*
* <p> If this method is invoked within a {@link Punctuator#punctuate(long)
* punctuation callback}, or while processing a record that was forwarded by a punctuation
* callback, it will return null.
*
* <p> If this method is invoked in a sub-topology due to a repartition, the returned key would be one sent
* to the repartition topic.
*
* <p> Always returns null if this method is invoked within a
* ProductionExceptionHandler.handle(ErrorHandlerContext, ProducerRecord, Exception)
*
* @return the raw byte of the value of the source message
*/
byte[] sourceRawValue();
. . .
} |
| Code Block |
|---|
|
public interface ProcessingExceptionHandler extends Configurable {
...
/**
* Enumeration that describes the response from the exception handler.
*/
enum ProcessingHandlerResponse {
. . .
public Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]> deadLetterQueueRecords();
public ProcessingHandlerResponse andAddToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords);
}
}
|
ErrorHandlerContext.java
Changes:
...