Kafka-16505: Adding dead letter queue in Kafka Streams #17942

Dabz · 2024-11-25T16:04:50Z

First Pull Request to implement KIP-1034 to add support of Dead Letter Queue in Kafka Streams.
A second PR will store the source record key and value byte[] in the context to add it to the DLQ records.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

cadonna

Thanks for the PR, @Dabz !

I started to review it, but did not finish since I need to think about how this KIP uses mutable enums which I guess is an anti-pattern.
Nevertheless, I wanted to give my first comments.

cadonna · 2024-11-27T12:57:48Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

@@ -588,6 +588,10 @@ public class StreamsConfig extends AbstractConfig {
    public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
        " The cluster must have a client metrics subscription which corresponds to a client.";

+    public static final String ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG = "errors.dead.letter.queue.topic.name";


Could you please add unit tests?

cadonna · 2024-11-27T13:26:07Z

streams/src/main/java/org/apache/kafka/streams/errors/CommonExceptionHandler.java

You cannot add a class to the public API that was not mentioned in the KIP.

I renamed this class to ExceptionHandlerUtils, make it packaged protected, and made all methods statics to avoid inheritance.

cadonna · 2024-11-27T13:26:49Z

streams/src/main/java/org/apache/kafka/streams/errors/CommonExceptionHandler.java

+    public static final String HEADER_ERRORS_OFFSET_NAME = "__streams.errors.offset";
+
+
+    public boolean shouldBuildDeadLetterQueueRecord() {


Why is this public? It is only used within this class.

Changed it to package protected

cadonna · 2024-11-27T14:30:42Z

streams/src/main/java/org/apache/kafka/streams/errors/DeserializationExceptionHandler.java

+            return this;
+        }
+
+        public List<ProducerRecord<byte[], byte[]>> drainDeadLetterQueueRecords() {


According to the KIP this should be named deadLetterQueueReords().

Good point, I renamed it

cadonna · 2024-11-27T14:35:12Z

streams/src/main/java/org/apache/kafka/streams/errors/CommonExceptionHandler.java

+/**
+ * {@code CommonExceptionHandler} Contains utilities method that could be used by all exception handlers
+ */
+public class CommonExceptionHandler implements Configurable {


It would be better to define this class as a immutable utility class in the internal package that provides some utility methods to the default exception handlers. That means, no inheritance.

lucasbru · 2024-12-03T08:57:01Z

Indeed - using a single collection of dead letter records here that will shared across all stream threads doesn't seem like the right approach to me. Was this intended?

I'm not sure if there is a way to replace the enum by a class of some kind where we can properly define a non-static collection of records and remain binary compatible.

It seems to me that this needs a new design proposal and an updated KIP?

cadonna · 2024-12-03T12:45:04Z

An option to fix this might be to add a new method deadLetterQueueRecords() to the handler interface that returns a list of records to add to the dead letter queue. By default the list is empty.

cadonna

Thanks for the updates @sebastienviale !

I did a first pass. Here my comments.

cadonna · 2025-01-16T13:39:26Z

streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java

-        return handle(record, exception);
+        throw new UnsupportedOperationException();


Shouldn't this still call the other deprecated method?
Imagine a user implemented

handle(final ProducerRecord<byte[], byte[]> record, final Exception exception)

but not

ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context, final ProducerRecord<byte[], byte[]> record, final Exception exception)

Streams would throw an UnsupportedOperationException although it did not before upgrading to this version.

Maybe it would be beneficial to add some unit tests that verify this redirection. With such unit tests, this removal would had happened without some thoughts about why the test failed.

indeed, I added a Unit Test that failed if an UnsupportedOperationException is thrown

cadonna · 2025-01-16T13:54:18Z

streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java

+    /**
+     * Represents the result of handling a production exception.
+     * <p>
+     * The {@code Response} class encapsulates a {@link ProductionExceptionHandlerResponse},


Suggested change

* The {@code Response} class encapsulates a {@link ProductionExceptionHandlerResponse},

* The {@code Response} class encapsulates a {@link Result},

cadonna · 2025-01-16T13:58:56Z

streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java

+     * {@link ProducerRecord} instances to be sent to a dead letter queue.
+     * </p>
+     */
+    class Response {


Could you please add some unit tests for this class?

cadonna · 2025-01-16T14:14:27Z

streams/src/main/java/org/apache/kafka/streams/errors/ExceptionHandlerUtils.java

+                                                                            final ErrorHandlerContext context,
+                                                                            final Exception e) {
+        if (deadLetterQueueTopicName == null) {
+            throw new InvalidConfigurationException(String.format("%s can not be null while building DeadLetterQueue record", StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG));


Suggested change

throw new InvalidConfigurationException(String.format("%s can not be null while building DeadLetterQueue record", StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG));

throw new InvalidConfigurationException(String.format("%s cannot be null while building dead letter queue record", StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG));

cadonna · 2025-01-16T14:14:52Z

streams/src/main/java/org/apache/kafka/streams/errors/ExceptionHandlerUtils.java

+    static ProducerRecord<byte[], byte[]> buildDeadLetterQueueRecord(final String deadLetterQueueTopicName,
+                                                                            final byte[] key,
+                                                                            final byte[] value,
+                                                                            final ErrorHandlerContext context,
+                                                                            final Exception e) {


Suggested change

static ProducerRecord<byte[], byte[]> buildDeadLetterQueueRecord(final String deadLetterQueueTopicName,

final byte[] key,

final byte[] value,

final ErrorHandlerContext context,

final Exception e) {

static ProducerRecord<byte[], byte[]> buildDeadLetterQueueRecord(final String deadLetterQueueTopicName,

final byte[] key,

final byte[] value,

final ErrorHandlerContext context,

final Exception e) {

cadonna · 2025-01-16T14:48:13Z

streams/src/main/java/org/apache/kafka/streams/errors/LogAndContinueExceptionHandler.java

@@ -50,6 +54,8 @@ public DeserializationHandlerResponse handle(final ProcessorContext context,
        return DeserializationHandlerResponse.CONTINUE;
    }

+    @SuppressWarnings("deprecation")
+    @Deprecated
    @Override
    public DeserializationHandlerResponse handle(final ErrorHandlerContext context,


You can remove those deprecated handler methods. They are not called anywhere in the Streams code.

cadonna · 2025-01-16T14:48:57Z

.../src/main/java/org/apache/kafka/streams/errors/LogAndContinueProcessingExceptionHandler.java

    @Override
-    public ProcessingHandlerResponse handle(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {
+    public ProcessingHandlerResponse handle(final ErrorHandlerContext context,


You can remove this deprecated handler method. It is not called anywhere in the Streams code.

cadonna · 2025-01-16T14:49:32Z

streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailExceptionHandler.java

+    public Response handleError(final ErrorHandlerContext context,
+                                final ConsumerRecord<byte[], byte[]> record,
+                                final Exception exception) {
+        log.warn(


This should be log.error(). It is failing, not resuming.

cadonna · 2025-01-16T14:49:53Z

streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailExceptionHandler.java

@@ -50,6 +54,8 @@ public DeserializationHandlerResponse handle(final ProcessorContext context,
        return DeserializationHandlerResponse.FAIL;
    }

+    @SuppressWarnings("deprecation")
+    @Deprecated
    @Override
    public DeserializationHandlerResponse handle(final ErrorHandlerContext context,


You can remove those deprecated handler methods. They are not called anywhere in the Streams code.

cadonna · 2025-01-16T14:50:32Z

streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailProcessingExceptionHandler.java

+    public ProcessingHandlerResponse handle(final ErrorHandlerContext context,
+                                            final Record<?, ?> record, final Exception exception) {


You can remove this deprecated handler method. It is not called anywhere in the Streams code.

sebastienviale · 2025-01-24T15:22:46Z

Thanks for the updates @sebastienviale !

I did a first pass. Here my comments.

@cadonna thanks for your review, I pushed changes

github-actions bot added the streams label Nov 25, 2024

Dabz force-pushed the KAFKA-16505-Dead-letter-queue-in-Kafka-Streams branch from 4fc7717 to 5698e60 Compare November 25, 2024 16:15

cadonna reviewed Nov 27, 2024

View reviewed changes

cadonna reviewed Jan 16, 2025

View reviewed changes

Dabz and others added 10 commits January 17, 2025 10:21

WIP Implementing KIP-1034 - adding new interfaces

38cc5b8

KAFKA-16505 Adding test for DLQ support

d0ff13a

KAFKA-16505 Adding test for DLQ support

bbb538c

KAFKA-16505: add source raw bytes in processorContex

c304dc2

Implementing KIP-1034 - adding new interfaces

914ccfb

KIP-1034 - adding new interfaces

01d3b10

KAFKA-16505 fixing unit test

2c0cb48

KAFKA-16505 simplifying default implementations

ca6196c

KAFKA-16505 deprecated old handler response enums

c0a508a

KAFKA-16505

93c831f

sebastienviale force-pushed the KAFKA-16505-Dead-letter-queue-in-Kafka-Streams branch from 93cd94d to 93c831f Compare January 17, 2025 09:27

sebastienviale added 2 commits January 17, 2025 16:08

KAFKA-16505 remove unused methods and add tests

40e950f

KAFKA-16505

fc4ac3c

		public static final String HEADER_ERRORS_OFFSET_NAME = "__streams.errors.offset";


		public boolean shouldBuildDeadLetterQueueRecord() {

		return handle(record, exception);
		throw new UnsupportedOperationException();

	* The {@code Response} class encapsulates a {@link ProductionExceptionHandlerResponse},
	* The {@code Response} class encapsulates a {@link Result},

	throw new InvalidConfigurationException(String.format("%s can not be null while building DeadLetterQueue record", StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG));
	throw new InvalidConfigurationException(String.format("%s cannot be null while building dead letter queue record", StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG));

		public ProcessingHandlerResponse handle(final ErrorHandlerContext context,
		final Record<?, ?> record, final Exception exception) {

Kafka-16505: Adding dead letter queue in Kafka Streams #17942

Are you sure you want to change the base?

Kafka-16505: Adding dead letter queue in Kafka Streams #17942

Conversation

Dabz commented Nov 25, 2024

Committer Checklist (excluded from commit message)

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lucasbru commented Dec 3, 2024

cadonna commented Dec 3, 2024

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sebastienviale commented Jan 24, 2025