Error handling and DLQ

StoatFlow has three independent exception handlers, one per failure stage: a record that fails to deserialize on the way in, a processor that throws while handling a record, and a production failure when serializing or sending an output record. Each handler decides whether to skip the record and continue, stop the application, or route the record to a dead-letter queue (DLQ) — and each is wired the same way, in code via streamsConfigOverrides or in application.yaml by class name.

For the design rationale — why these three stages, when "continue" is safe, and how DLQ records ride the same commit barrier as your output — see Error-handling model.

The surface follows Kafka Streams' KIP-1033 (processing exception handler, Kafka 3.9.0) and KIP-1034 (dead letter queue, Kafka 4.2.0), so it should look familiar if you have used those. Both were coauthored by Damien Gasparina, Loïc Greffier and Sébastien Viale at Michelin, and grew out of Michelin's Kstreamplify; their Current London 2025 talk Processing Exception Handling and Dead Letter Queue in Kafka Streams and Michelin's write-ups on processing error handling and the DLQ in Spring Kafka 4.1 are the best background.

The three handler stages

Each handler returns a decision:

• CONTINUE — skip the failed record and keep processing. Optionally emit one or more DLQ records. On the asynchronous production path below it skips the record across an epoch replay rather than in place, but the effect is the same: the failed record is skipped, everything else is processed.
• FAIL — stop the application. The in-flight epoch's work is discarded and the process exits; see Architecture for the recovery behaviour on restart. Optionally emit DLQ records before failing. The source offsets are not advanced on any path, including the asynchronous production one, so the epoch replays on restart — FAIL means stop before further harm.
• RETRY — production handlers only. On a synchronous send the runtime retries in place with exponential backoff, and writes any attached DLQ record only once those attempts are exhausted. On the asynchronous production path below the producer's own retries are already spent, so RETRY aborts and replays the epoch instead: identical to CONTINUE except the record is not quarantined, because the verdict says the record is fine and the send wasn't. Under AT_LEAST_ONCE there is no transaction to abort and no replay to offer, so RETRY on that path is treated as FAIL.

Built-in handlers

All built-in handlers live in io.stoatflow.core.exception.

DefaultProductionExceptionHandler is the only built-in production handler; it covers retry-vs-fail classification and optional DLQ routing in one place, so you configure behaviour by whether you pass it a dlqTopic rather than by swapping the class.

Configuring a handler in code

Set handlers through streamsConfigOverrides on the :runtime builder. The same builder methods exist on StreamsConfig.builder(...) if you run on :core directly.

This example routes deserialization failures to a DLQ topic (skip + capture), keeps the fail-fast processing default, and adds a DLQ topic for production failures:

import io.stoatflow.core.exception.DeadLetterQueueDeserializationExceptionHandler
import io.stoatflow.core.exception.DefaultProductionExceptionHandler
import io.stoatflow.runtime.StoatFlowRuntime
import org.apache.kafka.common.serialization.Serdes

val runtime = StoatFlowRuntime.fromConfig(
    topologyBuilder = { buildTopology(it) },
    configure = {
        streamsConfigOverrides {
            defaultKeySerde(Serdes.String())
            defaultValueSerde(Serdes.String())

            deserializationExceptionHandler(
                DeadLetterQueueDeserializationExceptionHandler(dlqTopic = "my-app.deserialization-errors.dlq"),
            )
            productionExceptionHandler(
                DefaultProductionExceptionHandler(dlqTopic = "my-app.production-errors.dlq"),
            )
        }
    },
)
runtime.start()
runtime.awaitTermination()

import io.stoatflow.core.exception.DeadLetterQueueDeserializationExceptionHandler;
import io.stoatflow.core.exception.DefaultProductionExceptionHandler;
import io.stoatflow.runtime.StoatFlowRuntime;
import org.apache.kafka.common.serialization.Serdes;

var runtime = StoatFlowRuntime.fromConfig(
    Main::buildTopology,
    builder -> builder.streamsConfigOverrides(cfg -> {
        cfg.defaultKeySerde(Serdes.String());
        cfg.defaultValueSerde(Serdes.String());

        cfg.deserializationExceptionHandler(
            new DeadLetterQueueDeserializationExceptionHandler("my-app.deserialization-errors.dlq"));
        cfg.productionExceptionHandler(
            new DefaultProductionExceptionHandler("my-app.production-errors.dlq"));
    })
);
runtime.start();
runtime.awaitTermination();

