Table of Contents

This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: Under DiscussionAdopted

Discussion thread: here

JIRA: here

...

To allow users to send a record in the Deal Dead letter queue, a new attribute "deadLetterQueueRecord'' will be added in each exception handler's responses. If this attribute is set, KafkaStreams will send the provided record to Kafka.

A new configuration will be added: errors.dead.letter.deadletterqueuequeue.topic.name. When set, this configuration indicates the default exception handler implementation to build a Dead letter queue record during the error handling.

...

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 ProcessingContextAdd a new getter and setter to configure dead letter queue records in the DeserializationExceptionHandler, ProductionExceptionHandler and ProcessExceptionHandler.
Add a new attribute "deadLetterQueueRecord" in the DeserializationHandlerResponse, ProductionExceptionHandlerResponse and ProcessExceptionHandlerResponse (KIP-1033) enum.
Add the processingContext 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_DEADLETTERQUEUEpublic static final String ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG = "errors.dead.letter.deadletterqueuequeue.topic.name".
Change the existing exception handler to produce a DeadLetterQueue record if the parameter errors.dead.letter.deadletterqueuequeue.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.

Default Dead letter queue record

If , "error during punctuate"

If messages exceed Kafka maximum size "message exceeding Kafka maximum record size"

Header:

Key	Key of the input message that triggered the sub-topology, null if triggered by punctuate
Value	If available, contains the value of the input message
that triggered the sub-topology, null if triggered by punctuate

Headers	Existing context headers are automatically forwarded into the new DLQ record
Header: __streams.errors.
exception	Name of the thrown exception
Header: __streams.errors.stacktrace	Stacktrace of the thrown exception
Header: __streams.errors.message	Thrown exception message
Header: __streams.errors.topic	Source input topic, null if triggered by punctuate
Header: __streams.errors.partition	Source input partition, null if triggered by punctuate
Header: __streams.errors.offset	Source input offset, null if triggered by punctuate

Public Interfaces

StreamsConfig.java

...

language	java

...

Default Dead letter queue topic

By default, this KIP proposes to have on DLQ topic per Kafka Streams application. This topic would not be automatically created by Kafka Streams.

The DLQ topic name is set through the configuration ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG

...

=

...

" errors.dead.letter.

...

queue.topic.name"

...

. Users can override the default behavior by implementing custom exception handlers to implement a different DLQ topic strategy if required.

Public Interfaces

StreamsConfig.java

Changes:

Adding the errors.dead.letter.queue.topic.name configuration. This configuration is only modifying the behavior of the out of the box exceptions handlers and would have no effect if a custom exception handlers is implemented, for example:
- if errors.dead.letter.queue.topic.name=null (default), then no records are sent to any dead letter queue
- if errors.dead.letter.queue.topic.name is set, exceptions happening during processing, production or deserialization will result in the raw source messages that trigger the topology to be send to the DLQ topic. The processing might or might not continue depending of the configuration of the processing.exception.handler, default.production.exception.handler and default.deserialization.exception.handler configurations

Code Block

language	java

public static final String ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG = "errors.dead.letter.queue.topic.name";

.define(ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG,
       Type.STRING,
       null, /* default */
       Importance.MEDIUM,
       ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_DOC)

If the user implements a custom exception handler, it is up to the custom handler to build DLQ records to send, in this case, the errors.dead.letter.queue.topic.name configuration has no impact.

Code Block

