Versions Compared

Old Version 43

changes.mady.by.user Loïc Greffier

Saved on

compared with

New Version Current

changes.mady.by.user Unknown User (chia7712)

Saved on

Key

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

Table of Contents

Status

Current state: AcceptedAdopted

Discussion thread: here

JIRA: here

...

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

...

  1. Add a new getter and setter to configure dead letter queue records in the DeserializationExceptionHandler, ProductionExceptionHandler and ProcessExceptionHandler.
  2. Add a new attribute public static final String ERRORS_DEADLETTERQUEUEDEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG = "errors.dead.letter.deadletterqueuequeue.topic.name".
  3. Change the existing exception handler to produce a DeadLetterQueue record if the parameter errors.deadletterqueuedead.letter.queue.topic.name is set.
  4. If the DeadLetterQueue record can not be sent to Apache Kafka, the exception would be sent to the Kafka Streams uncaughtExceptionHandler.

...

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

...

If the user implements a custom exception handler, it is up to the custom handler to build DLQ records to send, in this case, the the errors.deadletterqueuedead.letter.queue.topic.name configuration has no impact.

...

Code Block
languagejava
titleErrorHandlerContext.java
/**
 * RecordContext interface
 */
public interface RecordContext {
    . . .          
    /**
     * 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 {@code 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[] sourceRawKey();


    /**
     * 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 {@code 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 value of the source message
     */
    byte[] sourceRawValue();
       . . .
}

...

Code Block
languagejava
titleErrorHandlerContext.java
/**
 * ErrorHandlerContext interface
 */
public interface ErrorHandlerContext {
    . . .
     
    /**
    * 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 {@code 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.
     *
     * <p> Always returns null if this method is invoked within a
     * {@link ProductionExceptionHandler.handle(ErrorHandlerContext, ProducerRecord, Exception)}
     *
     * @return the raw byte of the key of the source message
     */
     byte[] sourceRawKey();
 
     /**
     * 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 {@code 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.
     *
     * <p> Always returns null if this method is invoked within a
     * {@link ProductionExceptionHandler.handle(ErrorHandlerContext, ProducerRecord, Exception)}
     *
     * @return the raw byte of the value of the source message
     */
     byte[] sourceRawValue();
 
    . . .
}

...

  • 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 overloaded by the user, to build the DLQ records.
  • Adding dead letter queue records to the enum: problematic because the collection would be shared by all stream threads, leading to unnecessary concurrency and potential transaction issues.