Kafka Streams Deserialization Exception Handler: Fix the Fail

Your Streams topology has been healthy for weeks. Then one malformed record lands on a source topic and the whole application shuts down. The stack trace points at a config property, not at your business logic, and the offending record is still sitting at the head of the partition — so every restart re-reads it and dies again. This is the Kafka Streams poison-pill loop, and the message that names it is one of the most-pasted Streams errors on the internet.

This is a distinct failure from a plain consumer poison pill (handled with ErrorHandlingDeserializer) and from a Serde type mismatch in your topology. Streams has its own deserialization error path with its own handler contract. Get that contract wrong and one bad byte takes down the client.

1. The Error

The exact message, as it appears when a StreamThread hits a record it cannot deserialize:

Exception in thread "orders-app-6d1e0c7f-1a2b-4c3d-9e8f-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If you would rather have the streaming pipeline continue after a deserialization error, please set the deserialization.exception.handler appropriately.
	at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:98)
	at org.apache.kafka.streams.processor.internals.RecordQueue.updateHead(RecordQueue.java:203)
	at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:134)
	at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:226)
	at org.apache.kafka.streams.processor.internals.StreamTask.doProcess(StreamTask.java:842)
	...
Caused by: org.apache.kafka.common.errors.SerializationException: Size of data received by LongDeserializer is not 8

On older clients (before Kafka 4.0) the last sentence names the property with its legacy prefix:

... please set the default.deserialization.exception.handler appropriately.

Immediately before the client goes down you get the thread-death log line:

[orders-app-...-StreamThread-1] ERROR org.apache.kafka.streams.processor.internals.StreamThread -
stream-thread [orders-app-...-StreamThread-1] Encountered the following exception during processing
and the thread is going to shut down:

followed by the uncaught-exception handler shutting the client down and the state moving RUNNING -> PENDING_ERROR -> ERROR.

Where it shows up: Kafka Streams (kafka-streams artifact) 2.x through 4.x. The Caused by varies with your Serde — SerializationException: Size of data received by LongDeserializer is not 8, Unknown magic byte! (JSON/plain bytes hitting a Confluent Avro/Protobuf deserializer), com.fasterxml.jackson.core.JsonParseException, or a RecordDeserializationException carrying the exact topic-partition and offset since Kafka 3.0 (KIP-334). Typical setup: a POC or production topology consuming a topic that also receives tombstones, hand-produced console messages, or output from a producer using a different serializer than you assumed.

2. How to Reproduce It

Spin up a broker and a topic, then feed the topology one good record and one poison record.

# docker-compose.yml — single-broker KRaft cluster
services:
  kafka:
    image: apache/kafka:4.0.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

docker compose up -d
docker exec -it kafka /opt/kafka/bin/kafka-topics.sh \
  --create --topic amounts --partitions 1 --replication-factor 1 \
  --bootstrap-server localhost:9092

A minimal topology that expects Long values and aggregates them:

// build.gradle: implementation 'org.apache.kafka:kafka-streams:4.0.0'
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long().getClass());

StreamsBuilder builder = new StreamsBuilder();
builder.stream("amounts", Consumed.with(Serdes.String(), Serdes.Long()))
       .peek((k, v) -> System.out.println("processed " + k + " -> " + v))
       .to("amounts-out");

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Produce one valid Long and one poison record (a non-numeric string that LongDeserializer cannot decode into 8 bytes):

# valid: LongSerializer-encoded value via the typed console producer
docker exec -it kafka /opt/kafka/bin/kafka-console-producer.sh \
  --topic amounts --bootstrap-server localhost:9092 \
  --property parse.key=true --property key.separator=: \
  --property value.serializer=org.apache.kafka.common.serialization.LongSerializer \
  --property key.serializer=org.apache.kafka.common.serialization.StringSerializer
>order-1:100
>^D

# poison: plain string bytes onto the same topic
docker exec -it kafka /opt/kafka/bin/kafka-console-producer.sh \
  --topic amounts --bootstrap-server localhost:9092 \
  --property parse.key=true --property key.separator=:
>order-2:not-a-long
>^D

The topology processes order-1, then hits order-2 and dies with the message from Section 1. Restart the app: because the failing record was never past, the StreamThread re-reads the same offset and dies again. That is the loop.

3. Why It Happens — Surface Level