language	java

    @Override
    public Response handleError(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {         
        List<ProducerRecord<byte[], byte[]>> records = Collections.singletonList(new ProducerRecord<>("app-dlq", "Hello".getBytes(StandardCharsets.UTF_8), "World".getBytes(StandardCharsets.UTF_8)));

        return ProcessingExceptionHandler.Response.resume(records);  
   }

ProductionExceptionHandler.java

Changes:

Adding a Response nested class,that contains a Result indicating whether to continue processing, fail, or retry and the list of records to be sent to the dead letter queue topic
Deprecating the handler() and handlerSerializationException() methods and adding two new methods: handlerError() and handleSerializationException(). The return type is the Response
Deprecating the ProductionExceptionHandlerResponse enum and create a new Result enum. The result enum contains the 3 fields RESUME, FAIL and RETRY and a deprecated method to convert the old enum to the new one
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

language	java

public interface ProductionExceptionHandler extends Configurable {
   ...   
     

    @Deprecated
    default ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
                                                      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 to continue or stop processing.
     *
     * @param context
     *     The error handler context metadata.
     * @param record
     *     The record that failed to produce.
     * @param exception
     *     The exception that occurred during production.
     *
     * @return a {@link Response} object
     */
    default Response handleError(final ErrorHandlerContext context,
                                                    final ProducerRecord<byte[], byte[]> record,
                                                    final Exception exception) {                          
          return new Response(Result.from(handle(context, record, exception)), Collections.emptyList());
     }

    @Deprecated
    default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
                                                                            final Exception exception) {
        ...
    }

    /**
     * Handles serialization exception and determine if the process should continue. The default implementation is to
     * fail the process.
     *
     * @param context
     *     The error handler context metadata.
     * @param record
     *     The record that failed to serialize.
     * @param exception
     *     The exception that occurred during serialization.
     * @param origin
     *     The origin of the serialization exception.
     *
     * @return a {@link Response} object
     */     
    default Response handleSerializationError(final ErrorHandlerContext context,
                                                                 final ProducerRecord record,
                                                                 final Exception exception,
                                                                 final SerializationExceptionOrigin origin) {                  
        return new Response(Result.from(handleSerializationException(context, record, exception, origin)), Collections.emptyList());
    }          
    
    @Deprecated
    enum ProductionExceptionHandlerResponse {
      ...
    }

        /**
     * Enumeration that describes the response from the exception handler.
     */
    enum Result {
        /** Resume processing.
         *
         * <p> For this case, output records which could not be written successfully are lost.
         * Use this option only if you can tolerate data loss.
         */
        RESUME(0, "RESUME"),
        /** Fail processing.
         *
         * <p> Kafka Streams will raise an exception and the {@code StreamsThread} will fail.
         * No offsets (for {@link org.apache.kafka.streams.StreamsConfig#AT_LEAST_ONCE at-least-once}) or transactions
         * (for {@link org.apache.kafka.streams.StreamsConfig#EXACTLY_ONCE_V2 exactly-once}) will be committed.
         */
        FAIL(1, "FAIL"),
        /** Retry the failed operation.
         *
         * <p> Retrying might imply that a {@link TaskCorruptedException} exception is thrown, and that the retry
         * is started from the last committed offset.
         *
         * <p> <b>NOTE:</b> {@code RETRY} is only a valid return value for
         * {@link org.apache.kafka.common.errors.RetriableException retriable exceptions}.
         * If {@code RETRY} is returned for a non-retriable exception it will be interpreted as {@link #FAIL}.
         */
        RETRY(2, "RETRY");

        /**
         * An english description for the used option. This is for debugging only and may change.
         */
        public final String name;

        /**
         * The permanent and immutable id for the used option. This can't change ever.
         */
        public final int id;

        Result(final int id, final String name) {
            this.id = id;
            this.name = name;
        }

        /**
         * Converts the deprecated enum ProductionExceptionHandlerResponse into the new Result enum.
         *
         * @param value the old ProductionExceptionHandlerResponse enum value
         * @return a {@link ProductionExceptionHandler.Result} enum value
         * @throws IllegalArgumentException if the provided value does not map to a valid {@link ProductionExceptionHandler.Result}
         */
        private static ProductionExceptionHandler.Result from(final ProductionExceptionHandlerResponse value) {
            switch (value) {
                case FAIL:
                    return Result.FAIL;
                case CONTINUE:
                    return Result.RESUME;
                case RETRY:
                    return Result.RETRY;
                default:
                    throw new IllegalArgumentException("No Result enum found for old value: " + value);
            }
        }
    }

    /**
     * Represents the result of handling a production exception.
     * <p>
     * The {@code Response} class encapsulates a {@link ProductionExceptionHandlerResponse},
     * indicating whether processing should continue or fail, along with an optional list of
     * {@link ProducerRecord} instances to be sent to a dead letter queue.
     * </p>
     */
    class Response {

        private Result result;

        private List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords;

        /**
         * Constructs a new {@code Response} object.
         *
         * @param result the result indicating whether processing should continue or fail;
         *                                  must not be {@code null}.
         * @param deadLetterQueueRecords    the list of records to be sent to the dead letter queue; may be {@code null}.
         */
        private Response(final Result result,
                         final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            this.result = result;
            this.deadLetterQueueRecords = deadLetterQueueRecords;
        }

        /**
         * Creates a {@code Response} indicating that processing should fail.
         *
         * @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
         * @return a {@code Response} with a {@link ProductionExceptionHandler.Result#FAIL} status.
         */
        public static Response fail(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new Response(Result.FAIL, deadLetterQueueRecords);
        }

        /**
         * Creates a {@code Response} indicating that processing should fail.
         *
         * @return a {@code Response} with a {@link ProductionExceptionHandler.Result#FAIL} status.
         */
        public static Response fail() {
            return fail(Collections.emptyList());
        }

        /**
         * Creates a {@code Response} indicating that processing should continue.
         *
         * @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
         * @return a {@code Response} with a {@link ProductionExceptionHandler.Result#RESUME} status.
         */
        public static Response resume(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new Response(Result.RESUME, deadLetterQueueRecords);
        }

        /**
         * Creates a {@code Response} indicating that processing should continue.
         *
         * @return a {@code Response} with a {@link ProductionExceptionHandler.Result#RESUME} status.
         */
        public static Response resume() {
            return resume(Collections.emptyList());
        }

        /**
         * Creates a {@code Response} indicating that processing should retry.
         *
         * @return a {@code Response} with a {@link ProductionExceptionHandler.Result#RETRY} status.
         */
        public static Response retry() {
            return new Response(Result.RETRY, Collections.emptyList());
        }

        /**
         * Retrieves the production exception handler result.
         *
         * @return the {@link Result} indicating whether processing should continue, fail or retry.
         */
        public Result result() {
            return result;
        }

        /**
         * Retrieves an unmodifiable list of records to be sent to the dead letter queue.
         * <p>
         * If the list is {@code null}, an empty list is returned.
         * </p>
         *
         * @return an unmodifiable list of {@link ProducerRecord} instances
         *         for the dead letter queue, or an empty list if no records are available.
         */
        public List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
            if (deadLetterQueueRecords == null) {
                return Collections.emptyList();
            }
            return Collections.unmodifiableList(deadLetterQueueRecords);
        }
    }
}

