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 three handler stages

Each handler returns a decision:

• CONTINUE — skip the failed record and keep processing. Optionally emit one or more DLQ records.
• 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.
• RETRY — production handlers only. Retry the send for transient, retriable errors before giving up.

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.

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.

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, and transactional DLQ delivery.
• 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.