Kafka Streams deserializes source records inside the StreamThread, before your stream(), map(), or process() code ever sees them. When the configured value (or key) Serde throws a SerializationException, Streams routes that exception to a DeserializationExceptionHandler. The default handler is LogAndFailExceptionHandler, which logs the record and returns FAIL. Streams honors that by wrapping the cause in a StreamsException and letting it propagate up the StreamThread, which then shuts down.

The default is fail-fast on purpose: silently dropping data is worse than stopping. But fail-fast plus an un-advanced offset equals an infinite crash loop, because Streams never commits past a record it could not deserialize.

4. Why It Happens — Under the Hood

The deserialization happens in RecordDeserializer.deserialize(), called from RecordQueue.addRawRecords() as records are pulled off the fetch buffer and pushed into per-partition queues. This is upstream of the processing graph — there is no try/catch you can put in your topology that will catch it, because your processors have not run yet.

When the Serde throws, RecordDeserializer invokes the handler's handle() method, passing the ProcessorContext, the raw ConsumerRecord, and the exception. The handler returns DeserializationHandlerResponse.FAIL or .CONTINUE. LogAndFailExceptionHandler returns FAIL, so RecordDeserializer throws the StreamsException you see. LogAndContinueExceptionHandler returns CONTINUE, so the record is skipped and the thread moves on.

The offset mechanics are the sharp edge. Streams tracks offsets per task and commits on commit.interval.ms (or when a transaction closes under exactly_once_v2). A FAIL response throws before the offset advances, so nothing is committed for that record. On restart the StreamThread resumes from the last committed offset — which is at or before the poison record — and hits it again. Nothing about the record changed, so the outcome is deterministic: crash, restart, crash.

The thread death is governed by the StreamsUncaughtExceptionHandler. Since Kafka 2.8 (KIP-671) the default response is SHUTDOWN_CLIENT: one bad record on one partition takes down the entire KafkaStreams instance, not just the affected thread. That is why a single poison pill can idle a whole service.

One more subtlety worth internalizing: the deserialization handler only catches errors from the source Serde. An exception thrown inside your processing logic (a NullPointerException in a mapValues, a downstream produce failure) is a different path — handled by the production exception handler or, since Kafka 3.9 (KIP-1033), the processing exception handler. Don't reach for deserialization.exception.handler to swallow business-logic bugs; it will never see them.

5. The Fix

There are three levers, and which you pull depends on whether the bad data is expected noise or a signal that something upstream is broken.

Lever 1 — Skip and continue (tolerate malformed input). Switch the handler to LogAndContinueExceptionHandler. Bad records are logged and skipped so the pipeline keeps moving.

  Properties props = new Properties();
  props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app");
  props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+ // Kafka 4.0+ (KIP-1056): the `default.` prefix is dropped
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+           LogAndContinueExceptionHandler.class.getName());

On Kafka 3.x and earlier, use the legacy key (still accepted, now deprecated):

// Kafka <= 3.x
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
          LogAndContinueExceptionHandler.class.getName());

If both the old and new configs are set, the new one wins and Streams logs a WARN that the deprecated one is ignored.

Lever 2 — Route bad records to a DLQ (tolerate, but keep the evidence). LogAndContinue throws the record away. In production you almost always want it captured. Kafka 3.9+ ships a built-in dead-letter implementation via DeserializationExceptionHandler, or you write a small custom handler that produces the raw bytes to a DLQ topic and returns CONTINUE:

public class DlqDeserializationHandler implements DeserializationExceptionHandler {
    private KafkaProducer<byte[], byte[]> dlq;
    private String dlqTopic;