DeserializationExceptionHandler.java

Changes:

Adding a Response nested class,that contains a Result indicating whether to continue processing or fail, and the list of records to be sent to the dead letter queue topic
Deprecating the handler() method and adding a new method: handlerError(). The return type is the Response
Deprecating the DeserializationHandlerResponse enum and create a new Result enum. The result enum contains the 2 fields RESUME and Fail and a deprecated method to convert the old enum to the new one

Code Block

language	java

public interface DeserializationExceptionHandler extends Configurable {

   ...      
  
    @Deprecated
    default DeserializationHandlerResponse handle(final ErrorHandlerContext context,
                                                  final ConsumerRecord<byte[], byte[]> record,
                                                  final Exception exception) {
        ...
    }

    /**
     * Inspects a record and the exception received during deserialization.
     *
     * @param context
     *     Error handler context.
     * @param record
     *     Record that failed deserialization.
     * @param exception
     *     The actual exception.
     *
     * @return a {@link Response} object
     */
    default Response handleError(final ErrorHandlerContext context, final ConsumerRecord<byte[], byte[]> record, final Exception exception) {           
        return new Response(Result.from(handle(context, record, exception)), Collections.emptyList());
    }     

    @Deprecated
    enum DeserializationHandlerResponse {
        ...
    }

    /**
     * Enumeration that describes the response from the exception handler.
     */
    enum Result {
        /** Continue processing. */
        RESUME(0, "RESUME"),
        /** Fail processing. */
        FAIL(1, "FAIL");

