> ## 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: A Serializer Is Not Compatible to the Actual Type - URL: https://www.ggorantala.dev/kafka-streams-serializer-not-compatible-serdes-fix/ - Published: 2026-06-11T13:55:00.000Z - Updated: 2026-08-29T08:34:58.000Z - Description: Kafka Streams throws 'A serializer is not compatible to the actual key or value type' or 'ClassCastException invoking Processor'. Here's the SerDe fix. - Author: Gopi Gorantala - Tags: kafka-streams, kafka-errors, StreamsException, classcastexception, serdes, Java, event-streaming You wired up a topology, ran it against a local broker, sent one record, and the `StreamThread` died with a `StreamsException` that names two serializer classes and your domain type in the same sentence. Or the mirror-image version: `ClassCastException invoking Processor`. Both are the same bug wearing different masks — the SerDe that Streams reached for does not match the type actually flowing through the topology at that point. This article confirms the error, reproduces both faces of it, explains the resolution machinery underneath, and shows the type-safe design that makes the whole class of failure impossible. This is the fourth Streams-category article in this series and is mechanically distinct from the others: it is not a lifecycle/interactive-query problem (`InvalidStateStoreException`), not FD exhaustion (`RocksDBException: Too many open files`), and not a state-directory lock fight (`LockException`). This is pure type/SerDe wiring. ## 1\. The Error Two verbatim variants. The **produce-side** one, thrown when a processor tries to write a record downstream (to a repartition topic, a changelog, a state store, or a sink): ```text org.apache.kafka.streams.errors.StreamsException: A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.StringSerializer) is not compatible to the actual key or value type (key type: java.lang.String / value type: com.example.orders.Order). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters (for example if using the DSL, #to(String topic, Produced produced) with Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:210) at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:85) ... Caused by: java.lang.ClassCastException: class com.example.orders.Order cannot be cast to class java.lang.String ``` The **consume/process-side** one, thrown when a deserialized record is fed into a processor whose input types don't line up with what came off the wire: ```text org.apache.kafka.streams.errors.StreamsException: ClassCastException invoking Processor. Do the Processor's input types match the deserialized types? Check the Serde setup and change the default Serdes in StreamConfig or provide correct Serdes via method parameters. Make sure the Processor can accept the deserialized input of type key: java.lang.String, and value: java.lang.String. Note that although incorrect Serdes are a common cause of error, the cast exception might have another cause (in user code, for example). For example, if a processor wires in a store, but casts the generics incorrectly, a class cast exception could be raised during processing, but the cause would not be wrong Serdes. at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:...) ``` Environment where this shows up: **kafka-clients / kafka-streams 2.x–4.0**, any broker version (this is 100% client-side — the broker never sees a type). Overwhelmingly a **local POC / first-run** failure: DSL app started with `Serdes.String()` as the default while the values are POJOs, Avro, or JSON. It also appears in production the first time a topology adds a `groupBy`, `join`, `through`/`repartition`, or windowed aggregation that forces a serialization boundary the default SerDe was never meant to cross. ## 2\. How to Reproduce It Minimal single-broker KRaft cluster: ```yaml # docker-compose.yml services: kafka: image: apache/kafka:3.9.0 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 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 CLUSTER_ID: 4L6g3nShT-eMCtK--X86sw ``` ```bash docker compose up -d docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \ --bootstrap-server localhost:9092 --create --topic orders --partitions 1 docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \ --bootstrap-server localhost:9092 --create --topic orders-by-customer --partitions 1 ``` A DSL topology that leaves the default value SerDe as `String` but pushes `Order` objects through it. `kafka-streams` 3.9.0: ```java // build.gradle: implementation 'org.apache.kafka:kafka-streams:3.9.0' import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.*; import org.apache.kafka.streams.kstream.*; import java.util.Properties; public class OrderTopology { public record Order(String customerId, long amountCents) {} public static void main(String[] args) { Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-router"); props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // BUG: default value serde is String, but values are Order. props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); StreamsBuilder builder = new StreamsBuilder(); KStream orders = builder .stream("orders", Consumed.with(Serdes.String(), orderSerde())) .map((k, v) -> KeyValue.pair(v.customerId(), v)); // No Produced.with(...) here -> Streams falls back to the DEFAULT // value serde (String) to write an Order. Boom at the sink. orders.to("orders-by-customer"); KafkaStreams streams = new KafkaStreams(builder.build(), props); streams.start(); Runtime.getRuntime().addShutdownHook(new Thread(streams::close)); } static Serde orderSerde() { // any working JSON/Avro serde; omitted for brevity return /* JsonSerde */ null; } } ``` Send one record and watch the thread die: ```bash docker compose exec kafka /opt/kafka/bin/kafka-console-producer.sh \ --bootstrap-server localhost:9092 --topic orders \ --property parse.key=true --property key.separator=: > o1:{"customerId":"c-42","amountCents":1999} ``` The `map` succeeds, but `to("orders-by-customer")` with no explicit `Produced` falls back to the default value SerDe (`StringSerializer`), which is handed an `Order`. `RecordCollectorImpl.send()` catches the `ClassCastException` and rethrows the descriptive `StreamsException`. To trigger the **process-side** variant instead, flip it: declare the source as `Consumed.with(Serdes.String(), Serdes.String())` (so the value deserializes as `String`) but write a `.mapValues((String v) -> ...)` processor that a later node treats as `Order`. Environment-specific triggers worth knowing: it fires **only at a serialization boundary**, so a topology that just does `filter`/`peek` can run clean for weeks and then blow up the day someone adds a `groupByKey().aggregate(...)` (which needs to serialize to a changelog + repartition topic), a `join` (repartition), or a windowed store (`WindowedSerdes`). ## 3\. Why It Happens — Surface Level Kafka Streams is generic at the Java type level (`KStream`) but **erased at runtime**. It cannot see that your `KStream` carries `Order`; it only has the SerDe you gave it, or — when you gave none — the `default.value.serde` from `StreamsConfig`. At every serialization boundary (sink topic, repartition topic, changelog, state store) Streams asks a SerDe to turn the object into bytes. If that SerDe is `StringSerializer` and the object is `Order`, the internal cast fails, and Streams wraps the raw `ClassCastException` in the friendly `StreamsException` that tells you exactly which serializer met which type. The mirror case is deserialization: bytes come off a topic, the configured deserializer produces a `String`, and the next processor's code expects `Order`. Same cast failure, thrown one node later. ## 4\. Why It Happens — Under the Hood Streams resolves the SerDe for every node with a strict precedence, and understanding it tells you exactly where the wrong one leaked in: 1. **Explicit operator SerDe** — `Consumed.with(...)`, `Produced.with(...)`, `Grouped.with(...)`, `Materialized.with(...)`, `Joined.with(...)`, `Repartitioned.with(...)`, `StreamJoined.with(...)`. Highest priority. If you set it, it wins. 2. **Inherited SerDe** — a few operators propagate the SerDe from the upstream operator when types are known to be unchanged (e.g. `filter` keeps the parent's). But a type-changing operator like `map`/`selectKey`/`groupBy` **breaks inheritance**: Streams no longer knows the new type, so it falls back to the default. 3. **Config default** — `default.key.serde` / `default.value.serde` (`StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG` / `DEFAULT_VALUE_SERDE_CLASS_CONFIG`). Last resort. If nothing above supplied a SerDe, this is used — and **this is where 90% of these errors originate**: the default was left at `Serdes.String()` (or worse, never set, so a config `ClassCastException`/`ConfigException` appears instead), while a POJO/Avro value silently relied on it. On the produce path, `RecordCollectorImpl.send(...)` invokes `keySerializer.serialize(topic, headers, key)` and `valueSerializer.serialize(...)`. `StringSerializer.serialize` signature is `serialize(String, String)`; passing an `Order` triggers the JVM-level `ClassCastException` **before** the serializer body even runs. `RecordCollectorImpl` catches `ClassCastException` specifically and rethrows the SerDe-diagnostic `StreamsException` you see, so the error is deliberately annotated, not a bare stack trace. On the consume path, `RecordDeserializer` uses the source node's deserializer to build the `ConsumerRecord`'s typed value, then `ProcessorNode.process(...)` casts it to the processor's declared input generic. `ProcessorNode` catches the `ClassCastException` and rethrows the "ClassCastException invoking Processor" variant — which is why the message hedges that the cause "might be in user code," e.g. a store retrieved with the wrong generic parameters. Two subtleties that trip up seniors: - **Windowed aggregations** don't use your key SerDe directly for the store/changelog — they wrap it in a `WindowedSerde` (`TimeWindowedSerde`/`SessionWindowedSerde`). That's why the error message literally suggests `WindowedSerdes.timeWindowedSerdeFrom(...)`. Supplying a plain `Serdes.String()` where a windowed key is expected produces the same failure. - **Repartition/changelog topics are created lazily.** A topology can pass `build()` and start cleanly; the SerDe mismatch only materializes when the first record reaches the boundary node at runtime — which is why "it compiled and started fine" is not evidence the SerDes are correct. ## 5\. The Fix The fix is always the same shape: supply the correct SerDe at the boundary, or set a correct default. Prefer explicit per-operator SerDes — they're compile-time-visible and local to the operator that needs them. **Before** (relies on a wrong/`String` default value SerDe): ```java props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); builder.stream("orders", Consumed.with(Serdes.String(), orderSerde())) .map((k, v) -> KeyValue.pair(v.customerId(), v)) .to("orders-by-customer"); // falls back to String default ``` **After** (explicit `Produced` at the sink; default no longer matters here): ```java builder.stream("orders", Consumed.with(Serdes.String(), orderSerde())) .map((k, v) -> KeyValue.pair(v.customerId(), v)) .to("orders-by-customer", Produced.with(Serdes.String(), orderSerde())); // <-- correct value serde ``` For aggregations, put the SerDe on `Grouped` **and** `Materialized`, because both the repartition topic and the changelog need it: ```java KTable totals = orders .groupBy((k, v) -> v.customerId(), Grouped.with(Serdes.String(), orderSerde())) // repartition topic .aggregate( () -> 0L, (customerId, order, sum) -> sum + order.amountCents(), Materialized.>as("totals") .withKeySerde(Serdes.String()) .withValueSerde(Serdes.Long())); // changelog + store ``` For a windowed key, use the windowed SerDe the message asked for: ```java Produced.with( WindowedSerdes.timeWindowedSerdeFrom(String.class, 60_000L), Serdes.Long()); ``` If you genuinely have one dominant value type across the whole topology, set the default correctly instead of repeating it — but do it with a real SerDe, not a placeholder: ```java props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde.class); props.put("schema.registry.url", "http://localhost:8081"); ``` When to use which: **per-operator SerDes** for any non-trivial topology (multiple value types, joins, windows) — they localize the failure and survive refactors. **Config default** only when the whole app is genuinely single-typed (e.g. a String→String router). Never mix a POJO topology with a `String` default and hope inheritance covers you across a `map`. ## 6\. Best Practices & The Better Design The root design flaw is treating SerDes as a global afterthought instead of a per-edge contract. The topology has typed edges; make each edge carry its own SerDe explicitly, and delete the ambiguous default so a missing SerDe fails loudly at wiring time rather than silently falling back. The "right way" version — no reliance on defaults, one SerDe per boundary, schema-backed values: ```java Serde orderSerde = new SpecificAvroSerde<>(); orderSerde.configure(Map.of("schema.registry.url", "http://localhost:8081"), false); StreamsBuilder builder = new StreamsBuilder(); KStream orders = builder.stream("orders", Consumed.with(Serdes.String(), orderSerde)); orders.map((k, v) -> KeyValue.pair(v.getCustomerId(), v)) .to("orders-by-customer", Produced.with(Serdes.String(), orderSerde)); orders.groupByKey(Grouped.with(Serdes.String(), orderSerde)) .aggregate(() -> 0L, (k, o, sum) -> sum + o.getAmountCents(), Materialized.>as("totals") .withKeySerde(Serdes.String()) .withValueSerde(Serdes.Long())); ``` Why this is fundamentally better: a schema registry (Avro/Protobuf/JSON-Schema) makes the value type a versioned contract shared across producers and consumers, so a type mismatch becomes a schema-compatibility failure at registration time (the `409` covered elsewhere in this series) rather than a runtime `ClassCastException` that kills a `StreamThread` in production. And because every boundary names its SerDe, a code reviewer can see the type contract without running the app. Two more habits that pay off: keep a single `Serde` instance per type and reuse it (don't `new` one per operator — it re-parses config each time), and never leave `default.value.serde` pointing at a type that only some of the topology uses. ## 7\. How to Prevent It Long-Term Reproduce the whole topology in a **`TopologyTestDriver`** unit test with the real SerDes — it runs the exact serialization boundaries in-process with no broker, so a SerDe mismatch fails in CI in milliseconds: ```java try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) { TestInputTopic in = driver.createInputTopic( "orders", Serdes.String().serializer(), orderSerde.serializer()); TestOutputTopic out = driver.createOutputTopic( "orders-by-customer", Serdes.String().deserializer(), orderSerde.deserializer()); in.pipeInput("o1", new Order("c-42", 1999)); assertThat(out.readValue().customerId()).isEqualTo("c-42"); } ``` Operationally, alert on **`StreamThread` deaths** via `KafkaStreams.setUncaughtExceptionHandler(...)` returning `SHUTDOWN_CLIENT`/`REPLACE_THREAD` decisions and on the `failed-stream-threads` metric — a SerDe mismatch surfaces as a dead thread, not a slow one, so lag and throughput alerts catch it late. Treat any recurring `StreamsException` containing "is not compatible" or "ClassCastException invoking Processor" as a wiring bug, never something to retry. In review, lint for `to(...)`/`groupBy(...)`/`aggregate(...)`/`join(...)` calls that omit their `Produced`/`Grouped`/`Materialized`/`Joined` SerDe on a topology whose default value SerDe isn't the value type — that omission is the entire bug. ## 8\. Key Takeaways - The error means the SerDe Streams reached for doesn't match the runtime type at a serialization boundary — produce-side ("A serializer ... is not compatible") or process-side ("ClassCastException invoking Processor"). Both are the same wiring bug. - SerDe resolution order is explicit operator SerDe → inherited (broken by type-changing ops like `map`/`groupBy`) → config default. Nearly every occurrence is a wrong `default.value.serde` leaking into a boundary that lost inheritance. - Fix by supplying SerDes per boundary: `Consumed`/`Produced`/`Grouped`/`Materialized`/`Joined`. Windowed keys need `WindowedSerdes`, not a plain SerDe. - It's 100% client-side and only fires at serialization boundaries, so "it started fine" proves nothing — adding a join/aggregate/window later can trigger it in prod. - Prevent it with `TopologyTestDriver` tests using real SerDes, a schema registry to make value types a versioned contract, and a lint that flags boundary operators missing explicit SerDes.