Serdes and Avro

Configure default and per-operator serdes, use the built-in Serdes, write custom serdes, serialize windowed keys, and wire up Avro with Confluent Schema Registry.

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:

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

If neither is set, serialization 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.

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

Defaults set in streamsConfigOverrides take precedence over the YAML class-name form — use the YAML keys for serdes with a no-arg constructor, and the code form whenever you need to pass a configured serde instance.

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.

Config	Operators	Factory	Builders
`Consumed<K,V>`	`stream`, `table`, `globalTable`	`Consumed.with(k, v)`	`.withKeySerde`, `.withValueSerde`
`Produced<K,V>`	`to`	`Produced.with(k, v)`	`.withKeySerde`, `.withValueSerde`
`Repartitioned<K,V>`	`repartition`	`Repartitioned.with(k, v)`	`.withKeySerde`, `.withValueSerde`
`Grouped<K,V>`	`groupByKey`, `groupBy`	`Grouped.with(k, v)`	`.withKeySerde`, `.withValueSerde`
`Materialized<K,V,S>`	`count`, `reduce`, `aggregate`, joins	`Materialized.with(k, v)`	`.withKeySerde`, `.withValueSerde`

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:

Type	Factory
`String`	`Serdes.String()`
`Long`	`Serdes.Long()`
`Integer`	`Serdes.Integer()`
`Short`	`Serdes.Short()`
`Float`	`Serdes.Float()`
`Double`	`Serdes.Double()`
`byte[]`	`Serdes.ByteArray()`
`ByteBuffer`	`Serdes.ByteBuffer()`
`Bytes`	`Serdes.Bytes()`
`UUID`	`Serdes.UUID()`
`Void`	`Serdes.Void()`

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.

A serde must round-trip a null cleanly — return null for null input. KTable changelogs and tombstones rely on null values to signal deletion, so a serde that throws on null will break stateful operators.

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().

KS deviation: the factory methods take a Serde<T> for the inner key (not a Class<T> as in Kafka Streams), and there is no window-size argument — the window bounds are encoded directly in the serialized bytes. See How StoatFlow differs from Kafka Streams.

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")
}

The Confluent serde transitively pulls kafka-clients:8.0.0-ccs, but StoatFlow targets the Apache kafka-clients 4.x APIs. Pin the Apache version so Confluent's -ccs build doesn't win dependency resolution:

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

The automatic propagation applies only to the default serdes. Any Avro serde you build yourself and pass to a Consumed, Produced, or Materialized must be configured by hand — StoatFlow never touches it.

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.

Scheduled sources

Generate records on an interval or a cron schedule with StreamsBuilder.scheduled() — a topology source that emits without consuming from Kafka.

How configuration works

How the StoatFlow runtime loads configuration — application.yaml on the classpath, overlay files, environment variables, system properties, and the layered precedence order.