        /**
         * An english description for the used option. This is for debugging only and may change.
         */
        public final String name;

        /**
         * The permanent and immutable id for the used option. This can't change ever.
         */
        public final int id;

        Result(final int id, final String name) {
            this.id = id;
            this.name = name;
        }

        /**
         * Converts the deprecated enum DeserializationHandlerResponse into the new Result enum.
         *
         * @param value the old DeserializationHandlerResponse enum value
         * @return a {@link Result} enum value
         * @throws IllegalArgumentException if the provided value does not map to a valid {@link Result}
         */
        private static DeserializationExceptionHandler.Result from(final DeserializationHandlerResponse value) {
            switch (value) {
                case FAIL:
                    return Result.FAIL;
                case CONTINUE:
                    return Result.RESUME;
                default:
                    throw new IllegalArgumentException("No Result enum found for old value: " + value);
            }
        }
    }

    /**
     * Represents the result of handling a deserialization exception.
     * <p>
     * The {@code Response} class encapsulates a {@link ProcessingExceptionHandler.Result},
     * indicating whether processing should continue or fail, along with an optional list of
     * {@link ProducerRecord} instances to be sent to a dead letter queue.
     * </p>
     */
    class Response {

        private Result result;

        private List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords;

        /**
         * Constructs a new {@code DeserializationExceptionResponse} object.
         *
         * @param result the result indicating whether processing should continue or fail;
         *                                  must not be {@code null}.
         * @param deadLetterQueueRecords    the list of records to be sent to the dead letter queue; may be {@code null}.
         */
        private Response(final Result result,
                         final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            this.result = result;
            this.deadLetterQueueRecords = deadLetterQueueRecords;
        }

        /**
         * Creates a {@code Response} indicating that processing should fail.
         *
         * @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
         * @return a {@code Response} with a {@link DeserializationExceptionHandler.Result#FAIL} status.
         */
        public static Response fail(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new Response(Result.FAIL, deadLetterQueueRecords);
        }

        /**
         * Creates a {@code Response} indicating that processing should fail.
         *
         * @return a {@code Response} with a {@link DeserializationExceptionHandler.Result#FAIL} status.
         */
        public static Response fail() {
            return fail(Collections.emptyList());
        }

        /**
         * Creates a {@code Response} indicating that processing should continue.
         *
         * @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
         * @return a {@code Response} with a {@link DeserializationExceptionHandler.Result#RESUME} status.
         */
        public static Response resume(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new Response(Result.RESUME, deadLetterQueueRecords);
        }

        /**
         * Creates a {@code Response} indicating that processing should continue.
         *
         * @return a {@code Response} with a {@link DeserializationHandlerResponse#CONTINUE} status.
         */
        public static Response resume() {
            return resume(Collections.emptyList());
        }

        /**
         * Retrieves the deserialization handler result.
         *
         * @return the {@link Result} indicating whether processing should continue or fail.
         */
        public Result result() {
            return result;
        }

        /**
         * Retrieves an unmodifiable list of records to be sent to the dead letter queue.
         * <p>
         * If the list is {@code null}, an empty list is returned.
         * </p>
         *
         * @return an unmodifiable list of {@link ProducerRecord} instances
         *         for the dead letter queue, or an empty list if no records are available.
         */
        public List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
            if (deadLetterQueueRecords == null) {
                return Collections.emptyList();
            }
            return Collections.unmodifiableList(deadLetterQueueRecords);
        }
    }
}

ProcessingExceptionHandler.java

Changes:

Adding a Response nested class,that contains a Result indicating whether to continue processing or fail, and the list of records to be sent to the dead letter queue topic
Deprecating the handler() method and adding a new method: handlerError(). The return type is the Response
Deprecating the ProcessingHandlerResponse enum and create a new Result enum. The result enum contains the 2 fields RESUME and Fail and a deprecated method to convert the old enum to the new one

Code Block

language	java

public interface ProcessingExceptionHandler extends Configurable {

