Serdes and Avro

A serde (serializer + deserializer) tells StoatFlow how to turn your keys and values into bytes on the wire and back. Every source, sink, grouping, and state store needs a key serde and a value serde. StoatFlow uses the standard Kafka org.apache.kafka.common.serialization.Serde<T> type, so any serde you already use with Kafka Streams works unchanged.

This page covers where serdes are resolved (defaults vs. per-operator), the built-in serdes, writing your own, serializing windowed keys, and the Avro + Schema Registry integration.

How serdes are resolved

StoatFlow resolves the serde for each operator in a fixed order:

1. An explicit serde on the operator's config object — Consumed, Produced, Grouped, or Materialized. This always wins.
2. The default key/value serde configured on the runtime — used whenever the operator config leaves the serde unset (null).

If neither is set, serialization for that operator fails at startup. The default serde is the convenient base case when most of your topology uses the same types (e.g. String keys); reach for per-operator serdes wherever an individual topic or store differs.

Boundary key serdes (lane assignment)

There is one more place StoatFlow needs a key serde that Kafka Streams does not: a sub-topology boundary. Where a re-keyed record reaches an operator that needs the new key's lane affinity — an aggregation, a join, toTable(), or an explicit repartition() — records are re-dispatched to key-affinity lanes, and the new key is serialized to bytes for the lane hash, in memory, with no repartition topic.

Since 1.0.0 there are fewer of these boundaries than there used to be: a re-key alone no longer opens one, and neither does a Processor API node (process() / processValues()), matching Kafka Streams — see Lanes and parallelism and, for what that costs after a many-to-one re-key, Key affinity after a re-key. A topology whose re-key feeds only stateless or Processor API operators now needs no boundary key serde at all.

StoatFlow resolves the boundary key serde from whatever your topology declares for that key, searching upstream first and then downstream, always within the region where the key keeps its type:

1. Upstream: the nearest ancestor declaration — the source's Consumed key serde, or a repartition(Repartitioned.withKeySerde(...)) the key flowed through.
2. Downstream: the nearest declaration for the new key — an explicit Repartitioned, a Grouped, a sink's Produced, or an aggregation's Materialized.withKeySerde(...).
3. Last resort: if you never configured a default key serde, the nearest declaration from before the re-key — inherited across it. See Last-resort inheritance below.
4. The configured default key serde otherwise.

Two rules keep steps 1 and 2 predictable:

• A declaration only counts for the key it actually describes. After a re-key, upstream serdes no longer apply; a Grouped serde describes the grouping key, never the key arriving at the groupBy; and two adjacent re-keys find nothing in either step — no declaration sits between them to describe the intermediate key, so that boundary falls through to the last two steps.
• If two different serde classes are reachable for the same key, StoatFlow declines to guess.

Last-resort inheritance

Steps 1 and 2 are strict: they only ever return a serde that was declared for this key. When both come up empty, StoatFlow walks upstream a second time, through the re-key, and takes the nearest declaration from the previous key region.

That is safe because a boundary key serde is used for lane hashing only. The bytes feed a MurmurHash3 lane assignment and nothing else — they are never stored, never sent to Kafka, and never deserialized back. So the serde only has to encode your key deterministically; it does not have to be the right serde for it in any other sense. (This is exactly why the same leniency is not applied to keyed timers, whose key bytes are persisted and read back.)

It works when the re-key preserves the declared key type — the common case for a re-key that reshapes a String id. When the type changes, the inherited serde cannot encode the new key and the boundary still fails on the first record, with the inherited serde named in the message.

A warning always names the edge ('parent' → 'child') and the inherited serde class, because the topology now starts where it previously died — and if the types happen not to line up, you want to know before the first record. It is raised when the topology compiles, so you see it from your own tests as well as at startup. Silence it, or make it fatal, with the topology.validation rule inherited-boundary-key-serde.

When nothing resolves

If you have configured a default key serde, that default wins outright — step 3 never runs, because your explicit choice is the answer for this key. This is the escape hatch for a genuinely byte[]-keyed boundary: set default.key.serde to Serdes.ByteArray() explicitly and inheritance stays out of the way. Without that, StoatFlow cannot tell a deliberate byte-array key from an unconfigured default — they are the same value. It is the presence of the setting that decides, not its value, so any route that sets one closes inheritance: the StreamsConfig constructor, .copy(), the builder, fromMap / fromProperties, or the YAML key. Having answered, you are also out of the unresolved-boundary-key-serde rule below at every severity — including error — because there is nothing left to report.

