Table of Contents

Status

Current state: AcceptedAdopted

Discussion thread: here

JIRA: here

...

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.

...

Add a new getter and setter to configure dead letter queue records in the DeserializationExceptionHandler, ProductionExceptionHandler and ProcessExceptionHandler.
Add a new attribute public static final String ERRORS_DEAD_DEADLETTERQUEUELETTER_QUEUE_TOPIC_NAME_CONFIG = "errors.dead.deadletterqueueletter.queue.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.

...

The DLQ topic name is set through the configuration ERRORS_DEAD_LETTER_DEADLETTERQUEUEQUEUE_TOPIC_NAME_CONFIG = " errors.dead.letter.deadletterqueuequeue.topic.name". Users can override the default behavior by implementing custom exception handlers to implement a different DLQ topic strategy if required.

...

Changes:

Adding the errors.dead.letter.deadletterqueuequeue.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.deadletterqueuequeue.topic.name=null (default), then no records are sent to any dead letter queue
- if errors.dead.letter.deadletterqueuequeue.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_DEADLETTERQUEUEQUEUE_TOPIC_NAME_CONFIG = "errors.dead.deadletterqueueletter.queue.topic.name";

.define(ERRORS_DEAD_DEADLETTERQUEUELETTER_QUEUE_TOPIC_NAME_CONFIG,
       Type.STRING,
       null, /* default */
       Importance.HIGHMEDIUM,
       ERRORS_DEAD_DEADLETTERQUEUELETTER_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 the errors.dead.letter.deadletterqueuequeue.topic.name configuration has no impact.

Code Block

language	java

    @Override
    public ProcessingHandlerResponseResponse 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 ProcessingExceptionResponseProcessingExceptionHandler.Response.continueProcessingresume(records);  
    }

ProductionExceptionHandler.java

Changes:

Adding a ProductionExceptionResponse Response nested class,that contains a ProductionExceptionHandlerResponse 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 ProductionExceptionResponseResponse
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 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) {
        throw new UnsupportedOperationException(); ...
    }

    /**
     * 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 ProductionExceptionResponseResponse} object
     */
    default ProductionExceptionResponseResponse handleError(final ErrorHandlerContext context,
                                                    final ProducerRecord<byte[], byte[]> record,
                                                    final Exception exception) {
                           
       final ProductionExceptionHandlerResponse response =return new Response(Result.from(handle(context, record, exception)), Collections.emptyList());
     }

     if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL == response) {@Deprecated
    default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
            return ProductionExceptionResponse.failProcessing();
        } else if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.RETRY == response) {
            return ProductionExceptionResponse.retryProcessing();
        }
        return ProductionExceptionResponse.continueProcessing();
    }

    @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  final Exception exception) {context metadata.
     *   return ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL;
    }

