> ## Content Index > Fetch the complete content index at: https://www.ggorantala.dev/llms.txt > Use this file to discover other available public pages before exploring further. # Kafka Streams Deserialization Exception Handler: Fix the Fail - URL: https://www.ggorantala.dev/kafka-streams-deserialization-exception-handler-fix-the-fail/ - Published: 2026-04-27T13:41:00.000Z - Updated: 2026-08-29T08:23:58.000Z - Description: Kafka Streams dies with 'Deserialization exception handler is set to fail upon a deserialization error.' Here's why one bad record kills the thread and how to fix it. - Author: Gopi Gorantala - Tags: kafka-streams, kafka-errors, StreamsException, deserialization-exception-handler, poison-pill, Java, event-streaming 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: ```text 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: ```text ... please set the default.deserialization.exception.handler appropriately. ``` Immediately before the client goes down you get the thread-death log line: ```text [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. ```yaml # 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 ``` ```bash 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: ```java // 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): ```bash # 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. ```diff 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): ```java // 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`: ```java public class DlqDeserializationHandler implements DeserializationExceptionHandler { private KafkaProducer dlq; private String dlqTopic; @Override public void configure(Map configs) { Map 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 record, Exception exception) { ProducerRecord 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(); } } ``` ```java 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): ```java 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: ```java Map serdeConfig = Map.of( "schema.registry.url", "http://schema-registry:8081", "auto.register.schemas", false, "use.latest.version", true ); SpecificAvroSerde 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.