Otherwise the boundary falls back to the raw ByteArray default and a warning names the edge when the topology compiles. If your keys are not raw bytes, that boundary will fail on the first record with LaneKeySerializationException — declare a serde at any of the places above, or set default.key.serde. The warning stays a warning by default because a genuinely byte-array-keyed boundary is valid and indistinguishable at build time; the unresolved-boundary-key-serde rule below lets you make it fatal anyway.

Validation severities

Both boundary conditions above are rules in the topology.validation family, and each can be turned off or promoted to a build failure:

stoatflow:
  topology:
    validation:
      unresolved-boundary-key-serde: error   # off | warn | error (default: warn)
      inherited-boundary-key-serde: off

error fails topology compilation with a single message naming every offending boundary at once, so a topology with several takes one build-run cycle to clean up rather than several.

All three severities are decided at that one point, when the topology compiles — not at engine start — so TopologyTestDriver reaches them as well. At the default warn that means the per-boundary line appears in your own test suite's output, with no broker; at error the driver refuses to build; at off neither happens, in tests or in production.

See the configuration reference for the full rule list.

Default serdes

Set the application-wide default key and value serdes through streamsConfigOverrides when you build the runtime. These apply to every operator that doesn't specify its own serde.

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())
        }
    },
)

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());
    })
);

You can also set the defaults declaratively in application.yaml using the fully-qualified serde class name. The class must have a no-argument constructor; for parameterised serdes (e.g. Avro), set the instance in code instead.

stoatflow:
  default-key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde
  # default-value-serde: ...

Per-operator serdes

Each operator that touches Kafka or a state store takes a config object that carries serdes. All four follow the same shape: a with(keySerde, valueSerde) factory plus chained .withKeySerde(...) / .withValueSerde(...) builders.

Source and sink

Consumed controls how a source topic is deserialized; Produced controls how a sink topic is serialized.

import io.stoatflow.core.topology.Consumed
import io.stoatflow.core.topology.Produced
import org.apache.kafka.common.serialization.Serdes

// Source: String keys, Long values
val numbers = builder.stream(
    "input",
    Consumed.with(Serdes.String(), Serdes.Long()),
)

// Sink: String keys, Long values
numbers.to(
    "output",
    Produced.with(Serdes.String(), Serdes.Long()),
)

import io.stoatflow.core.topology.Consumed;
import io.stoatflow.core.topology.KStream;
import io.stoatflow.core.topology.Produced;
import org.apache.kafka.common.serialization.Serdes;

// Source: String keys, Long values
KStream<String, Long> numbers = builder.stream(
    "input",
    Consumed.with(Serdes.String(), Serdes.Long()));

// Sink: String keys, Long values
numbers.to(
    "output",
    Produced.with(Serdes.String(), Serdes.Long()));

When you only need to override one of the two serdes (the other coming from the defaults), use the named factory or the builder form. Consumed.as(...) / Produced.as(...) set only the operator name and leave both serdes on the defaults — that is the pattern the first app uses, where the sink overrides just the value serde because counts are Long:

.to(
    "word-counts",
    Produced.`as`<String, Long>("sink")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long()),
)

.to(
    "word-counts",
    Produced.<String, Long>as("sink")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long()));

Grouping and aggregation

A groupBy / groupByKey re-serializes records as it re-keys them, so the grouping serdes must match the post-grouping key and value types. Grouped carries those serdes; if a downstream aggregation materializes state, Materialized carries the store's serdes (which may differ — the aggregate value type is often not the input value type).

import io.stoatflow.core.state.StateStore
import io.stoatflow.core.topology.Grouped
import io.stoatflow.core.topology.Materialized
import org.apache.kafka.common.serialization.Serdes

stream
    .groupBy(
        { _, value -> value.category },
        Grouped.with(Serdes.String(), productSerde),
    )
    .count(
        Materialized.`as`<String, Long, StateStore>("counts-by-category")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long()),
    )

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.Grouped;
import io.stoatflow.core.topology.Materialized;
import org.apache.kafka.common.serialization.Serdes;

stream
    .groupBy(
        (key, value) -> value.category(),
        Grouped.with(Serdes.String(), productSerde))
    .count(
        Materialized.<String, Long, StateStore>as("counts-by-category")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long()));

Materialized carries far more than serdes (store type, changelog, caching, retention). See State stores for the rest; this page covers only its serde configuration.

Built-in serdes

The standard Kafka serdes cover the common primitive and byte types. They're factory methods on org.apache.kafka.common.serialization.Serdes:

These are plain Kafka client types — the same ones you'd reach for in a Kafka Streams app. The full list is whatever your kafka-clients version exposes on Serdes.

Custom serdes

For your own value types — JSON, Protobuf, a hand-rolled binary format — implement org.apache.kafka.common.serialization.Serde<T>, or build one from a Serializer<T> / Deserializer<T> pair with Serdes.serdeFrom(serializer, deserializer).