@param record
     /**
     *The Handlesrecord serializationthat exceptionfailed andto determineserialize.
 if the process should continue.* The@param defaultexception
 implementation is to
  *   * fail theThe process.
exception that occurred during  *serialization.
     * @param contextorigin
     *     The errororigin of handlerthe contextserialization metadataexception.
     * @param record
     * @return    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 ProductionExceptionResponsea {@link Response} object
     */     
    default ProductionExceptionResponseResponse handleSerializationError(final ErrorHandlerContext context,
                                                                 final ProducerRecord record,
                                                                 final Exception exception,
                                                                 final SerializationExceptionOrigin origin) {
                  
   final   ProductionExceptionHandlerResponse response =return new Response(Result.from(handleSerializationException(context, record, exception, origin)), Collections.emptyList());
    }          if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL == response) {
    
    @Deprecated
    enum ProductionExceptionHandlerResponse {
  return ProductionExceptionResponse.failProcessing();
    ...
    }

 else  if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.RETRY == response) {     /**
     * Enumeration that describes the response from the returnexception ProductionExceptionResponsehandler.retryProcessing();
        }*/
    enum Result {
        /** returnResume ProductionExceptionResponseprocessing.continueProcessing();
     }

    ...
*
    /**
     * Represents<p> theFor resultthis ofcase, handlingoutput arecords productionwhich exception.
could not be written successfully *are <p>lost.
     * The {@code Response} class* encapsulatesUse athis {@link ProductionExceptionHandlerResponse},
     * indicating whether processing should continue or fail, along with an optional list of
option only if you can tolerate data loss.
         */
        * {@link ProducerRecord} instances to be sent to a dead letter queue.
 RESUME(0, "RESUME"),
        /** Fail processing.
    * </p>
     */
    class ProductionExceptionResponse {

   * <p> Kafka Streams will privateraise ProductionExceptionHandlerResponse productionExceptionHandlerResponse;

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

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
     * Constructs   a* new(for {@code ProductionExceptionResponse} object@link org.apache.kafka.streams.StreamsConfig#EXACTLY_ONCE_V2 exactly-once}) will be committed.
         */
         * @param productionExceptionHandlerResponse the response indicating whether processing should continue or fail;FAIL(1, "FAIL"),
        /** Retry the failed operation.
         *
         * <p> Retrying might imply that a {@link TaskCorruptedException} exception is thrown, and that the retry
         * is started mustfrom notthe belast {@codecommitted null}offset.
         *
 @param deadLetterQueueRecords    the list of records* to be sent to the dead letter queue; may be {@code null}.<p> <b>NOTE:</b> {@code RETRY} is only a valid return value for
         */
 {@link org.apache.kafka.common.errors.RetriableException retriable exceptions}.
         private* ProductionExceptionResponse(final ProductionExceptionHandlerResponse productionExceptionHandlerResponse,
             If {@code RETRY} is returned for a non-retriable exception it will be interpreted as {@link #FAIL}.
         */
        RETRY(2, "RETRY");

        /**
     final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            this.productionExceptionHandlerResponse = productionExceptionHandlerResponse;   * An english description for the used option. This is for debugging only and may change.
            this.deadLetterQueueRecords = deadLetterQueueRecords;
*/
        public final String }name;

        /**
         * Creates a {@code ProductionExceptionResponse} indicating that processing should fail The permanent and immutable id for the used option. This can't change ever.
         */
        public *final @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
int id;

        Result(final int id, final String name) {
          * @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#FAIL} status.
this.id = id;
            this.name = */name;
        }

 public static ProductionExceptionResponse failProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {/**
         * Converts the returndeprecated newenum ProductionExceptionResponse(ProductionExceptionHandlerResponse.FAIL, deadLetterQueueRecords);
        }

 into the new Result enum.
         /**
         * Creates@param avalue {@code ProductionExceptionResponse} indicating that processing should fail.
the old ProductionExceptionHandlerResponse enum value
         * @return a {@link ProductionExceptionHandler.Result} enum *value
         * @return a {@code ProductionExceptionResponse} with a @throws IllegalArgumentException if the provided value does not map to a valid {@link DeserializationExceptionHandlerProductionExceptionHandler.DeserializationHandlerResponse#FAILResult} status.
         */
        publicprivate static ProductionExceptionResponseProductionExceptionHandler.Result failProcessingfrom(final ProductionExceptionHandlerResponse value) {
            returnswitch failProcessing(Collections.emptyList());
value) {
         }

       case /**FAIL:
         * Creates a {@code ProductionExceptionResponse} indicating that processing should continue.
   return Result.FAIL;
     *
           *case @paramCONTINUE:
 deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.
   return Result.RESUME;
     * @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
  case RETRY:
      */
        public static ProductionExceptionResponse continueProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) { return Result.RETRY;
            return  new ProductionExceptionResponse(ProductionExceptionHandlerResponse.CONTINUE, deadLetterQueueRecords); default:
        }

        /**
    throw new IllegalArgumentException("No Result enum *found Createsfor aold {@code ProductionExceptionResponse} indicating that processing should continue.
value: " + value);
            *}
        }
    }

    /**
 @return a {@code ProductionExceptionResponse} with* aRepresents {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
    the result of handling a production exception.
     */ <p>
     * The {@code publicResponse} staticclass ProductionExceptionResponseencapsulates continueProcessing()a {@link ProductionExceptionHandlerResponse},
     * indicating whether processing should continue or return continueProcessing(Collections.emptyList());
         }

fail, along with an optional list of
     * {@link ProducerRecord} instances /**
to be sent to a dead letter queue.
  * Creates a {@code ProductionExceptionResponse} indicating that processing should retry.* </p>
     */
    class Response {

   *
     private Result result;

  * @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.private List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords;

         /**/
        public static* ProductionExceptionResponse retryProcessing()Constructs a new {
@code Response} object.
         *
 return new ProductionExceptionResponse(ProductionExceptionHandlerResponse.RETRY, Collections.emptyList());
     * @param result }

the result indicating whether processing should continue or /**fail;
         *    Retrieves the production exception handler response.
         *
         * @return the {@link ProductionExceptionHandlerResponse} indicating whether processingmust shouldnot continuebe or{@code failnull}.
         */
 @param deadLetterQueueRecords    the list publicof ProductionExceptionHandlerResponserecords response() {
            return productionExceptionHandlerResponse;
to be sent to the dead letter queue; may be {@code null}.
         */
   }

     private Response(final Result /**result,
         *   Retrieves an unmodifiable list of records to be sent to the dead letter queue.
    final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
     * <p>
      this.result = result;
 * If the list is {@code null}, an empty list is returnedthis.
deadLetterQueueRecords = deadLetterQueueRecords;
       * </p>}

         /**
         * @return an unmodifiable list ofCreates a {@link@code ProducerRecordResponse} instancesindicating that processing should fail.
         *
         * @param deadLetterQueueRecords the list of records to be sent forto the dead letter queue, or an empty list if no records are available; 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 ifnew Response(deadLetterQueueRecords == null) {Result.FAIL, deadLetterQueueRecords);
        }

        return Collections.emptyList();
/**
         * 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.unmodifiableListemptyList(deadLetterQueueRecords));
        }

    }
     ...  
}

DeserializationExceptionHandler.java

Changes:

Adding a DeserializationExceptionResponse nested class,that contains a DeserializationHandlerResponse 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 DeserializationExceptionResponse

Code Block

language	java

public interface DeserializationExceptionHandler extends Configurable {

   ...      
  
    @Deprecated /**
         * Creates a {@code Response} indicating that processing should continue.
    default DeserializationHandlerResponse  handle(final ErrorHandlerContext context,*
         * @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.
      final ConsumerRecord<byte[], byte[]> record, */
        public static Response resume(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new Response(Result.RESUME, deadLetterQueueRecords);
        }

        /**
    final Exception exception) {
    *    throw new UnsupportedOperationException();
    }
Creates a {@code Response} indicating that processing should continue.
    /**
     *
  Inspects a record and the exception received during* deserialization.
@return a {@code Response} with *
a {@link ProductionExceptionHandler.Result#RESUME} status.
  * @param context
     * /
    Error handler context.
   public static *Response @paramresume() record{
     *     Record that failedreturn deserialization.resume(Collections.emptyList());
     *  @param exception}

     *   /**
  The actual exception.
     *
     * @return a {@link DeserializationExceptionResponse} object
 Creates a {@code Response} indicating that processing should retry.
         */
    default DeserializationExceptionResponse handleError(final ErrorHandlerContext context, final ConsumerRecord<byte[], byte[]> record, final Exception exception) {
  * @return a {@code Response} with a {@link ProductionExceptionHandler.Result#RETRY} status.
         */
      if (DeserializationHandlerResponse.FAIL == handle(context, record, exception) public static Response retry() {
            return new DeserializationExceptionResponse.failProcessing(Response(Result.RETRY, Collections.emptyList());
        }

        return DeserializationExceptionResponse.continueProcessing(Collections.emptyList());
    }     

/**
     ...

    /**
 Retrieves the production exception * Represents thehandler result.
 of handling a deserialization exception.
     * <p>
     * The {@code Response} class* encapsulates@return athe {@link ProcessingExceptionHandler.ProcessingHandlerResponse},
     *Result} indicating whether processing should continue, or fail, alongor withretry.
 an optional list of
     */
 {@link ProducerRecord} instances to be sent to apublic deadResult letter queue.result() {
     * </p>
     */
    class DeserializationExceptionResponse {
return result;
        private DeserializationHandlerResponse deserializationHandlerResponse;}

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

/**
         /**
 Retrieves an unmodifiable list of records to be *sent Constructsto athe newdead {@code DeserializationExceptionResponse} objectletter queue.
         * <p>
         * @paramIf deserializationHandlerResponsethe thelist responseis indicating whether processing should continue or fail;{@code null}, an empty list is returned.
         * </p>
         *
         * @return an unmodifiable list of {@link ProducerRecord} instances
       must not be* {@code null}.
       for the *dead @paramletter deadLetterQueueRecordsqueue, or an empty thelist listif ofno records toare beavailable.
 sent to the dead letter queue; may be {@code null}. */
         */
   public List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
     private DeserializationExceptionResponse(final DeserializationHandlerResponse deserializationHandlerResponse,
    if (deadLetterQueueRecords == null) {
                return Collections.emptyList();
            }
            final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            this.deserializationHandlerResponse = deserializationHandlerResponsereturn Collections.unmodifiableList(deadLetterQueueRecords);
            this.deadLetterQueueRecords = 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,
            
        /**
         * Creates a {@code DeserializationExceptionResponse} 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 DeserializationExceptionResponse} with a {@link DeserializationHandlerResponse#FAIL} status.
         */
        public static DeserializationExceptionResponse failProcessing(final List<ProducerRecord<byteConsumerRecord<byte[], byte[]>> deadLetterQueueRecords) {
> record,
                return new DeserializationExceptionResponse(DeserializationHandlerResponse.FAIL, deadLetterQueueRecords);
        }

        /**
         * Creates a {@code DeserializationExceptionResponse} indicating thatfinal processingException shouldexception) fail.{
        ...
    *}

    /**
     * @returnInspects a {@code DeserializationExceptionResponse} with a {@link DeserializationHandlerResponse#FAIL} status.
    record and the exception received during deserialization.
     *
     * @param context
     *     Error handler context.
     * @param record
     *     Record that failed deserialization.
     */
 @param exception
     *     The actual exception.
 public static DeserializationExceptionResponse failProcessing() {*
     * @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 return failProcessing(Collections.emptyList());
        }ProcessingExceptionHandler extends Configurable {

   ...     /**
         * Creates a {@code DeserializationExceptionResponse} indicating that processing should continue.  
    
         *@Deprecated
    default ProcessingHandlerResponse handle(final ErrorHandlerContext context, *final @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.Record<?, ?> record, final Exception exception){
       ...
    };

    /**
     * @returnInspects a {@coderecord DeserializationExceptionResponse}and withthe aexception {@linkreceived DeserializationHandlerResponse#CONTINUE}during statusprocessing.
     *
     */
 @param context
     *     Processing publiccontext metadata.
 static DeserializationExceptionResponse continueProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) { * @param record
     *     Record where returnthe newexception DeserializationExceptionResponse(DeserializationHandlerResponse.CONTINUE, deadLetterQueueRecords);
occurred.
     * @param exception
    }

 *     The actual /**exception.
     *
     * Creates@return a {@code@link DeserializationExceptionResponseResponse} indicatingobject
  that processing should continue.
         *
         * @return a {@code DeserializationExceptionResponse} with a {@link DeserializationHandlerResponse#CONTINUE} status.
         */
        public static DeserializationExceptionResponse continueProcessing() {
      */
    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 {
      return continueProcessing(Collections.emptyList());...
       } }

        /**
     * Enumeration that describes the *response Retrievesfrom the deserializationexception handler response.
         */
    enum Result {
   * @return the {@link DeserializationHandlerResponse} indicating/** whetherResume processing should continue or fail.
         */
        public DeserializationHandlerResponse response() {
 RESUME(1, "RESUME"),
        /** Fail processing. return deserializationHandlerResponse;
*/
         }FAIL(2, "FAIL");

        /**
         * RetrievesAn anenglish unmodifiable list of records to be sent todescription for the deadused letter queueoption.
 This is for debugging only and   * <p>may change.
         */
 If the list is {@code null}, an emptypublic listfinal is returned.String name;

         * </p>/**
         *
 The permanent and      * @return an unmodifiable list of {@link ProducerRecord} instances
    immutable id for the used option. This can't change ever.
     *    */
     for the dead letterpublic queue,final or an empty list if no records are available.
         */
int id;

        Result(final int id, final String name) {
          public List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() { this.id = id;
            if (deadLetterQueueRecords == null) {
this.name = name;
        }

        /**
  return Collections.emptyList();
      * Converts the deprecated enum ProcessingHandlerResponse }
into the new Result enum.
        return Collections.unmodifiableList(deadLetterQueueRecords); *
         * }
@param value the old }DeserializationHandlerResponse  enum 

}

ProcessingExceptionHandler.java

Changes:

Adding a ProcessingExceptionResponse nested class,that contains a ProcessingHandlerResponse 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 ProcessingExceptionResponse

Code Block

language	java

public interface ProcessingExceptionHandler extends Configurable {

   ...   value
         * @return a {@link ProcessingExceptionHandler.Result} enum value
    
    @Deprecated
 * @throws IllegalArgumentException defaultif ProcessingHandlerResponsethe handle(finalprovided ErrorHandlerContextvalue context,does finalnot Record<?, ?> record, final Exception exception){map to a valid {@link ProcessingExceptionHandler.Result}
        throw new UnsupportedOperationException(); */
    };

    /**
private static ProcessingExceptionHandler.Result from(final ProcessingHandlerResponse *value) Inspects{
 a record and the exception received during processing.
     *
switch (value) {
   * @param context
     *     Processing contextcase metadata.FAIL:
     * @param record
     *     Record where the exceptionreturn occurredResult.FAIL;
     * @param exception
     *    case The actual exception.
CONTINUE:
      *
     * @return a {@link ProcessingExceptionResponse} object
     */return Result.RESUME;
    default ProcessingExceptionResponse handleError(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {default:
           if  (ProcessingHandlerResponse.FAIL == handle(context, record, exception)) {
  throw new IllegalArgumentException("No Result enum found for old value: " return ProcessingExceptionResponse.failProcessing(+ value);
            }
        return ProcessingExceptionResponse.continueProcessing();}
    }
     ...

     /**
     * Represents the result of handling a processing exception.
     * <p>
     * The {@code Response} class encapsulates a {@link ProcessingHandlerResponseResult},
     * 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 ProcessingExceptionResponseResponse {

        private ProcessingHandlerResponseResult processingHandlerResponseresult;

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

        /**
         * Constructs a new {@code ProcessingExceptionResponse} object.
         *
         * @param processingHandlerResponseresult the responseresult 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 ProcessingExceptionResponseResponse(final ProcessingHandlerResponseResult processingHandlerResponseresult,
                                            final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            this.processingHandlerResponseresult = processingHandlerResponseresult;
            this.deadLetterQueueRecords = deadLetterQueueRecords;
        }

        /**
         * Creates a {@code ProcessingExceptionResponseResponse} 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 ProcessingExceptionResponseResponse} with a {@link ProcessingHandlerResponse#FAILProcessingExceptionHandler.Result#FAIL} status.
         */
        public static ProcessingExceptionResponseResponse failProcessingfail(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new ProcessingExceptionResponseResponse(ProcessingHandlerResponseResult.FAIL, deadLetterQueueRecords);
        }

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

        /**
         * Creates a {@code ProcessingExceptionResponseResponse} 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 ProcessingHandlerResponse#CONTINUEProcessingExceptionHandler.Result#RESUME} status.
         */
        public static ProcessingExceptionResponseResponse continueProcessingresume(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new ProcessingExceptionResponseResponse(ProcessingHandlerResponseResult.CONTINUERESUME, deadLetterQueueRecords);
        }

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

        /**
         * Retrieves the processing handler responseresult.
         *
         * @return the {@link ProcessingHandlerResponseResult} indicating whether processing should continue or fail.
         */
        public ProcessingHandlerResponseResult responseresult() {
            return processingHandlerResponseresult;
        }

        /**
         * 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);
        }
    }
  

}

RecordContext.java

Changes:

...

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 {@code 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) 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 {@code 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

language	java
title	ErrorHandlerContext.java

/**
 * ErrorHandlerContext interface
 */
public interface ErrorHandlerContext {
    . . .
     
    /**
    * 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 {@code 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
     * {@link 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 {@code 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.
     *
     * <p> Always returns null if this method is invoked within a
     * {@link ProductionExceptionHandler.handle(ErrorHandlerContext, ProducerRecord, Exception)}
     *
     * @return the raw byte of the value of the source message
     */
     byte[] sourceRawValue();
 
    . . .
}

...

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 overloaded by the user, to build the DLQ 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 30

New Version Current

Key

Status

ProductionExceptionHandler.java

DeserializationExceptionHandler.java

DeserializationExceptionHandler.java

ProcessingExceptionHandler.java

ProcessingExceptionHandler.java

RecordContext.java

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 30

New Version Current

Key

Status

ProductionExceptionHandler.java

DeserializationExceptionHandler.java

DeserializationExceptionHandler.java

ProcessingExceptionHandler.java

ProcessingExceptionHandler.java

RecordContext.java