Page History

/**
* Interface that specifies how an exception when attempting to produce a result to
* Kafka should be handled.
*/
public interface ProductionExceptionHandler extends Configurable {
   /**
    * Inspect a record that we attempted to produce, and the exception that resulted
    * from attempting to produce it and determine whether or not to continue processing.
    *
    * @param record The record that failed to produce
    * @param exception The exception that occurred during production
    * @deprecated Please use the ProductionExceptionHandlerResponse.handle(record, exception, context)
    */
   @Deprecated
   ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
                                             final Exception exception);



   /**
    * Inspect a record that we attempted to produce, and the exception that resulted
    * from attempting to produce it and determine whether or not to continue processing.
    *
    * @param record The record that failed to produce
    * @param exception The exception that occurred during production
    * @param context ProcessorProcessing context
    */
   @SuppressWarnings("deprecation")
   default ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
                                                     final Exception exception,
                                                     final ProcessorContextProcessingContext context) {
       return handle(record, exception);
   }


   /**
    * Handles serialization exception and determine if the process should continue. The default implementation is to
    * fail the process.
    *
    * @param record        the record that failed to serialize
    * @param exception     the exception that occurred during serialization
    * @deprecated          Please use the handleSerializationException(record, exception, context)
    */
   @Deprecated
   default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
                                                                           final Exception exception) {
       return ProductionExceptionHandlerResponse.FAIL;
   }


   /**
    * Handles serialization exception and determine if the process should continue. The default implementation is to
    * fail the process.
    *
    * @param record        the record that failed to serialize
    * @param exception     the exception that occurred during serialization
    * @param context       ProcessorProcessing context
    */
   @SuppressWarnings("deprecation")
   default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
                                                                           final Exception exception,
                                                                           final ProcessorContextProcessingContext context
   ) {
       return handleSerializationException(record, exception);
   }


   enum ProductionExceptionHandlerResponse {
       /* continue processing */
       CONTINUE(0, "CONTINUE"),
       /* fail processing */
       FAIL(1, "FAIL");


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


       /**
        * the permanent and immutable id of an API--this can't change ever
        */
       public final int id;


       public ProducerRecord<byte[], byte[]> deadLetterQueueRecord;


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


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

public interface DeserializationExceptionHandler extends Configurable {


   /**
    * Inspect a record and the exception received.
    * <p>
    * Note, that the passed in {@link ProcessorContext} only allows to access metadata like the task ID.
    * However, it cannot be used to emit records via {@link ProcessorContext#forward(Object, Object)};
    * calling {@code forward()} (and some other methods) would result in a runtime exception.
    *
    * @param context processor context
    * @param record record that failed deserialization
    * @param exception the actual exception
    */
   @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
   DeserializationHandlerResponse handle(final ProcessingContextProcessorContext context,
                                         final ConsumerRecord<byte[], byte[]> record,
                                         final Exception exception);


   /**
    * Enumeration that describes the response from the exception handler.
    */
   enum DeserializationHandlerResponse {
       /* continue with processing */
       CONTINUE(0, "CONTINUE"),
       /* fail the processing and stop */
       FAIL(1, "FAIL");


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


       /** the permanent and immutable id of an API--this can't change ever */
       public final int id;


       public ProducerRecord<byte[], byte[]> deadLetterQueueRecord;


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


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