A serde is just a serializer and a deserializer bundled together:

import com.fasterxml.jackson.databind.ObjectMapper
import org.apache.kafka.common.serialization.Deserializer
import org.apache.kafka.common.serialization.Serde
import org.apache.kafka.common.serialization.Serdes
import org.apache.kafka.common.serialization.Serializer

data class Order(val id: String, val amount: Double)

class JsonOrderSerializer : Serializer<Order> {
    private val mapper = ObjectMapper()
    override fun serialize(topic: String?, data: Order?): ByteArray? =
        data?.let { mapper.writeValueAsBytes(it) }
}

class JsonOrderDeserializer : Deserializer<Order> {
    private val mapper = ObjectMapper()
    override fun deserialize(topic: String?, data: ByteArray?): Order? =
        data?.let { mapper.readValue(it, Order::class.java) }
}

val orderSerde: Serde<Order> =
    Serdes.serdeFrom(JsonOrderSerializer(), JsonOrderDeserializer())

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;

record Order(String id, double amount) {}

class JsonOrderSerializer implements Serializer<Order> {
    private final ObjectMapper mapper = new ObjectMapper();
    @Override
    public byte[] serialize(String topic, Order data) {
        if (data == null) return null;
        try {
            return mapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

class JsonOrderDeserializer implements Deserializer<Order> {
    private final ObjectMapper mapper = new ObjectMapper();
    @Override
    public Order deserialize(String topic, byte[] data) {
        if (data == null) return null;
        try {
            return mapper.readValue(data, Order.class);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

Serde<Order> orderSerde =
    Serdes.serdeFrom(new JsonOrderSerializer(), new JsonOrderDeserializer());

Use orderSerde exactly like a built-in — pass it to Consumed.with(...), Produced.with(...), Materialized.withValueSerde(...), or set it as the default value serde.

Windowed keys

Windowed aggregations (windowedBy(...).count(), session windows, sliding windows) produce keys of type Windowed<K> — the original key plus the window's start and end. To write those keys to a topic (or read them back), you need a serde that encodes the window bounds alongside the key bytes. WindowedSerdes builds those for you from the inner key serde.

The state store doesn't need one: windowed Materialized takes the plain key serde (Serdes.String()) — StoatFlow wraps it into a Serde<Windowed<K>> internally, exactly like Kafka Streams. So the windowed serde belongs on the output (Produced), not the Materialized:

import io.stoatflow.core.state.StateStore
import io.stoatflow.core.topology.Materialized
import io.stoatflow.core.topology.Produced
import io.stoatflow.core.topology.WindowedSerdes
import org.apache.kafka.common.serialization.Serdes

// Encodes Windowed<String> keys for a topic — used on the output, not the store.
val windowedKeySerde = WindowedSerdes.timeWindowedSerdeFrom(Serdes.String())

stream
    .groupByKey()
    .windowedBy(/* window definition */)
    .count(
        // Plain key serde — StoatFlow wraps it into Serde<Windowed<String>> for the result table.
        Materialized.`as`<String, Long, StateStore>("windowed-counts")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long()),
    ).toStream()
    .to("windowed-counts", Produced.with(windowedKeySerde, Serdes.Long()))

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.Materialized;
import io.stoatflow.core.topology.Produced;
import io.stoatflow.core.topology.WindowedSerdes;
import io.stoatflow.core.topology.Windowed;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;

// Encodes Windowed<String> keys for a topic — used on the output, not the store.
Serde<Windowed<String>> windowedKeySerde =
    WindowedSerdes.timeWindowedSerdeFrom(Serdes.String());

stream
    .groupByKey()
    .windowedBy(/* window definition */)
    .count(
        // Plain key serde — StoatFlow wraps it into Serde<Windowed<String>> for the result table.
        Materialized.<String, Long, StateStore>as("windowed-counts")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long()))
    .toStream()
    .to("windowed-counts", Produced.with(windowedKeySerde, Serdes.Long()));

Use sessionWindowedSerdeFrom(inner) instead for session windows. Both factories also exist as constructors — WindowedSerdes.TimeWindowedSerde(inner) and WindowedSerdes.SessionWindowedSerde(inner) — and expose the unwrapped inner serde via keySerde().

Avro with Schema Registry

Avro is the most common production serialization format with Kafka. StoatFlow has no Avro serde of its own — you use the Confluent SpecificAvroSerde (or the generic/reflection variants), the same one you'd use with Kafka Streams. StoatFlow's only role is to hand Schema Registry's URL to your default serdes; serdes you build yourself, you configure yourself.

Build setup

Generate Java classes from your .avsc schemas (the com.github.davidmc24.gradle.plugin.avro plugin against org.apache.avro:avro is one common choice), then depend on the Confluent serde from the Confluent Maven repository:

// build.gradle.kts
repositories {
    mavenCentral()
    maven("https://packages.confluent.io/maven/")
}

dependencies {
    implementation("io.stoatflow:stoatflow-runtime:<stoatflow-version>")
    implementation("io.confluent:kafka-streams-avro-serde:8.0.0")
}

configurations.all {
    resolutionStrategy.eachDependency {
        if (requested.group == "org.apache.kafka" && requested.name == "kafka-clients") {
            useVersion("4.3.1") // your Apache kafka-clients version
        }
    }
}

Pointing serdes at Schema Registry

Set the registry URL once in application.yaml. StoatFlow propagates it to the default key and value serdes at startup, so an Avro serde set as the default value serde is configured for you:

stoatflow:
  application-id: news-article-publish-processor
  bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:-localhost:9092}
  schema-registry-url: ${SCHEMA_REGISTRY_URL:-http://localhost:8081}
  default-key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde

import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
import io.stoatflow.runtime.StoatFlowRuntime
import org.apache.kafka.common.serialization.Serdes

val runtime = StoatFlowRuntime.fromConfig(
    topologyBuilder = { buildTopology(it) },
    configure = {
        streamsConfigOverrides {
            defaultKeySerde(Serdes.String())
            // Configured automatically with schema-registry-url from YAML
            defaultValueSerde(SpecificAvroSerde<Article>())
        }
    },
)

import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
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());
        // Configured automatically with schema-registry-url from YAML
        cfg.defaultValueSerde(new SpecificAvroSerde<>());
    }));

