public static final String ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG = "errors.deadletterqueue.topic.name";

.define(ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG, // required with no default value
       Type.STRING,
       null, /* default */
       Importance.HIGH,

       ERRORS_DEADLETTERQUEUE_TOPIC_NAME_DOC)

public interface ProcessorContextProcessingContext {

. . . 

/**
* Return the non-deserialized byte[] of the input message key if the context has been triggered by a message.
*
* <p> If this method is invoked within a {@link Punctuator#punctuate(long)
* punctuation callback}, or while processing a record that was forwarded by a punctuation
* callback, it will return null.
*
* <p> If this method is invoked in a sub-topology due to a repartition, the returned key would be one sent
* to the repartition topic.
*
* @return the raw byte of the key of the source message
*/
byte[] source_raw_key();


/**
* Return the non-deserialized byte[] of the input message value if the context has been triggered by a message.
*
* <p> If this method is invoked within a {@link Punctuator#punctuate(long)
* punctuation callback}, or while processing a record that was forwarded by a punctuation
* callback, it will return null.
*
* <p> If this method is invoked in a sub-topology due to a repartition, the returned value would be one sent
* to the repartition topic.
*
* @return the raw byte of the value of the source message
*/
byte[] source_raw_value();


. . . 

}

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 ProcessorContextProcessingContext 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;
       }
   }
}