To skip bad records without a DLQ, swap in the log-and-continue handler instead:

import io.stoatflow.core.exception.LogAndContinueDeserializationExceptionHandler

streamsConfigOverrides {
    deserializationExceptionHandler(LogAndContinueDeserializationExceptionHandler())
}

import io.stoatflow.core.exception.LogAndContinueDeserializationExceptionHandler;

builder.streamsConfigOverrides(cfg ->
    cfg.deserializationExceptionHandler(new LogAndContinueDeserializationExceptionHandler()));

The DLQ handlers take optional flags: includeStackTrace (default true) controls whether the full stack trace is written to a header, and the deserialization handler additionally takes includeOriginalHeaders (default true) to copy the source record's headers onto the DLQ record under a prefix.

Configuring a handler in YAML

The :runtime module can resolve handlers by fully qualified class name under stoatflow.*. Builder overrides (above) take precedence over YAML; YAML takes precedence over the defaults.

stoatflow:
  application-id: my-app
  bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:-localhost:9092}

  deserialization-exception-handler: io.stoatflow.core.exception.LogAndContinueDeserializationExceptionHandler
  processing-exception-handler: io.stoatflow.core.exception.LogAndFailProcessingExceptionHandler
  production-exception-handler: io.stoatflow.core.exception.DefaultProductionExceptionHandler

DLQ records and headers

When a DLQ handler fires, it builds a ProducerRecord targeting your configured dlqTopic and returns it to the engine. That record is sent through the same transactional producer as your normal output, committed atomically on the next commit barrier — so a DLQ record exists if and only if the rest of that epoch committed. No separate producer, no lost-or-duplicated error records.

The original key and value bytes are preserved where available:

• Deserialization failures carry the raw source bytes verbatim (key and value), since deserialization never succeeded.
• Processing failures serialize the in-flight key and value with toString().toByteArray() — the raw bytes are no longer available after the record was deserialized for processing.
• Production failures carry the (already-serialized) record bytes that failed to send — except a size rejection, which carries no value at all.

Size rejections carry no value

When the broker refuses an output record with RecordTooLargeException, the dead letter is emitted with a null value, three __stoatflow.errors.value.* markers, and a capped stack trace. Carrying the refused value would make the dead letter strictly larger than the record the broker just rejected — the original headers plus ten-odd metadata headers go on top of it — so a DLQ topic sized like the output topic would reject it too, and that second rejection surfaces only as a later commit-time failure with no dead letter written at all.

The payload is still recoverable: the dead letter records (__stoatflow.errors.topic, .partition, .offset), so seek to those coordinates on the source topic — while that topic's retention lasts. A short retention.ms on a busy source topic turns this recovery path into a dead link, so size the retention against your mean time-to-response.

Size the DLQ topic for the metadata too. The __stoatflow.errors.* header set is roughly 600 bytes before the stack trace, so a DLQ topic wants a few KiB of max.message.bytes no matter how small your values are.

Header reference

All DLQ headers use the __stoatflow.errors.* namespace. The double-underscore prefix follows the Kafka convention for system headers, so DLQ tooling that filters __-prefixed headers must opt in to surface these.

The deserialization handler is the only one that prefixes the failed record's own headers, and it appends them last. The processing and production handlers copy them verbatim and first, so lastHeader() on a __stoatflow.errors.* key always resolves to the error metadata rather than to a colliding original.

Inspect a DLQ record with the console consumer:

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-app.deserialization-errors.dlq --from-beginning \
  --property print.headers=true --property print.key=true

Production-error classification

The production stage distinguishes errors that are worth retrying from those that are not. DefaultProductionExceptionHandler treats Kafka client RetriableExceptions and TimeoutException (including when they appear as the cause) as retriable and returns RETRY; serialization errors and non-retriable send errors return FAIL (routing to the DLQ first if a topic is configured).

Underneath, StoatFlow classifies producer exceptions into three categories that determine engine-level behaviour:

