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.

A few operators look further before falling back to the default:

• merge keeps the serdes the two sides' declarations allow: a side with no serde takes the other side's, the same serde carries through, and only two different declared serdes leave the merged stream on the default. Kafka Streams always drops both sides' serdes on merge.
• An aggregation with both Grouped and Materialized serdes stores with the Materialized ones — windowed and session aggregations included, as in Kafka Streams.
• groupByKey(Grouped) with no serdes (for example Grouped.as(name)) keeps the key and value serdes the stream already declared, as groupByKey() does. Kafka Streams uses the defaults here.
• A stream-stream join without StreamJoined serdes stores each side with its own stream's value serde, and keys both stores by the key serde the two sides agree on (a side with none takes the other side's). Kafka Streams uses the defaults.
• A KTable filter, mapValues or transformValues with a Materialized that sets no serde writes its store with the parent table's serdes. Kafka Streams writes it with the defaults.
• cogroup(...).aggregate(...) — plain, windowed or session — takes the Materialized key serde, else the key serde its legs agree on (from their Grouped or upstream declarations), else the default. Kafka Streams goes straight from the Materialized to the default.
• table(topic, Consumed, Materialized) follows Kafka Streams: the Consumed serdes (or the default, where one is unset) serve both the source and the store, and the Materialized serdes are ignored. Only table(topic, Materialized), with no Consumed, deserializes the source with the Materialized serdes.

These rules matter only where the configured default would write different bytes than the serde declared upstream. Where the default cannot serialize the value at all, Kafka Streams fails and StoatFlow works; where both serdes write the same bytes, nothing differs. Where both work but write different bytes — say a JSON default over Consumed.with(Serdes.String(), …) keys — StoatFlow's stores, changelogs and outputs follow the upstream serde. For a port from Kafka Streams that carries its state over, set topology.validation.declared-serde-over-default=error (default off): the build then lists every such site, and declaring the serde you want there keeps the bytes Kafka Streams wrote.

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.

A Processor API source added with a key deserializer — Topology.addSource(name, keyDeserializer, …), or the matching addGlobalStore / addReadOnlyStateStore overloads — declares no key serde: a deserializer can read a key but not write one. A boundary whose key comes from such a source — directly, or through other Processor API processors — has nothing to resolve and lands here too, reported by the unresolved-boundary-key-serde rule like any other undeclared boundary. Declare the serde instead with 🆕 addSource(name, Consumed.with(keySerde, valueSerde), topics) (the store overloads have the same Consumed twin); it maps exactly as StreamsBuilder.stream(topic, consumed) does. Setting a default.key.serde that serializes the deserializer's type (for a StringDeserializer source, Serdes.String()) also works. The same choice keys such a processor's timers (see Timer key serde).

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

The runtime StreamsConfig is the only place defaults come from. As in Kafka Streams, a StreamsBuilder(TopologyConfig(config)) carries no serde defaults into the topology: the defaults the running application is configured with always apply.

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 deserializer or serializer instance you pass to the Processor API (Topology.addSource(name, keyDeserializer, …), addSink(name, topic, keySerializer, …)) is called by several threads at once — the dispatcher, parallel deserialization and every lane — so it must be thread-safe. (Kafka Streams also shares one instance across its tasks, but with its default single stream thread never calls it concurrently, so code that relied on that needs a thread-safe (de)serializer here.) A serde — one you declare (Consumed, Produced, …) or the configured default — is asked for its deserializer() / serializer() on every call, so a serde that hands out a fresh instance per call stays isolated.

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.

Header-aware deserializers

At the source, StoatFlow calls the record-headers overload deserialize(topic, headers, data), as Kafka Streams does. A deserializer that reads a header, such as a schema id or a content-type, gets the incoming record's headers. Deserializer's default 3-argument method delegates to the 2-argument one, so a deserializer that ignores headers behaves exactly as before. TopologyTestDriver passes the piped record's headers the same way.

Header-aware serializers

The other direction matches Kafka Streams too. Wherever StoatFlow serializes a key or value for a record, it calls serialize(topic, headers, data) with the record's headers: the sink key and value, the key bytes that pick a lane on an in-memory repartition(), and queryMetadataForKey (KIP-1321). A serializer that ignores headers produces the same bytes either way. As in Kafka Streams, the headers are the record's own object, shared by every sink the record reaches: a header a serializer adds — a schema id, say — is also visible to a sibling sink and to processors later in the same traversal. Each produced record keeps its own copy from the moment it is sent. A record that crosses an in-memory sub-topology boundary — a repartition(), or a re-key followed by an operator that needs key affinity (an aggregation, a join or toTable()) — also gets its own copy there, as a Kafka Streams repartition topic gives each record its own headers: what a processor or sink adds after the boundary is not visible before it, nor on the other branches. The one exception follows Kafka Streams too: the boundary key is serialized with the record's own headers before the copy, so a header a key serializer writes there (Confluent's header mode, for example) is visible upstream as well.

At sinks, this is what makes Confluent's schema id in headers (key.schema.id.serializer / value.schema.id.serializer = HeaderSchemaIdSerializer) work: that serializer writes the id into a header and needs the headers to write to.

Header-aware key serializers are meant for sinks. Before an in-memory repartition() — and the grouped aggregation or join it feeds — avoid a key serde whose bytes depend on the contents of the headers: the same key arriving with different headers would be spread over several lanes of one shared store, and its timers may run on another. A serde that only adds a header, like Confluent's header mode, keeps one lane per key — but in this release a header-mode key cannot key a state store or a Processor API timer yet.

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.