Kafka Streams: A Serializer Is Not Compatible to the Actual Type

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):

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<K, V> 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:

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:

# 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

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:

// 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<String, Order> 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<Order> orderSerde() {
        // any working JSON/Avro serde; omitted for brevity
        return /* JsonSerde<Order> */ null;
    }
}

Send one record and watch the thread die:

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<K, V>) 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):

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):

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:

KTable<String, Long> totals = orders
    .groupBy((k, v) -> v.customerId(),
             Grouped.with(Serdes.String(), orderSerde()))     // repartition topic
    .aggregate(
        () -> 0L,
        (customerId, order, sum) -> sum + order.amountCents(),
        Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("totals")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long()));                  // changelog + store

For a windowed key, use the windowed SerDe the message asked for:

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:

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:

Serde<Order> orderSerde = new SpecificAvroSerde<>();
orderSerde.configure(Map.of("schema.registry.url", "http://localhost:8081"), false);

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Order> 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.<String, Long, KeyValueStore<Bytes, byte[]>>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<T> 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:

try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
    TestInputTopic<String, Order> in = driver.createInputTopic(
        "orders", Serdes.String().serializer(), orderSerde.serializer());
    TestOutputTopic<String, Order> 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.