   ...   
    
    @Deprecated
    default ProcessingHandlerResponse handle(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception){
       ...
    };

    /**
     * Inspects a record and the exception received during processing.
     *
     * @param context
     *     Processing context metadata.
     * @param record
     *     Record where the exception occurred.
     * @param exception
     *     The actual exception.
     *
     * @return a {@link Response} object
     */
    default Response handleError(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {                          
        return new Response(ProcessingExceptionHandler.Result.from(handle(context, record, exception)), Collections.emptyList());
    }   

    @Deprecated
    enum ProcessingHandlerResponse {
       ...
    } 

    /**
     * Enumeration that describes the response from the exception handler.
     */
    enum Result {
        /** Resume processing. */
        RESUME(1, "RESUME"),
        /** Fail processing. */
        FAIL(2, "FAIL");

        /**
         * An english description for the used option. This is for debugging only and may change.
         */
        public final String name;

        /**
         * The permanent and immutable id for the used option. This can't change ever.
         */
        public final int id;

        Result(final int id, final String name) {
            this.id = id;
            this.name = name;
        }

        /**
         * Converts the deprecated enum ProcessingHandlerResponse into the new Result enum.
         *
         * @param value the old DeserializationHandlerResponse enum value
         * @return a {@link ProcessingExceptionHandler.Result} enum value
         * @throws IllegalArgumentException if the provided value does not map to a valid {@link ProcessingExceptionHandler.Result}
         */
        private static ProcessingExceptionHandler.Result from(final ProcessingHandlerResponse value) {
            switch (value) {
                case FAIL:
                    return Result.FAIL;
                case CONTINUE:
                    return Result.RESUME;
                default:
                    throw new IllegalArgumentException("No Result enum found for old value: " + value);
            }
        }
    }
          /**
     * Represents the result of handling a processing exception.
     * <p>
     * The {@code Response} class encapsulates a {@link Result},
     * indicating whether processing should continue or fail, along with an optional list of
     * {@link org.apache.kafka.clients.producer.ProducerRecord} instances to be sent to a dead letter queue.
     * </p>
     */
    class Response {

        private Result result;

        private List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords;

        /**
         * Constructs a new {@code ProcessingExceptionResponse} object.
         *
         * @param result the result indicating whether processing should continue or fail;
         *                                  must not be {@code null}.
         * @param deadLetterQueueRecords    the list of records to be sent to the dead letter queue; may be {@code null}.
         */
        private Response(final Result result,
                         final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            this.result = result;
            this.deadLetterQueueRecords = deadLetterQueueRecords;
        }

        /**
         * Creates a {@code Response} indicating that processing should fail.
         *
         * @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
         * @return a {@code Response} with a {@link ProcessingExceptionHandler.Result#FAIL} status.
         */
        public static Response fail(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new Response(Result.FAIL, deadLetterQueueRecords);
        }

        /**
         * Creates a {@code Response} indicating that processing should fail.
         *
         * @return a {@code Response} with a {@link ProcessingExceptionHandler.Result#FAIL} status.
         */
        public static Response fail() {

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[] source_raw_key();


/**
* 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[] source_raw_value();


. . . 

}

ProductionExceptionHandler.java

Changes:

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,
                                        return fail(Collections.emptyList());
        }

    final Exception exception,
  /**
         * Creates a {@code Response} indicating that processing should continue.
         *
         * @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; finalmay ProcessingContextbe context) {
{@code null}.
        return handle(record, exception);
   }


   /**
    * Handles serialization exception and determine if the process should continue. The default implementation is to
    * fail the process.* @return a {@code Response} with a {@link ProcessingExceptionHandler.Result#RESUME} status.
         */
        public static Response resume(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
    *
    * @param record  return new Response(Result.RESUME, deadLetterQueueRecords);
   the record that failed to serialize
}

        /**
 @param exception     the exception that* occurredCreates duringa serialization
{@code Response} indicating that *processing @deprecatedshould continue.
         Please*
 use the handleSerializationException(record, exception, context)
    */
 @return a @Deprecated
{@code Response} with defaulta ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
{@link ProcessingExceptionHandler.Result#RESUME} status.
         */
        public static Response resume() {
            return resume(Collections.emptyList());
        }

        /**
         * Retrieves the processing handler result.
         *
   final Exception exception) {
   * @return the {@link return ProductionExceptionHandlerResponse.FAIL;
   }


   /**Result} indicating whether processing should continue or fail.
    *  Handles serialization exception and*/
 determine if the process should continue. The defaultpublic implementationResult isresult() to{
    * fail the process.
    *
  return result;
 * @param record     }

   the record that failed to serialize/**
    * @param exception   * Retrieves thean exceptionunmodifiable thatlist occurredof duringrecords serialization
to be sent to *the @paramdead context   letter queue.
    Processing context
    */ <p>
   @SuppressWarnings("deprecation")
   default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord* record,
If the list is {@code null}, an empty list is returned.
         * </p>
         *
         * @return an unmodifiable list of {@link ProducerRecord} instances
         *         for the dead letter queue, or an empty list if no finalrecords Exceptionare exception,available.
         */
        public List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
            if (deadLetterQueueRecords == null) {
                return Collections.emptyList();
            }
           final ProcessingContext context
return Collections.unmodifiableList(deadLetterQueueRecords);
    ) {
   }
    return handleSerializationException(record, exception);
   }


   enum ProductionExceptionHandlerResponse}
 }

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 {
    . . . /*  continue processing */
       
   CONTINUE(0, "CONTINUE"),
   /**
     /* failReturn processing */
       FAIL(1, "FAIL");


       /**the non-deserialized byte[] of the input message key if the context has been triggered by a message.
     *
   * an english* description<p> of the api--thisIf this method is forinvoked debuggingwithin anda can change{@link Punctuator#punctuate(long)
     * punctuation  */
       public final String name;


callback}, or while processing a record that was forwarded by a punctuation
     * callback, it /**
will return {@code null}.
     *
 the permanent and immutable id* of<p> anIf API--this can'tmethod changeis ever
invoked in       */
       public final int id;


a sub-topology due to a repartition, the returned key would be one sent
     * to public ProducerRecord<byte[], byte[]> deadLetterQueueRecord;

the repartition topic.
       ProductionExceptionHandlerResponse(final int id,
*
     * @return the raw byte of the key of the source message
     */
    byte[] sourceRawKey();


    /**
     * Return the non-deserialized byte[] of the input message finalvalue Stringif name)the {
context has been triggered by a message.
     this.id*
 = id;
   * <p> If this method is invoked within this.name = name;a {@link Punctuator#punctuate(long)
     * punctuation callback}


, or while processing a record that publicwas ProductionExceptionHandlerResponse withDeadLetterQueueRecord(ProducerRecord<byte[], byte[]> deadLetterQueueRecord) {forwarded by a punctuation
     * callback, it will return  this.deadLetterQueueRecord = deadLetterQueueRecord;
{@code null}.
     *
     * <p> returnIf this;
 method is invoked in a sub-topology }
due to  }
}

DeserializationExceptionHandler.java

Changes:

Adding the public ProducerRecord<byte[], byte[]> deadLetterQueueRecord; attribute in the ProductionExceptionHandlerResponse

Code Block

language	java

public interface DeserializationExceptionHandler extends Configurable {


   /**
a repartition, the returned key would be one sent
     * Inspect a record andto the exceptionrepartition receivedtopic.
     *
 <p>
    * Note,@return thatthe theraw passedbyte inof {@linkthe ProcessorContext}value onlyof allowsthe tosource accessmessage
 metadata like the task ID.*/
    byte[] sourceRawValue();
       . . .
}

ErrorHandlerContext.java

Changes:

Adding the public byte[] sourceRawKey and byte[] sourceRawValue in the ErrorHandlerContext pointing to the source record data

Code Block

language	java
title	ErrorHandlerContext.java

/**
 * ErrorHandlerContext interface
 */
public interface ErrorHandlerContext {
    . . .
     