    @Override
    public void configure(Map<String, ?> configs) {
        Map<String, Object> p = new HashMap<>();
        p.put(BOOTSTRAP_SERVERS_CONFIG, configs.get(BOOTSTRAP_SERVERS_CONFIG));
        p.put(KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        p.put(VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        this.dlq = new KafkaProducer<>(p);
        this.dlqTopic = (String) configs.get("dlq.topic");
    }

    @Override
    public DeserializationHandlerResponse handle(ErrorHandlerContext context,
                                                 ConsumerRecord<byte[], byte[]> record,
                                                 Exception exception) {
        ProducerRecord<byte[], byte[]> dead =
            new ProducerRecord<>(dlqTopic, record.key(), record.value());
        dead.headers().add("dlq.source.topic", context.topic().getBytes(UTF_8));
        dead.headers().add("dlq.exception", exception.getClass().getName().getBytes(UTF_8));
        dlq.send(dead);
        return DeserializationHandlerResponse.CONTINUE;
    }

    @Override public void close() { if (dlq != null) dlq.close(); }
}

props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
          DlqDeserializationHandler.class.getName());
props.put("dlq.topic", "amounts-dlq");

Lever 3 — Keep failing, but survive the individual record (unblock without dropping data). If you must not skip data and the bad record is a genuine bug you need to fix upstream, keep LogAndFail but stop the poison pill from killing the whole client. Set the uncaught-exception handler to replace the thread rather than shut down, and manually seek past the offset once you have captured it, using the topic/partition/offset that RecordDeserializationException carries (KIP-334):

streams.setUncaughtExceptionHandler(ex ->
    StreamThreadExceptionResponse.REPLACE_THREAD);

Note that REPLACE_THREAD alone will still loop on a truly poison record — pair it with a DLQ handler or an offset reset. Use this only for transient deserialization failures (a flaky schema-registry lookup), not for permanently malformed bytes.

6. Best Practices & The Better Design

Skipping or DLQ-ing is damage control. The class of bug is "the bytes on the topic don't match what the consumer expects," and the real fix is a schema contract enforced at the producer.

Use a schema registry with a concrete Serde and disable client-side auto-registration so producers cannot quietly evolve the schema:

Map<String, Object> serdeConfig = Map.of(
    "schema.registry.url", "http://schema-registry:8081",
    "auto.register.schemas", false,
    "use.latest.version", true
);
SpecificAvroSerde<Amount> valueSerde = new SpecificAvroSerde<>();
valueSerde.configure(serdeConfig, false);

builder.stream("amounts", Consumed.with(Serdes.String(), valueSerde))
       .to("amounts-out");

With Avro/Protobuf plus a registry, malformed records are rejected at the producer before they ever hit the topic, and the Unknown magic byte! class of failure disappears because everything on the topic carries a schema id. Pair that with a DLQ handler as a backstop for the residual cases (corruption, a rogue console producer, a topic reused across teams). The pattern that survives production is: strict schema at write time, LogAndContinue-plus-DLQ at read time, and an alert on DLQ throughput so a silent skip never becomes silent data loss.

Finally, separate your handlers by concern: deserialization.exception.handler for source-Serde failures, production.exception.handler for sink write failures, and (Kafka 3.9+) processing.exception.handler for exceptions your topology code throws. Conflating them — especially trying to catch business-logic bugs with the deserialization handler — is how teams end up masking real defects.

7. How to Prevent It Long-Term

Make deserialization handling a deployment default, not a per-app afterthought. Bake DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG into your shared Streams config factory so every service ships with a DLQ handler rather than the crash-looping default.

Monitor the right signals. Alert on the KafkaStreams state() transitioning to ERROR and on StreamThread count dropping below num.stream.threads. Track DLQ topic produce rate — a spike is your early warning that an upstream producer changed its serialization. Watch records-lag-max per partition: a poison-pill loop shows as a partition whose lag climbs while others drain.

Test it before production. A Testcontainers or TopologyTestDriver case that feeds one valid record and one un-deserializable record, then asserts the topology stays RUNNING and the bad record lands in the DLQ, catches a missing or misconfigured handler in CI. Add a schema-compatibility check (schema-registry-maven-plugin test-compatibility) to the producer's pipeline so incompatible writes fail the build, not the stream. Chaos-test by producing garbage bytes to a source topic in staging and confirming the app survives.

8. Key Takeaways

• The message names the config, not your code. Deserialization exception handler is set to fail upon a deserialization error means a source-topic record failed the key/value Serde before your topology ran. The Caused by (e.g. Size of data received by LongDeserializer is not 8, Unknown magic byte!) is the real diagnosis.
• The default is fail-fast and it loops. LogAndFailExceptionHandler returns FAIL, the offset never advances, and every restart re-reads the same poison record. Since KIP-671 the default takes down the whole client, not just one thread.
• Pick the lever deliberately: LogAndContinueExceptionHandler to skip, a custom/built-in DLQ handler to skip-and-capture, or keep FAIL plus REPLACE_THREAD only for transient failures.
• In Kafka 4.0 the config lost its default. prefix (KIP-1056): use deserialization.exception.handler; the old default. key still works but is deprecated and loses if both are set.
• The durable fix is a schema contract: strict registry Serdes with auto.register.schemas=false reject bad bytes at the producer; keep a DLQ handler and DLQ-rate alerting as the read-side backstop.