Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

  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.
  6. If the DeadLetterQueue record can not be sent to Apache Kafka, the exception would be sent to the Kafka Streams uncaughtExceptionHandler.

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

, null if triggered by

punctuate, "error during

punctuate

"
  • If messages exceed Kafka maximum size "message exceeding Kafka maximum record size"
  • Headers

    Existing context headers are automatically forwarded into the new DLQ record

    Header: __streams.errors.exception

    Header: exception

    Name of the thrown exception

    Header: __streams.errors.stacktrace

    Stacktrace of the thrown exception 

    Header: __streams.errors.message

    Thrown exception message

    Header: __streams.errors.topic

    Source input topic, null if triggered by punctuate

    Header: __streams.errors.partition

    Source input partition, null if triggered by punctuate

    Header: __streams.errors.offset

    Source input offset, null if triggered by punctuate

    Default Dead letter queue topic

    By default, this KIP proposes to have on DLQ topic per Kafka Streams application. This topic would not be automatically created by Kafka Streams.

    The DLQ topic name is set through the configuration DEFAULT_ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG = "errors.deadletterqueue.topic.name". Users can override the default behavior by implementing custom exception handlers to implement a different DLQ topic strategy if required.

    Public Interfaces

    StreamsConfig.java

    ...