    /* * 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.
    *
    * @paramReturn contextthe processor context
    * @param record record that failed deserialization
    * @param exception the actual exception
non-deserialized byte[] of the input message key if the context has been triggered by a message.
     */
    @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
   DeserializationHandlerResponse handle(final ProcessorContext context, * <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 {@code null}.
     *
     *  final ConsumerRecord<byte[], byte[]> record,
                <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 finalthis Exception exception);


   /**
method is invoked within a
     * Enumeration that describes the response from the exception handler.{@link ProductionExceptionHandler.handle(ErrorHandlerContext, ProducerRecord, Exception)}
     *
     */
 @return the enumraw DeserializationHandlerResponsebyte {
of the key of the source message
 /* continue with processing */
     byte[]  CONTINUE(0, "CONTINUE"),sourceRawKey();
 
      /* fail the processing and stop */
     * Return FAIL(1, "FAIL");


       /** an english description of the api--this is for debugging and can change */
       public final String name;


  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)
     /** the permanent and immutable id of an API--this can't change ever */ punctuation callback}, or while processing a record that was forwarded by a punctuation
     * callback, it publicwill finalreturn int id;


{@code null}.
     *
    public ProducerRecord<byte[], byte[]> deadLetterQueueRecord;


       DeserializationHandlerResponse(final int id, final String name) {
   * <p> If this method is invoked in a sub-topology due to a repartition, the returned value would be one sent
     * to the repartition thistopic.id
 = id;
   *
     * <p> Always this.namereturns =null name;
if this method is invoked within  }
a

