Page History

...

If the default values are not suitable for an application, developers could still reimplement the required exception handlers to build custom DLQ records.

This proposal is to:

Capture the initial message key and values bytes and expose them in the ProcessingContext.
Add a new attribute "deadLetterQueueRecord" in the DeserializationHandlerResponse, ProductionExceptionHandlerResponse and ProcessExceptionHandlerResponse (KIP-1033) enum.
Add the processingContext 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.

Note: to be complete and rely on the ProcessingMetadata container class proposed in KIP-1033, this KIP has a hard dependency on KIP-1033.

Default Dead letter queue record

...

Code Block

language	java

public static final String ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG = "errors.deadletterqueue.topic.name";

.define(ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, // required with no default value
       Type.STRING,
       null, /* default */
       Importance.HIGH,
       ERRORS_DEADLETTERQUEUE_TOPIC_NAME_DOC)

ProcessingContext.java

Code Block

language	java

public interface ProcessingContext {

. . . 

/**
* 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.
*
* @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 value would be one sent
* to the repartition topic.
*
* @return the raw byte of the value of the source message
*/
byte[] sourceRawValue();


. . . 

}

ProductionExceptionHandler.java

...

Adding the ProcessingContext attribute in the handle and handleSerialization methods, ensure the backward compatibility of previous implementations of this handler by providing default implementation.
Adding the public ProducerRecord<byte[], byte[]> deadLetterQueueRecord; attribute in the ProductionExceptionHandlerResponse
Deprecate the previous method

Code Block

language	java

/**
* 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 {
       . . . 

       public ProductionExceptionHandlerResponse withDeadLetterQueueRecord(ProducerRecord<byte[], byte[]> deadLetterQueueRecord, String deadLetterQueueTopicName) {
           this.deadLetterQueueRecord = deadLetterQueueRecord;
           return this;
       }
   }
}

...

Space shortcuts

Child pages

Versions Compared

Old Version 11

New Version 12

Key

Default Dead letter queue record

ProcessingContext.java

ProductionExceptionHandler.java