Status

Current state: Under Discussion

Discussion thread: https://lists.apache.org/thread/8rcj09wrz98qjlmy4srvt593b776y9yz

JIRA: KAFKA-20604 - Getting issue details... STATUS

Motivation

KIP-1034: Dead letter queue in Kafka Streams added built-in Dead Letter Queue support to Kafka Streams by extending all three exception handler interfaces with a Response class that can carry DLQ records. However, the built-in handler implementations shipped with Kafka Streams are asymmetric:

DefaultProductionExceptionHandler always returns Response.fail()  for both handleError and handleSerializationError. Even when errors.dead.letter.queue.topic.name  is configured, the failed record is sent to the DLQ but the application still shuts down. Users who want to log-and-continue on production or serialization errors must write a custom handler — something that isn't required for deserialization or processing errors.

This gap matters in practice:

• Poison pill records. Long-lived topics accumulate records across schema versions. A single record that fails output serialization — due to an incompatible or deprecated schema — will block the application on every restart, creating a crash loop that requires manual intervention to resolve.
• Schema evolution mismatches. A record may deserialize fine but fail to serialize for the output topic if the output schema has evolved independently.
• Record size violations. A record that processes successfully may exceed max.message.bytes after transformation or enrichment.

This need has been raised independently by multiple community projects (e.g., Kstreamplify issue #524)

Public Interfaces

Classes addition/modification

A new built-in ProductionExceptionHandler implementation in package org.apache.kafka.streams.errors:

package org.apache.kafka.streams.errors;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.streams.StreamsConfig;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

import static org.apache.kafka.streams.errors.internals.ExceptionHandlerUtils.maybeBuildDeadLetterQueueRecords;

/**
 * Production exception handler that logs the error and instructs the
 * processing pipeline to continue processing more records.
 * <p>
 * If a Dead Letter Queue topic is configured via
 * {@link StreamsConfig#ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG},
 * the failed record will be forwarded to the DLQ topic before continuing.
 */
public class LogAndContinueProductionExceptionHandler implements ProductionExceptionHandler {
    private static final Logger log = LoggerFactory.getLogger(LogAndContinueProductionExceptionHandler.class);

    private String deadLetterQueueTopic = null;

    @Override
    public Response handleError(final ErrorHandlerContext context,
                                final ProducerRecord<byte[], byte[]> record,
                                final Exception exception) {
        log.warn(
            "Exception caught during production, taskId: {}, topic: {}, partition: {}, offset: {}",
            context.taskId(),
            context.topic(),
            context.partition(),
            context.offset(),
            exception
        );

        if (exception instanceof RetriableException) {
            return Response.retry();
        }

        return Response.resume(maybeBuildDeadLetterQueueRecords(deadLetterQueueTopic, context.sourceRawKey(), context.sourceRawValue(), context, exception));
    }

    @SuppressWarnings("rawtypes")
    @Override
    public Response handleSerializationError(final ErrorHandlerContext context,
                                             final ProducerRecord record,
                                             final Exception exception,
                                             final SerializationExceptionOrigin origin) {
        log.warn(
            "Exception caught during serialization, taskId: {}, topic: {}, partition: {}, offset: {}",
            context.taskId(),
            context.topic(),
            context.partition(),
            context.offset(),
            exception
        );

        return Response.resume(maybeBuildDeadLetterQueueRecords(deadLetterQueueTopic, context.sourceRawKey(), context.sourceRawValue(), context, exception));
    }

    @Override
    public void configure(final Map<String, ?> configs) {
        if (configs.get(StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG) != null)
            deadLetterQueueTopic = String.valueOf(configs.get(StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG));
    }
}

For naming consistency with the other handler pairs, the existing DefaultProductionExceptionHandler will be renamed to LogAndFailProductionExceptionHandler. The old class will be kept as a deprecated subclass so existing configurations continue to work without changes.

LogAndFailProductionExceptionHandler contains the exact same logic that was previously in DefaultProductionExceptionHandler — it logs at ERROR level and returns Response.fail(). No behavioral change. The only differences compared to the new LogAndContinueProductionExceptionHandler are the log level (error vs warn) and the response type (fail vs resume).

/**
 * @deprecated Since 4.4. Use {@link LogAndFailProductionExceptionHandler} instead.
 */
@Deprecated
public class DefaultProductionExceptionHandler extends LogAndFailProductionExceptionHandler {
}

No new configuration keys

production.exception.handler=org.apache.kafka.streams.errors.LogAndContinueProductionExceptionHandler
errors.dead.letter.queue.topic.name=my-app-dlq

The default value of production.exception.handler remains DefaultProductionExceptionHandler (fail behavior). No existing behavior changes.

Complete handler matrix after this KIP

Proposed Changes

What gets added

A single new class, LogAndContinueProductionExceptionHandler, following the exact same pattern as the existing continue handlers for deserialization and processing. The class:

1. Logs the error at WARN level (including task ID, topic, partition, offset, and the exception)
2. Builds DLQ records via ExceptionHandlerUtils.maybeBuildDeadLetterQueueRecords() if errors.dead.letter.queue.topic.name is configured
3. Returns Response.resume() to skip the failed record and continue processing
4. Returns Response.retry() for RetriableException in handleError (consistent with existing behavior)

Serialization errors vs. production errors

The ProductionExceptionHandler interface covers two distinct error paths. They behave differently and it's worth calling out how each one works with RESUME:

Serialization errors (handleSerializationError): Invoked synchronously on the stream thread. The record fails before it ever reaches the Kafka producer — there's nothing async going on, no transaction impact. Returning RESUME simply skips the record. This is the exact same pattern as LogAndContinueExceptionHandler on the deserialization side and works identically under both ALOS and EOS.

Production errors (handleError): Invoked asynchronously from the producer's callback thread, after the broker rejects a successfully serialized record. Note that the producer sends records in batches — if a batch fails, the handler is called for every record in the batch, not just one. Under ALOS, RESUME is straightforward — no transactions involved, the records are skipped and processing continues. Under EOS (exactly_once_v2), a failed produce can poison the ongoing transaction. Even if the handler returns RESUME, the commit at the end of the interval may fail. When that happens, Kafka Streams throws TaskCorruptedException, the task gets revoked, restarts from the last committed offset, and reprocesses the batch. If the computation is deterministic, the same input records will produce the same bad output record, which will fail again — potentially resulting in a crash-loop rather than eventual recovery. See Known Limitations below.

This is not new behavior — the Result.RESUME enum already exists in trunk, RecordCollectorImpl already handles RESUME for both paths, and custom handlers returning RESUME have always been possible. This KIP simply provides a built-in handler that returns it. The production path under EOS has known limitations that are not introduced or addressed by this KIP (see Known Limitations).

Known Limitations

Production errors under EOS: Under exactly-once semantics, a production error that triggers RESUME may poison the ongoing transaction. The task will restart from the last committed offset and reprocess the same input records. If the computation is deterministic, the same bad output record will be produced again, leading to a repeated cycle of failure → restart → failure. This is an inherent limitation of the RESUME semantics on the production path under EOS, not something introduced by this KIP — custom handlers returning RESUME have always faced this same issue. Prior KIPs attempted to address the EOS production error handling (KIP-1038, KIP-1059) but neither was approved. Users should be aware of this behavior when configuring LogAndContinueProductionExceptionHandler with EOS enabled.

For serialization errors, this limitation does not apply — the error occurs synchronously before the record reaches the producer, so no transaction is affected and RESUME works cleanly under both ALOS and EOS.

DLQ failure behavior

Problem

Currently, DLQ records produced by any exception handler are sent through the same RecordCollectorImpl.send() path as regular records. The send() method registers a producer callback (recordSendError()) that invokes the ProductionExceptionHandler on failure. This creates two problems when a continue handler is configured:

1. Infinite loop in the production path

When handleException() or recordSendError() in RecordCollectorImpl produces a DLQ record via send(), and that DLQ send also fails:

1. Original record fails → handler returns RESUME with DLQ record
2. DLQ record is sent via send() → DLQ produce fails
3. recordSendError() callback invokes handler again → handler returns RESUME with another DLQ record
4. Repeat indefinitely

Today this loop doesn't surface because DefaultProductionExceptionHandler returns FAIL, which sets sendException and breaks the chain after one iteration.

2. Silent data loss in the deserialization and processing paths

DLQ records from RecordDeserializer (deserialization errors) and StreamTask (processing errors) are also sent through collector.send(). If the DLQ produce fails, the callback calls recordSendError(), which invokes the ProductionExceptionHandler. With a continue handler, the handler returns CONTINUE — the DLQ failure is logged as an error and the dropped records metric is bumped, but no sendException is set. The task keeps running. The original bad record was already skipped by the deserialization/processing handler, and the DLQ record is now also lost. The data disappears with only a log line as evidence.

Today this isn't a problem because DefaultProductionExceptionHandler returns FAIL, so the task would crash on DLQ failure. But introducing a built-in continue handler makes this silent data loss easy to trigger.

Current code (before this KIP)

Both handleException() (serialization errors, synchronous) and recordSendError() (production errors, asynchronous callback) use the same pattern to send DLQ records:

// RecordCollectorImpl.java — current code in both handleException() and recordSendError()
final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords = response.deadLetterQueueRecords();
if (!deadLetterQueueRecords.isEmpty()) {
    for (final ProducerRecord<byte[], byte[]> deadLetterQueueRecord : deadLetterQueueRecords) {
        // Uses the same send() as regular records.
        // This registers recordSendError() as the producer callback,
        // which invokes the ProductionExceptionHandler if the DLQ produce fails,
        // creating the loop described above.
        this.send(
                deadLetterQueueRecord.key(),
                deadLetterQueueRecord.value(),
                processorNodeId,
                context,
                deadLetterQueueRecord
        );
    }
}

RecordDeserializer, ProcessorNode and StreamTask use the same pattern to send DLQ records through the normal send() path:

// RecordDeserializer.java — current code in handleDeserializationFailure()
final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords = response.deadLetterQueueRecords();
if (!deadLetterQueueRecords.isEmpty()) {
    final RecordCollector collector = ((RecordCollector.Supplier) processorContext).recordCollector();
    for (final ProducerRecord<byte[], byte[]> deadLetterQueueRecord : deadLetterQueueRecords) {
        collector.send(
                deadLetterQueueRecord.key(),
                deadLetterQueueRecord.value(),
                sourceNodeName,
                (InternalProcessorContext) processorContext,
                deadLetterQueueRecord
        );
    }
}

// StreamTask.java — current code in processing exception handler
final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords = processingExceptionResponse.deadLetterQueueRecords();
if (!deadLetterQueueRecords.isEmpty()) {
    final RecordCollector collector = ((RecordCollector.Supplier) processorContext).recordCollector();
    for (final ProducerRecord<byte[], byte[]> deadLetterQueueRecord : deadLetterQueueRecords) {
        collector.send(
                deadLetterQueueRecord.key(),
                deadLetterQueueRecord.value(),
                node.name(),
                processorContext,
                deadLetterQueueRecord);
    }
}

// ProcessorNode.java — current code in process() exception handler
final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords = response.deadLetterQueueRecords();
if (!deadLetterQueueRecords.isEmpty()) {
    if (!(internalProcessorContext instanceof RecordCollector.Supplier)) {
        log.warn("Dead letter queue records cannot be sent for global store/KTable processors. ...");
    } else {
        final RecordCollector collector = ((RecordCollector.Supplier) internalProcessorContext).recordCollector();
        for (final ProducerRecord<byte[], byte[]> deadLetterQueueRecord : deadLetterQueueRecords) {
            collector.send(deadLetterQueueRecord.key(), deadLetterQueueRecord.value(),
                    name(), internalProcessorContext, deadLetterQueueRecord);
        }
    }
}

Solution

new sendDlqRecord() method on RecordCollector interface

This KIP adds sendDlqRecord() to the RecordCollector interface so all DLQ callers — RecordCollectorImpl, RecordDeserializer, ProcessorNode.process() and StreamTask — can use it:

// RecordCollector.java — new interface method
/**
 * Sends a dead letter queue record directly to the producer, bypassing the production exception handler.
 * If sending to the DLQ fails, the task hard-fails immediately, preventing any infinite loop.
 *
 * @param dlqRecord the record to send to the dead letter queue topic
 */
void sendDlqRecord(ProducerRecord<byte[], byte[]> dlqRecord);

// RecordCollectorImpl.java
@Override
public void sendDlqRecord(final ProducerRecord<byte[], byte[]> dlqRecord) {
    streamsProducer.send(dlqRecord, (metadata, exception) -> {
        if (exception != null) {
            log.error("Failed to send record to dead letter queue topic {}",
                dlqRecord.topic(), exception);
            sendException.compareAndSet(null, new StreamsException(
                String.format("Unable to send record to dead letter queue topic %s",
                    dlqRecord.topic()),
                exception
            ));
        } else if (metadata.offset() >= 0L) {
            log.debug(
                "Successfully sent record to dead letter queue topic {} partition {} offset {}",
                metadata.topic(), metadata.partition(), metadata.offset());
        }
    });
}

All five DLQ call sites are updated to use sendDlqRecord():

// Updated code in RecordCollectorImpl.handleException(), RecordCollectorImpl.recordSendError(),
// RecordDeserializer.handleDeserializationFailure(), StreamTask processing error handler,
// and ProcessorNode.process() processing error handler
final List<ProducerRecord<byte[], byte[]>> deadLetterQueueRecords = response.deadLetterQueueRecords();
if (!deadLetterQueueRecords.isEmpty()) {
    for (final ProducerRecord<byte[], byte[]> deadLetterQueueRecord : deadLetterQueueRecords) {
        sendDlqRecord(deadLetterQueueRecord);  // bypasses exception handler
    }
}

In RecordDeserializer, ProcessorNode and StreamTask, called as collector.sendDlqRecord(deadLetterQueueRecord) via the RecordCollector interface.

Guarantees

This is consistent with current observable behavior — the default handler already returns FAIL for DLQ production failures, so the task already dies today. The change makes the hard-fail explicit and prevents the loop for any handler that returns RESUME.

Future KIPs may introduce more granular control over DLQ failure behavior (e.g., a separate handler or retry policy) if there is demand.

Compatibility, Deprecation, and Migration Plan

• Fully backward compatible. The default production exception handler remains DefaultProductionExceptionHandler with fail behavior. No existing application behavior changes. The internal default class reference in StreamsConfig for both production.exception.handler and default.production.exception.handler is updated from DefaultProductionExceptionHandler.class.getName() to LogAndFailProductionExceptionHandler.class.getName() to avoid deprecation warnings during compilation (-Werror). The runtime behavior is unchanged since DefaultProductionExceptionHandler extends LogAndFailProductionExceptionHandler.
• DefaultProductionExceptionHandler is deprecated in favor of LogAndFailProductionExceptionHandler for naming consistency. The deprecated class extends the new one, so existing configs and code using the old name continue to work. The rename is cosmetic and can be dropped from this KIP if reviewers feel it adds unnecessary scope.
• Migration: Users who want the new behavior change one config value. No code changes, no API migrations.
• Removal timeline: DefaultProductionExceptionHandler would follow the standard deprecation policy — kept for at least two major releases before removal

Test Plan

New Tests

RecordCollectorTest (streams/src/test/.../processor/internals/RecordCollectorTest.java):

• shouldBuildDeadLetterQueueRecordsInLogAndContinueExceptionHandlerDuringDeserialization() — Verifies that LogAndContinueProductionExceptionHandler produces a DLQ record on serialization error and continues processing (no exception thrown).
• shouldBuildDeadLetterQueueRecordsInLogAndContinueExceptionHandler() — Verifies that LogAndContinueProductionExceptionHandler produces a DLQ record on production error (KafkaException) and continues processing.

StreamsConfigTest (streams/src/test/.../StreamsConfigTest.java):

• shouldGetDefaultValueProductionExceptionHandler() — Verifies the default production exception handler is LogAndFailProductionExceptionHandler.

Modified Tests

RecordCollectorTest:

• shouldBuildDeadLetterQueueRecordsInLogAndFailExceptionHandlerDuringDeserialization() — Renamed from ...DefaultExceptionHandler...; now uses LogAndFailProductionExceptionHandler directly.
• shouldBuildDeadLetterQueueRecordsInLogAndFailExceptionHandler() — Renamed from ...DefaultExceptionHandler...; now uses LogAndFailProductionExceptionHandler directly.
• shouldThrowStreamsExceptionUsingDefaultExceptionHandler() — Updated to instantiate LogAndFailProductionExceptionHandler instead of DefaultProductionExceptionHandler.
• Default field productionExceptionHandler changed from DefaultProductionExceptionHandler to LogAndFailProductionExceptionHandler.

• MockRecordCollector (streams/src/test/.../test/MockRecordCollector.java):

  • Added sendDlqRecord() implementation — collects the DLQ record into the collected list, consistent with the existing send() mock behavior.

• KeyValueStoreTestDriver (streams/src/test/.../state/KeyValueStoreTestDriver.java):

  Updated import and instantiation from DefaultProductionExceptionHandler to LogAndFailProductionExceptionHandler for consistency with the rename.

Rejected Alternatives

1. Allow DLQ failures to go through the exception handler

We considered letting DLQ produce failures invoke the exception handler like any other production error. This would give users full control but creates an unbounded loop when a continue handler is configured and the DLQ produce keeps failing (e.g., RecordTooLargeException). Hard-failing on DLQ produce failure is the safer default — if you can't write to your DLQ, something is fundamentally wrong. More granular DLQ failure handling can be explored in a future KIP if there is demand.