Because StoatFlow runs as a single instance, there is no task to migrate a fenced producer to — so producer-fencing and related errors that Kafka Streams would treat as recoverable task migrations are FATAL here. The application exits and a fresh process resumes from the last committed barrier.

Custom handlers

Implement the handler interface for the stage you want to control. The handle methods use the Kafka Streams-compatible parameter order (context, record, exception): the context is the KS-shaped ErrorHandlerContext, and the record is the in-flight Record (processing), raw ConsumerRecord<ByteArray, ByteArray> (deserialization), or ProducerRecord<ByteArray?, ByteArray?> (production). A handler returns a response object combining the CONTINUE / FAIL decision with an optional list of DLQ ProducerRecords — the engine sends those transactionally for you.

import io.stoatflow.core.exception.ErrorHandlerContext
import io.stoatflow.core.exception.ProcessingExceptionHandler
import io.stoatflow.core.exception.ProcessingHandlerResponse
import io.stoatflow.core.processor.Record

class SkipNullPointerHandler : ProcessingExceptionHandler {
    override fun handle(
        context: ErrorHandlerContext,
        record: Record<*, *>,
        exception: Exception,
    ): ProcessingHandlerResponse =
        if (exception is NullPointerException) {
            ProcessingHandlerResponse.continueProcessing()
        } else {
            ProcessingHandlerResponse.fail()
        }
}

import io.stoatflow.core.exception.ErrorHandlerContext;
import io.stoatflow.core.exception.ProcessingExceptionHandler;
import io.stoatflow.core.exception.ProcessingHandlerResponse;
import io.stoatflow.core.processor.Record;

public class SkipNullPointerHandler implements ProcessingExceptionHandler {
    @Override
    public ProcessingHandlerResponse handle(
            ErrorHandlerContext context, Record<?, ?> record, Exception exception) {
        if (exception instanceof NullPointerException) {
            return ProcessingHandlerResponse.continueProcessing();
        }
        return ProcessingHandlerResponse.fail();
    }
}

The ErrorHandlerContext carries the source topic, partition, offset, timestamp, headers, failing processor node id, and the raw source key/value bytes, so your handler can log or tag DLQ records precisely. To emit a DLQ record from a custom handler, build a ProducerRecord<ByteArray?, ByteArray?> and pass it to continueProcessing(listOf(...)) or fail(listOf(...)).

A custom handler referenced from YAML must have a public no-arg constructor; one configured in code can take any constructor arguments you like.

Application-level DLQ routing

The exception handlers above catch failures — records that throw or fail to (de)serialize. For business-rule rejections (a record that deserializes fine but fails your validation), route it explicitly inside the topology with an ordinary sink. This is just normal DSL: branch the invalid records and .to(...) a DLQ topic.

val validated = builder.stream<String, Order>("orders")

validated
    .filterNot({ _, order -> order.isValid() }, Named.`as`("invalid"))
    .to("orders.invalid.dlq", Produced.`as`("dlq-sink"))

validated
    .filter({ _, order -> order.isValid() }, Named.`as`("valid"))
    .mapValues({ order -> order.normalize() })
    .to("orders.normalized")

KStream<String, Order> validated = builder.stream("orders");

validated
    .filterNot((k, order) -> order.isValid(), Named.as("invalid"))
    .to("orders.invalid.dlq", Produced.as("dlq-sink"));

validated
    .filter((k, order) -> order.isValid(), Named.as("valid"))
    .mapValues(Order::normalize)
    .to("orders.normalized");

This pattern gives you full control over the DLQ record shape (you choose the serde and payload) and is the right tool when "this record is bad" is a domain decision rather than an exception. The stock-tick-filter example app uses exactly this approach to send validation failures to a DLQ topic in Avro form.

Related

• Error-handling model — the why behind the three stages, CONTINUE-vs-FAIL trade-offs, transactional DLQ delivery, and the one path where continue costs an epoch of output.
• Metrics reference — the stoatflow.dlq.poison.* meters and the stoatflow.dropped.records.total reason vocabulary.
• Serdes — where deserialization failures originate, and how to make malformed data fail predictably.
• Event time and watermarks — late-record policies, which are configured separately from these exception handlers.
• Architecture — what a FAIL decision does to the running process and how recovery works on restart.