You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 2 Next »

This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: Under Discussion

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

KIP inspired by Michelin kstreamplify and coauthored by Damien Gasparina, Loic Greffier and Sebastien Viale.

Kafka Streams does have multiple exception handlers to handle issues while processing messages. Each handler proposes two options: either to log the faulty message and continue processing, or to fail and stop KafkaStreams, the default value.

Both out-of-the-box implementations are not suitable for most use-cases as stopping Kafka Streams due to a single faulty message might be problematic and logging and skipping is at high risk of being missed if the user does not actually check the logs.

Most applications tend to rely on the Dead Letter Queue (DLQ) pattern: in case of an issue, the faulty message that can not be processed is stored in a separate topic. This approach has many advantages:

  1. It is easy to access or replay faulty messages.
  2. It is easy to configure an automated notification if messages are produced in the DLQ topic.
  3. Many metadata could be added in the DLQ record, e.g. stacktrace, source topic/partition/offset, etc… 
  4. The DLQ topic could have a different retention than the application log.


DLQ pattern is becoming a standard, it is already available out of the box in Kafka Connect. Many applications I worked with already implemented this pattern in Kafka Streams. Including a DLQ feature directly in Kafka Streams would allow users to configure production-ready error handlers without having to write custom code.

Proposed Changes

To allow users to send a record in the Deal letter queue, a new attribute "deadLetterQueueRecord'' will be added in each exception handler's responses. If this attribute is set, KafkaStreams will send the provided record to Kafka.

A new configuration will be added: errors.deadletterqueue.topic.name. When set, this configuration indicates the default exception handler implementation to build a Dead letter queue record during the error handling.

In order to build a valid Dead letter queue payload, some additional information needs to be captured and forwarded in the processor context: the source message raw key and value.

Storing the raw key and the raw value allows us to send those raw information in the DLQ topic without having to infer the right serializer. All metadata, e.g. Exceptions, StackTrace, topic, partitions and offset would be provided in the record headers by default.

Additionally, the ProcessingContext would need to be available in each ExceptionHandler. It is currently not available in the ProductionExceptionHandler, thus the handle method will need to be overloaded to provide the context and a default implementation needs to be provided to ensure backward compatibility.

If the default values are not suitable for an application, developers could still reimplement the required exception handlers to build custom DLQ records.

This proposal is to:

  1. Capture the initial message key and values bytes and expose them in the ProcessingContext.
  2. Add a new attribute "deadLetterQueueRecord" in the DeserializationHandlerResponse, ProductionExceptionHandlerResponse and ProcessExceptionHandlerResponse (KIP-1033) enum.
  3. Add the processingContext attribute in the ProductionExceptionHandlerResponse.handle method and ensure backward compatibility with previous implementations of the interface.
  4. Add a new attribute public static final String DEFAULT_ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG = "errors.deadletterqueue.topic.name".
  5. Change the existing exception handler to produce a DeadLetterQueue record if the parameter errors.deadletterqueue.topic.name is set.

Default Dead letter queue record

Key

Key of the input message, null if triggered by punctuate

Value

  • If available, contains the value of the input message
  • If triggered by punctuate, "error during punctuate"
  • If messages exceed Kafka maximum size "message exceeding Kafka maximum record size"

Header: exception

Name of the thrown exception

Header: stacktrace

Stacktrace of the thrown exception 

Header: message

Thrown exception message

Header: topic

Source input topic, null if triggered by punctuate

Header: partition

Source input partition, null if triggered by punctuate

Header: offset

Source input offset, null if triggered by punctuate


Public Interfaces

StreamsConfig.java

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)


ProcessingContext.java

public interface ProcessingContext {

. . . 

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


. . . 

}



ProductionExceptionHandler.java

Changes:

  • Adding the ProcessingContext attribute in the handle and handleSerialization methods, ensure the backward compatibility of previous implementations of this handler by providing default implementation. 
  • Adding the public ProducerRecord<byte[], byte[]> deadLetterQueueRecord; attribute in the ProductionExceptionHandlerResponse
  • Deprecate the previous method
/**
* 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 Processor context
    */
   @SuppressWarnings("deprecation")
   default ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
                                                     final Exception exception,
                                                     final ProcessorContext 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       Processor context
    */
   @SuppressWarnings("deprecation")
   default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
                                                                           final Exception exception,
                                                                           final ProcessorContext 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;
       }
   }
}


DeserializationExceptionHandler.java

Changes:

  • Adding the public ProducerRecord<byte[], byte[]> deadLetterQueueRecord; attribute in the ProductionExceptionHandlerResponse 
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 ProcessingContext 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;
       }
   }
}


ProcessingExceptionHandler

With KIP-1033, a similar behavior would be added to the potential new ProcessingExceptionHandler: adding a public ProducerRecord<byte[], byte[]> deadLetterQueueRecord; attribute in the ProcessingExceptionHandlerResponse 


Compatibility, Deprecation, and Migration Plan

To build a valid record for the DeadLetterQueue, the ProductionExceptionHandler.handle method needs to have access to the ProcessorContext. To ensure backward compatibility, the previous interface would be deprecated and the default implementation of the new interface would invoke the previous one.

All other changes are backward compatible and should not impact existing applications.

Test Plan

  • Tests to ensure the backward compatibility of the ProductionExceptionHandler class
  • Tests to ensure that default exception handlers are sending record to the DLQ topic if the DLQ topic name is set
  • Ensure that failure to send the DLQ record kills the StreamThread
  • Ensure that punctuator triggered exceptions are producing the expected payload

Rejected Alternatives

  • 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 overload by the user, to build the DLQ record.


  • No labels