Page History

...

Code Block

language	java

    @Override
    public ProcessingHandlerResponse handlehandleError(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {
      addRecordsToDeadLetterQueue(         
        List<ProducerRecord<byte[], byte[]>> records = Collections.singletonList(new ProducerRecord<>("app-dlq", "Hello".getBytes(StandardCharsets.UTF_8), "World".getBytes(StandardCharsets.UTF_8)));
       );      
      return ProcessingHandlerResponse.CONTINUE;
    ProcessingExceptionResponse.continueProcessing(records);     }

ProductionExceptionHandler.java

Changes:

Adding the public getter ProducerRecord<byte[], byte[]> deadLetterQueueRecord() in the ProductionExceptionHandlerResponse as default implementationAdding the method addRecordsToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) to request records to be produced as default implementationa ProductionExceptionResponse nested class,that contains a ProductionExceptionHandlerResponse 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 ProductionExceptionResponse
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 {
   ...  

    /**
 a list of Kafka records

 to  publish, e.g. in a Dead Letter Queue topic */ @Deprecated
    default ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,
    Queue<ProducerRecord<byte[], byte[]>> deadLetterQueueRecordsQueue = new ConcurrentLinkedQueue<>();

    default void addRecordsToDeadLetterQueue(final Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
        if (deadLetterQueueRecords == null) {
            return;
        }
        for (final ProducerRecord<byte[], byte[]> deadLetterQueueRecordrecord,
 : deadLetterQueueRecords) {
            this.deadLetterQueueRecordsQueue.add(deadLetterQueueRecord);
        }
    }

    default List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
        final LinkedList<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords = new LinkedList<>();
     final Exception  while (trueexception) {
        throw new UnsupportedOperationException();
    }

  final ProducerRecord<byte[], byte[]> record = this.deadLetterQueueRecordsQueue.poll();
            if (record == null) { /**
     * Inspect a record that we attempted to produce, and the exception that resulted
     * from attempting to produce it and determine to continue  break;or stop processing.
     *
     * @param }context
     *     The error handler deadLetterQueueRecords.add(record);context metadata.
     * @param  }record
     *   return deadLetterQueueRecords;
 The record that }

failed to produce.
     * @param ...  }

DeserializationExceptionHandler.java

Changes:

Adding the public getter ProducerRecord<byte[], byte[]> deadLetterQueueRecord() in the DeserializationExceptionHandlerResponse as default implementation
Adding the method addRecordsToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) to request records to be produced as default implementation

exception
     *     The exception that occurred during production.
     *
     * @return a {@link ProductionExceptionResponse} object
     */
    default ProductionExceptionResponse handleError(final ErrorHandlerContext context,
                                                    final ProducerRecord<byte[], byte[]> record,
                                                    final Exception exception) {
        final ProductionExceptionHandlerResponse response =  handle(context, record, exception);
        if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL == response) {
            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) {
        return ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL;
    }

    /**
     * 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 ProductionExceptionResponse} object
     */     
    default ProductionExceptionResponse handleSerializationError(final ErrorHandlerContext context,
                                                                 final ProducerRecord record,
                                                                 final Exception exception,
                                                                 final SerializationExceptionOrigin origin) {
        final ProductionExceptionHandlerResponse response =  handleSerializationException(context, record, exception, origin);
        if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.FAIL == response) {
            return ProductionExceptionResponse.failProcessing();
        } else if (ProductionExceptionHandler.ProductionExceptionHandlerResponse.RETRY == response) {
            return ProductionExceptionResponse.retryProcessing();
        }
        return ProductionExceptionResponse.continueProcessing();
    }

    ...

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

        private ProductionExceptionHandlerResponse productionExceptionHandlerResponse;

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

        /**
         * Constructs a new {@code ProductionExceptionResponse} object.
         *
         * @param productionExceptionHandlerResponse the response 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 ProductionExceptionResponse(final ProductionExceptionHandlerResponse productionExceptionHandlerResponse,
                                            final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            this.productionExceptionHandlerResponse = productionExceptionHandlerResponse;
            this.deadLetterQueueRecords = deadLetterQueueRecords;
        }

        /**
         * Creates a {@code ProductionExceptionResponse} 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 ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#FAIL} status.
         */
        public static ProductionExceptionResponse failProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new ProductionExceptionResponse(ProductionExceptionHandlerResponse.FAIL, deadLetterQueueRecords);
        }

        /**
         * Creates a {@code ProductionExceptionResponse} indicating that processing should fail.
         *
         * @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#FAIL} status.
         */
        public static ProductionExceptionResponse failProcessing() {
            return new ProductionExceptionResponse(ProductionExceptionHandlerResponse.FAIL, Collections.emptyList());
        }

        /**
         * Creates a {@code ProductionExceptionResponse} 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 ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
         */
        public static ProductionExceptionResponse continueProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new ProductionExceptionResponse(ProductionExceptionHandlerResponse.CONTINUE, deadLetterQueueRecords);
        }

        /**
         * Creates a {@code ProductionExceptionResponse} indicating that processing should continue.
         *
         * @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
         */
        public static ProductionExceptionResponse continueProcessing() {
            return new ProductionExceptionResponse(ProductionExceptionHandlerResponse.CONTINUE, Collections.emptyList());
        }

        /**
         * Creates a {@code ProductionExceptionResponse} indicating that processing should retry.
         *
         * @return a {@code ProductionExceptionResponse} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
         */
        public static ProductionExceptionResponse retryProcessing() {
            return new ProductionExceptionResponse(ProductionExceptionHandlerResponse.RETRY, Collections.emptyList());
        }

        /**
         * Retrieves the production exception handler response.
         *
         * @return the {@link ProductionExceptionHandlerResponse} indicating whether processing should continue or fail.
         */
        public ProductionExceptionHandlerResponse response() {
            return productionExceptionHandlerResponse;
        }

        /**
         * 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 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
    default DeserializationHandlerResponse handle(final ErrorHandlerContext context,
                                                  final ConsumerRecord<byte[], byte[]> record,
                                                  final Exception exception) {
        throw new UnsupportedOperationException();
    }

    /**
     * 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 DeserializationExceptionResponse} object
     */
    default DeserializationExceptionResponse handleError(final ErrorHandlerContext context, final ConsumerRecord<byte[], byte[]> record, final Exception exception) {
        if (DeserializationHandlerResponse.FAIL == handle(context, record, exception)) {
            return DeserializationExceptionResponse.failProcessing(Collections.emptyList());
        }
        return DeserializationExceptionResponse.continueProcessing(Collections.emptyList());
    }     

    ...

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

        private DeserializationHandlerResponse deserializationHandlerResponse;

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

        /**
         * Constructs a new {@code DeserializationExceptionResponse} object.
         *
         * @param deserializationHandlerResponse the response 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 DeserializationExceptionResponse(final DeserializationHandlerResponse deserializationHandlerResponse,
                                                 final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            this.deserializationHandlerResponse = deserializationHandlerResponse;
            this.deadLetterQueueRecords = deadLetterQueueRecords;
        }

        /**
         * 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<byte[], byte[]>> deadLetterQueueRecords) {
            return new DeserializationExceptionResponse(DeserializationHandlerResponse.FAIL, deadLetterQueueRecords);
        }

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

        /**
         * Creates a {@code DeserializationExceptionResponse} 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 DeserializationExceptionResponse} with a {@link DeserializationHandlerResponse#CONTINUE} status.
         */
        public static DeserializationExceptionResponse continueProcessing(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new DeserializationExceptionResponse(DeserializationHandlerResponse.CONTINUE, deadLetterQueueRecords);
        }

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

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

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

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

    /**
     * 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 ProcessingExceptionResponse} object
     */
    default ProcessingExceptionResponse handleError(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {
        if (ProcessingHandlerResponse.FAIL == handle(context, record, exception)) {
            return ProcessingExceptionResponse.failProcessing();
        }
        return ProcessingExceptionResponse.continueProcessing();
    }   ...

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

        private ProcessingHandlerResponse processingHandlerResponse;

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

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

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

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

        /**
         * Creates a {@code ProcessingExceptionResponse} 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#CONTINUE} status.
         */
        public static ProcessingExceptionResponse continueProcessing(final List<ProducerRecord<byte

Code Block

language	java

public interface DeserializationExceptionHandler extends Configurable {

   ...

    /** a list of Kafka records to publish, e.g. in a Dead Letter Queue topic */
    Queue<ProducerRecord<byte[], byte[]>> deadLetterQueueRecordsQueue = new ConcurrentLinkedQueue<>();

    default void addRecordsToDeadLetterQueue(final Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
        if (deadLetterQueueRecords == null) {
return new           returnProcessingExceptionResponse(ProcessingHandlerResponse.CONTINUE, deadLetterQueueRecords);
        }

        for (final ProducerRecord<byte[], byte[]> deadLetterQueueRecord : deadLetterQueueRecords) {
/**
         * Creates a {@code ProcessingExceptionResponse} indicating that processing should this.deadLetterQueueRecordsQueue.add(deadLetterQueueRecord);
continue.
         }*
    }

    default List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
        final LinkedList<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords = new LinkedList<>();
* @return a {@code ProcessingExceptionResponse} with a {@link ProcessingHandlerResponse#CONTINUE} status.
         */
        public static whileProcessingExceptionResponse continueProcessing(true) {
            final ProducerRecord<byte[], byte[]> record = this.deadLetterQueueRecordsQueue.pollreturn continueProcessing(Collections.emptyList());
        }

    if (record == null) {/**
         * Retrieves the processing handler response.
  break;
       *
       }
  * @return the {@link ProcessingHandlerResponse} indicating whether processing should continue or deadLetterQueueRecordsfail.add(record);
        }
 */
        public ProcessingHandlerResponse response() {
    return deadLetterQueueRecords;
    }

   ...
  
}

ProcessingExceptionHandler.java

Changes:

Adding the public getter ProducerRecord<byte[], byte[]> deadLetterQueueRecord() in the ProcessingExceptionHandler as default implementation
Adding the method addRecordsToDeadLetterQueue(Iterable<org.apache.kafka.clients.producer.ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) to request records to be produced as default implementation

Code Block

language	java

public interface ProcessingExceptionHandler extends Configurable {

   ...

    Queue<ProducerRecord<byte[], byte[]>> deadLetterQueueRecordsQueue = new ConcurrentLinkedQueue<>();

    default void addRecordsToDeadLetterQueue(Iterable<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
  return processingHandlerResponse;
        }

        /**
         * Retrieves an unmodifiable list of records to be sent to the dead letter queue.
      if (deadLetterQueueRecords == null)* {<p>
         * If the return;
list is {@code null}, an empty list is }returned.
        for (ProducerRecord<byte[], byte[]> record : deadLetterQueueRecords) {
 * </p>
         *
        deadLetterQueueRecordsQueue.add(record);
        } * @return an unmodifiable list of {@link ProducerRecord} instances
    }

     * default List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
    for the dead letter List<ProducerRecord<byte[]queue, byte[]>> deadLetterQueueRecords = new LinkedList<>();
    or an empty list if no records are available.
    while (true) {
   */
        public ProducerRecord<byteList<ProducerRecord<byte[], byte[]> record = deadLetterQueueRecordsQueue.poll>> deadLetterQueueRecords(); {
            if (recorddeadLetterQueueRecords == null) {
                breakreturn Collections.emptyList();
            }
            deadLetterQueueRecordsreturn Collections.addunmodifiableList(recorddeadLetterQueueRecords);
        }
        return deadLetterQueueRecords;
    }

  ...  

}

RecordContext.java

Changes:

...

Space shortcuts

Child pages

Versions Compared

Old Version 28

New Version 29

Key

ProductionExceptionHandler.java

DeserializationExceptionHandler.java

DeserializationExceptionHandler.java

ProcessingExceptionHandler.java

ProcessingExceptionHandler.java

RecordContext.java