Page History

...

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

    @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) {                  
        return new ProductionExceptionResponseResponse(Result.from(handleSerializationException(context, record, exception, origin)), Collections.emptyList());
    }     
     }

    ...
     @Deprecated

     /**enum ProductionExceptionHandlerResponse {
     * Represents the result of handling a production exception.
...
    }

         /** <p>
     * The {@code Response} class encapsulates a {@link ProductionExceptionHandlerResponse},Enumeration that describes the response from the exception handler.
     */
 indicating whether processing shouldenum continueResult or{
 fail, along with an optional list of
 /** Resume processing.
  * {@link ProducerRecord} instances to be sent to*
 a dead letter queue.
     * </p>
<p> For this case, output */
records which could not classbe Responsewritten {

successfully are lost.
      private ProductionExceptionHandlerResponse productionExceptionHandlerResponse;

 * Use this option only if you private List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords;

can tolerate data loss.
         */**
        RESUME(0, "RESUME"),
    * Constructs a new {@code Response} object/** Fail processing.
         *
         * <p> Kafka @paramStreams productionExceptionHandlerResponsewill theraise responsean indicatingexception whetherand processingthe should{@code continueStreamsThread} orwill 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              must notexactly-once}) will be {@code null}committed.
         */
 @param deadLetterQueueRecords    the list of records to be sent to the dead letter queue; may be {@code null}FAIL(1, "FAIL"),
        /** Retry the failed operation.
         */
        private Response(final* ProductionExceptionHandlerResponse<p> productionExceptionHandlerResponse,
Retrying might imply that a {@link TaskCorruptedException} exception is thrown, and that the retry
         * is started final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
from the last committed offset.
         *
    this.productionExceptionHandlerResponse = productionExceptionHandlerResponse;
   * <p> <b>NOTE:</b> {@code RETRY} is only a valid this.deadLetterQueueRecordsreturn =value deadLetterQueueRecords;for
        }

   * {@link org.apache.kafka.common.errors.RetriableException   /**retriable exceptions}.
         * CreatesIf a {{@code ResponseRETRY} indicatingis thatreturned processingfor should fail.
         *a non-retriable exception it will be interpreted as {@link #FAIL}.
         */
 @param deadLetterQueueRecords the list of records to be sent to the dead letter queue; may be {@code null}.RETRY(2, "RETRY");

        /**
         * @return a {@code Response} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#FAIL} status An english description for the used option. This is for debugging only and may change.
         */
        public staticfinal Response fail(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {String name;

        /**
         * The permanent returnand new Response(ProductionExceptionHandlerResponse.FAIL, deadLetterQueueRecords);
        }

immutable id for the used option. This can't change ever.
         */**
        public *final Creates a {@code Response} indicating that processing should fail.
         *
int id;

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

  public static Response fail() {
  /**
         * return fail(Collections.emptyList());
        }

Converts the deprecated enum ProductionExceptionHandlerResponse into the new Result enum.
         /**
         * Creates@param avalue {@codethe Response}old indicatingProductionExceptionHandlerResponse that processing should continue.enum value
         *
 @return a {@link ProductionExceptionHandler.Result} enum value
   * @param deadLetterQueueRecords the list of records* to@throws beIllegalArgumentException sent to the dead letter queue; may be {@code null}.if the provided value does not map to a valid {@link ProductionExceptionHandler.Result}
         * @return a {@code Response} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status./
         */@Deprecated
        public static ResponseProductionExceptionHandler.Result resumefrom(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecordsProductionExceptionHandlerResponse value) {
            return newswitch Response(ProductionExceptionHandlerResponse.CONTINUE, deadLetterQueueRecords);(value) {
        }

        case /**FAIL:
         * Creates a {@code Response} indicating that processing should continue.
  return Result.FAIL;
      *
         * @returncase aCONTINUE:
 {@code Response} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
         */
    return Result.RESUME;
   public static Response resume() {
         case RETRY:
  return resume(Collections.emptyList());
        }

        /**
 return Result.RETRY;
       * Creates a {@code Response} indicating that processing should retry.default:
         *
         * @return athrow {@code Response} with a {@link DeserializationExceptionHandler.DeserializationHandlerResponse#CONTINUE} status.
         */new IllegalArgumentException("No Result enum found for old value: " + value);
        public static Response retry() {}
            return new Response(ProductionExceptionHandlerResponse.RETRY, Collections.emptyList());
    }
    }

        /**
     * Represents the result *of Retrieveshandling thea production exception handler response.
     *    *<p>
     * The {@code Response} *class @returnencapsulates thea {@link ProductionExceptionHandlerResponse},
     * indicating whether processing should continue or fail.
, along with an optional list of
     */
 {@link ProducerRecord} instances to be sent to publica ProductionExceptionHandlerResponsedead response() {
  letter queue.
     * </p>
     */
    class returnResponse productionExceptionHandlerResponse;{

        private }Result result;

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

        /**
 * Retrieves an unmodifiable list of records to be* sentConstructs toa thenew dead{@code letterResponse} queueobject.
         * <p>
         * @param Ifresult the listresult indicating iswhether {@code null}, an empty list is returned.
processing should continue or fail;
         *            * </p>
         *
         * @return an unmodifiablemust listnot ofbe {@link@code ProducerRecordnull} instances.
         * @param deadLetterQueueRecords    the list of records to be sent forto the dead letter queue,; ormay anbe empty list if no records are available{@code null}.
         */
        private Response(final publicResult List<ProducerRecord<byte[]result,
 byte[]>> deadLetterQueueRecords() {
            if (deadLetterQueueRecords == null         final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            this.result = result;
           return Collections.emptyList()this.deadLetterQueueRecords = deadLetterQueueRecords;
        }

        }
/**
         * Creates a {@code Response} indicating that processing returnshould Collectionsfail.unmodifiableList(deadLetterQueueRecords);
         }*
        }
 *    ...  
}

DeserializationExceptionHandler.java

Changes:

Adding a Response 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,
         @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);
        }

        /**
         * final ConsumerRecord<byte[], byte[]> record,
     Creates a {@code Response} indicating that processing should fail.
         *
         * @return a {@code Response} with a {@link ProductionExceptionHandler.Result#FAIL} status.
         */
        public finalstatic ExceptionResponse exceptionfail() {
        throw new UnsupportedOperationException();
    }

    /**return fail(Collections.emptyList());
     * Inspects a record and the exception received during deserialization.
 }

         /**
     * @param context
  * Creates a *{@code Response} indicating that processing Errorshould handler contextcontinue.
     * @param record
     *
     Record that failed deserialization.
 * @param deadLetterQueueRecords the *list @paramof exception
records to be sent to *the dead letter queue; may Thebe actual{@code exceptionnull}.
     *
     * @return a {@code Response} with a {@link DeserializationExceptionResponseProductionExceptionHandler.Result#RESUME} objectstatus.
         */
      default DeserializationExceptionResponse handleError(finalpublic ErrorHandlerContextstatic context,Response resume(final ConsumerRecord<byteList<ProducerRecord<byte[], byte[]> record, final Exception exception) {        >> deadLetterQueueRecords) {
            return new DeserializationExceptionResponse(handle(context, record, exception),   Collections.emptyList());
    }     

    ...    
Response(Result.RESUME, deadLetterQueueRecords);
        }

        /**
     * Represents the result of* handlingCreates a deserialization exception.
   {@code Response} indicating that processing should continue.
         *
   <p>
      * @return Thea {@code Response} class encapsulateswith a {@link ProcessingExceptionHandlerProductionExceptionHandler.ProcessingHandlerResponseResult#RESUME}, status.
     * indicating whether processing should*/
 continue or fail, along with an optional listpublic of
static Response resume() {
  * {@link ProducerRecord} instances to be sent to a dead letterreturn queue.
resume(Collections.emptyList());
      * </p>  }

        /**/
    class Response {

   * Creates a {@code Response} privateindicating DeserializationHandlerResponse deserializationHandlerResponse;

  that processing should retry.
      private List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords;

 *
         /**
 @return a {@code      * ConstructsResponse} with a new {@code@link DeserializationExceptionResponseProductionExceptionHandler.Result#RETRY} objectstatus.
         */
        public *static @paramResponse deserializationHandlerResponse the response indicating whether processing should continue or fail;
retry() {
            return new Response(Result.RETRY, Collections.emptyList());
  *      }

        /**
         * Retrieves the production exception handler result.
     must not be {@code null}.*
         * @param deadLetterQueueRecords   @return the list{@link ofResult} recordsindicating towhether beprocessing sentshould tocontinue, thefail dead letter queue; may be {@code null}or retry.
         */
        public privateResult Responseresult(final DeserializationHandlerResponse deserializationHandlerResponse,) {
            return result;
        }

    final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {/**
         * Retrieves an this.deserializationHandlerResponseunmodifiable =list deserializationHandlerResponse;
of records to be sent to the dead letter queue.
   this.deadLetterQueueRecords =  deadLetterQueueRecords;
   * <p>
    }

     * If the /**
list is {@code null}, an empty list is returned.
 * Creates a {@code Response} indicating that processing should fail.* </p>
         *
         * @param@return deadLetterQueueRecordsan theunmodifiable list of records to be sent to the dead letter queue; may be {@code null}.
 {@link ProducerRecord} instances
         *         for the dead letter *queue, @returnor aan {@codeempty Response}list withif ano {@linkrecords DeserializationHandlerResponse#FAIL}are statusavailable.
         */
        public static Response fail(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords() {
            returnif new Response(DeserializationHandlerResponse.FAIL, deadLetterQueueRecords);
 ==       }

 null) {
       /**
         * Creates a {@code Response} indicating that processing should fail.return Collections.emptyList();
         *
    }
     * @return a {@code Response} with a {@link DeserializationHandlerResponse#FAIL} status.return Collections.unmodifiableList(deadLetterQueueRecords);
         */}
    }
}

DeserializationExceptionHandler.java

Changes:

Adding a Response 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
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 DeserializationExceptionResponse} object
     */
    default DeserializationExceptionResponse 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}
         */
        @Deprecated
        public 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);
    public static Response fail() {
        }
    return fail(Collections.emptyList());
        }}
}

ProcessingExceptionHandler.java

Changes:

Adding a Response 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
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 {

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

    /**
     * @returnInspects a {@coderecord Response}and withthe aexception {@linkreceived DeserializationHandlerResponse#CONTINUE}during statusprocessing.
         */
     * @param context
 public static Response resume(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
 *     Processing     context metadata.
   return new Response(DeserializationHandlerResponse.CONTINUE, deadLetterQueueRecords);
   * @param record
     * }

    Record where the exception /**occurred.
     * @param exception
   * Creates a* {@code Response} indicating that processingThe shouldactual continueexception.
     *
     *
 @return a {@link ProcessingExceptionResponse} object
     */
    @returndefault a {@code Response} with a {@link DeserializationHandlerResponse#CONTINUE} status.
         */
        public static Response resume() {ProcessingExceptionResponse handleError(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {                          
        return new Response(ProcessingExceptionHandler.Result.from(handle(context,  return resume(record, exception)), Collections.emptyList());
    }   

    }@Deprecated

    enum ProcessingHandlerResponse   /**{
       ...
  * Retrieves the deserialization handler response.} 

    /**
     *
 Enumeration that describes the response from the exception *handler.
 @return the {@link DeserializationHandlerResponse} indicating*/
 whether processing should continueenum orResult fail.{
        /** Resume processing. */
        public DeserializationHandlerResponse response() {RESUME(1, "RESUME"),
        /** Fail   return deserializationHandlerResponse;
processing. */
         }FAIL(2, "FAIL");

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

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

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

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

  }

ProcessingExceptionHandler.java

Changes:

Adding a Response 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){
     * @return a {@link ProcessingExceptionHandler.Result} enum value
         * @throws IllegalArgumentException if the provided value does not map to a valid {@link ProcessingExceptionHandler.Result}
         */
        @Deprecated
        public throwstatic newProcessingExceptionHandler.Result UnsupportedOperationExceptionfrom();
final ProcessingHandlerResponse value) {
        };

     /**switch (value) {
     * Inspects a record and the exception received during processing.
  case FAIL:
  *
     * @param context
     *     Processing contextreturn metadataResult.FAIL;
     * @param record
     *     Record wherecase theCONTINUE:
 exception occurred.
     * @param exception
     *     The actualreturn exceptionResult.RESUME;
      *
     * @return a {@link ProcessingExceptionResponse} object
default:
         */
     default ProcessingExceptionResponse handleError(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {         throw new IllegalArgumentException("No Result enum found for old value: " + value);
        
    }
   return new ProcessingExceptionResponse(handle(context, record, exception), Collections.emptyList()); }
    }   }

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

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

        /**
         * Creates a {@code Response} indicating that processing should fail.
         *
         * @return a {@code Response} with a {@link ProcessingHandlerResponse#FAILProcessingExceptionHandler.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 ProcessingHandlerResponse#CONTINUEProcessingExceptionHandler.Result#RESUME} status.
         */
        public static Response resume(final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords) {
            return new Response(ProcessingHandlerResponseResult.CONTINUERESUME, deadLetterQueueRecords);
        }

        /**
         * Creates a {@code Response} indicating that processing should continue.
         *
         * @return a {@code Response} with a {@link ProcessingHandlerResponse#CONTINUEProcessingExceptionHandler.Result#RESUME} status.
         */
        public static Response resume() {
            return resume(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);
        }
    }
 }

...

Space shortcuts

Child pages

Versions Compared

Old Version 33

New Version 34

Key

DeserializationExceptionHandler.java

DeserializationExceptionHandler.java

ProcessingExceptionHandler.java

ProcessingExceptionHandler.java