Per-operator Avro serdes

When different topics carry different Avro types — or you want the value serde on a specific source rather than as the global default — build each serde and call configure(...) with the registry URL before handing it to the operator. The boolean second argument is isKey (true for key serdes, false for value serdes):

import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
import org.apache.kafka.common.serialization.Serde

fun createStockTickSerde(schemaRegistryUrl: String): Serde<StockTick> {
    val serde = SpecificAvroSerde<StockTick>()
    serde.configure(mapOf("schema.registry.url" to schemaRegistryUrl), false)
    return serde
}

import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
import org.apache.kafka.common.serialization.Serde;
import java.util.Map;

static Serde<StockTick> createStockTickSerde(String schemaRegistryUrl) {
    SpecificAvroSerde<StockTick> serde = new SpecificAvroSerde<>();
    serde.configure(Map.of("schema.registry.url", schemaRegistryUrl), false);
    return serde;
}

Then wire the configured serde into the source or sink exactly like any other serde:

import io.stoatflow.core.topology.Consumed
import io.stoatflow.core.topology.Produced
import org.apache.kafka.common.serialization.Serdes

val stockTickSerde = createStockTickSerde(schemaRegistryUrl)

builder
    .stream("stock-ticks", Consumed.with(Serdes.String(), stockTickSerde))
    // ... processing ...
    .to(
        "filtered-stock-ticks",
        Produced.with(Serdes.String(), stockTickSerde),
    )

import io.stoatflow.core.topology.Consumed;
import io.stoatflow.core.topology.Produced;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;

Serde<StockTick> stockTickSerde = createStockTickSerde(schemaRegistryUrl);

builder
    .stream("stock-ticks", Consumed.with(Serdes.String(), stockTickSerde))
    // ... processing ...
    .to(
        "filtered-stock-ticks",
        Produced.with(Serdes.String(), stockTickSerde));

Testing Avro topologies

For unit tests against the in-memory test driver, configure your Avro serdes against Confluent's mock Schema Registry (mock://...) instead of a live one — it keeps schemas in-process, so no broker or registry is needed:

import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig
import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde

fun stockTickSerde(): SpecificAvroSerde<StockTick> {
    val serde = SpecificAvroSerde<StockTick>()
    serde.configure(
        mapOf(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG to "mock://schema-registry"),
        false,
    )
    return serde
}

import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
import java.util.Map;

static SpecificAvroSerde<StockTick> stockTickSerde() {
    SpecificAvroSerde<StockTick> serde = new SpecificAvroSerde<>();
    serde.configure(
        Map.of(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "mock://schema-registry"),
        false);
    return serde;
}

When schema-registry-url is configured, the runtime also exposes a Schema Registry health check on /health/ready that auto-enables itself — no extra wiring required.

Next steps

• State stores — the rest of Materialized: store types, changelog, caching, retention.
• Windowing — where Windowed<K> keys come from and how windows close.
• Error handling and DLQ — what happens when a record fails to deserialize.
• Testing — drive a topology end-to-end in-memory with the test driver.