     *  public DeserializationHandlerResponse withDeadLetterQueueRecord(ProducerRecord<byte[], byte[]> deadLetterQueueRecord) {
{@link ProductionExceptionHandler.handle(ErrorHandlerContext, ProducerRecord, Exception)}
     *
     * @return this.deadLetterQueueRecordthe =raw deadLetterQueueRecord;
byte of the value of the source message
    return this;*/
     byte[] sourceRawValue();
 }
    . . }
}

ProcessingExceptionHandler

...

.
}

Compatibility, Deprecation, and Migration Plan

...

All other changes are backward compatible and should not impact existing applications.
...
Tests to ensure the backward compatibility of the ProductionExceptionHandler class
Tests to ensure that default exception handlers are sending record to the DLQ topic if the DLQ topic name is set
Ensure that failure to send the DLQ record kills the StreamThread
Ensure that punctuator triggered exceptions are producing the expected payload

Rejected Alternatives

...

Managing DeadLetterQueue directly in the DSL by extending the KStreams interface.
Providing no default implementation to build the Dead letter queue record and delegating this task to the user.
Only providing exception and metadata information in the default DLQ implementation.
Adding a new interface, that could be overload overloaded by the user, to build the DLQ record.records.
Adding dead letter queue records to the enum: problematic because the collection would be shared by all stream threads, leading to unnecessary concurrency and potential transaction issues.

Space shortcuts

Child pages

Versions Compared

Old Version 5

New Version Current

Key

Status

Default Dead letter queue record

Public Interfaces

StreamsConfig.java

Default Dead letter queue topic

Public Interfaces

StreamsConfig.java

ProductionExceptionHandler.java

DeserializationExceptionHandler.java

ProcessingExceptionHandler.java

ProcessingContext.java

ProductionExceptionHandler.java

RecordContext.java

DeserializationExceptionHandler.java

ErrorHandlerContext.java

ProcessingExceptionHandler

Compatibility, Deprecation, and Migration Plan

Rejected Alternatives

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 5

New Version Current

Key

Status

Default Dead letter queue record

Public Interfaces

StreamsConfig.java

Default Dead letter queue topic

Public Interfaces

StreamsConfig.java

ProductionExceptionHandler.java

DeserializationExceptionHandler.java

ProcessingExceptionHandler.java

ProcessingContext.java

ProductionExceptionHandler.java

RecordContext.java

DeserializationExceptionHandler.java

ErrorHandlerContext.java

ProcessingExceptionHandler

Compatibility, Deprecation, and Migration Plan

Rejected Alternatives