Sources and sinks

Read records into a topology with builder.stream / table / globalTable and Consumed, and write them back out with to() and Produced — serdes, offset reset, watermarks, partitioners, and the fan-out rule.

Every topology starts at a source and ends at a sink. StreamsBuilder opens source topics — stream for an event stream, table / globalTable for a changelog table — and KStream.to writes records back to Kafka. Consumed configures the read side (serdes, offset reset, watermarks); Produced configures the write side (serdes, partitioner, dynamic topic routing).

This page covers the source and sink methods only. For the operators in between, see KStream and KTable. For the engine that runs the topology, see Architecture.

Reading a stream

builder.stream subscribes to one or more topics and returns a KStream<K, V>. Pass a Consumed to configure deserialization and source behaviour; omit it to fall back to the default serdes from your config.

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

fun buildTopology(builder: StreamsBuilder) {
    val orders =
        builder.stream(
            "orders",
            Consumed.with(Serdes.String(), Serdes.String()),
        )
    // orders: KStream<String, String>
}

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

void buildTopology(StreamsBuilder builder) {
    KStream<String, String> orders =
        builder.stream(
            "orders",
            Consumed.with(Serdes.String(), Serdes.String()));
}

Multiple topics and patterns

stream also accepts a collection of topic names — all merged into one stream — or a java.util.regex.Pattern for dynamic subscription, where new topics matching the pattern are picked up at runtime.

import java.util.regex.Pattern

// Several named topics, merged into one stream
builder.stream(listOf("orders-eu", "orders-us"), Consumed.with(keySerde, valueSerde))

// Pattern subscription — matching topics are added dynamically
builder.stream(Pattern.compile("orders-.*"), Consumed.with(keySerde, valueSerde))

import java.util.List;
import java.util.regex.Pattern;

// Several named topics, merged into one stream
builder.stream(List.of("orders-eu", "orders-us"), Consumed.with(keySerde, valueSerde));

// Pattern subscription — matching topics are added dynamically
builder.stream(Pattern.compile("orders-.*"), Consumed.with(keySerde, valueSerde));

A topic may only be registered as a source once per topology. Subscribing to it from two separate builder.stream(...) calls is rejected when the topology is built. To process the same input two ways, reuse one KStream reference — see The fan-out rule below.

Reading a table

builder.table interprets a topic as a changelog — the latest value per key is the current state — and returns a KTable<K, V> backed by a state store. Unlike a plain stream, a table is always materialized: when you don't pass a Materialized, StoatFlow auto-creates a store named {topic}-store. (See State stores for store configuration.)

import io.stoatflow.core.state.StateStore
import io.stoatflow.core.topology.Consumed
import io.stoatflow.core.topology.Materialized

// Auto-materialized into "customers-store"
val customers = builder.table("customers", Consumed.with(Serdes.String(), Serdes.String()))

// Explicit store name + serdes
val customersNamed =
    builder.table(
        "customers",
        Consumed.`as`("customers-source"),
        Materialized.`as`<String, String, StateStore>("customers")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.String()),
    )

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.Consumed;
import io.stoatflow.core.topology.KTable;
import io.stoatflow.core.topology.Materialized;

// Auto-materialized into "customers-store"
KTable<String, String> customers =
    builder.table("customers", Consumed.with(Serdes.String(), Serdes.String()));

// Explicit store name + serdes
KTable<String, String> customersNamed =
    builder.table(
        "customers",
        Consumed.as("customers-source"),
        Materialized.<String, String, StateStore>as("customers")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.String()));

builder.globalTable(...) has the same signatures and returns a GlobalKTable. In StoatFlow's single-instance model all state is already global (see Architecture), so globalTable is functionally identical to table — it exists for Kafka Streams API compatibility.

Source-topic reuse for compacted tables

When the source topic is log-compacted, StoatFlow can restore the table's state directly from the source topic instead of creating a separate changelog topic — avoiding duplicated data. This is automatic by default (controlled globally) and can be forced per-table with Consumed.withMaterializeFromSourceTopic(...):

// Force source-topic reuse — no separate changelog created
builder.table("users", Consumed.with(Serdes.String(), userSerde).withMaterializeFromSourceTopic(true))

// Force a dedicated changelog topic even if the source is compacted
builder.table("events", Consumed.with(Serdes.String(), eventSerde).withMaterializeFromSourceTopic(false))

true forces source-topic restoration (no changelog, no compaction check); false forces a dedicated changelog topic; null (the default) uses the global setting plus automatic compaction detection.

Configuring the read — `Consumed`

Consumed<K, V> carries everything about how a source is read. Create one with a static factory, then chain with* methods — each returns a new immutable instance.

Factory	Produces
`Consumed.with(keySerde, valueSerde)`	both serdes set
`Consumed.keySerde(serde)` / `Consumed.valueSerde(serde)`	one serde set
`Consumed.with(watermarkStrategy)`	watermark strategy set
`Consumed.as("name")`	named source, default serdes
`Consumed.offsetResetPolicy(policy)`	offset-reset policy set

Builder method	Effect
`.withKeySerde(serde)` / `.withValueSerde(serde)`	override the key / value deserializer
`.withName(name)`	give the source node a stable name (same as `Named`)
`.withOffsetResetPolicy(AutoOffsetReset.…)`	per-source `auto.offset.reset` override
`.withWatermarkStrategy(strategy)`	per-source event-time + watermark strategy
`.withMaterializeFromSourceTopic(bool?)`	KTable changelog reuse (tables only)

When a serde is not set on Consumed, the source falls back to the default key/value serde configured for the application.

Consumed.as(...) and .withName(...) give the source a stable name that flows into the topology graph, metrics, and state-store identity. Naming your sources (and operators) makes /topology and your dashboards readable — see how the first app names every node.

Offset reset policy

AutoOffsetReset controls where the consumer starts when there is no committed offset for a partition. The per-source override beats the global auto.offset.reset config. It mirrors the KIP-1106 shape of Kafka's AutoOffsetReset: a sealed type with static factories — in Kotlin you can also reference the objects (AutoOffsetReset.Earliest) directly.

Factory	Behaviour
`AutoOffsetReset.earliest()`	start from the beginning of the partition
`AutoOffsetReset.latest()`	start from the end — only records produced after start
`AutoOffsetReset.none()`	throw if no committed offset exists
`AutoOffsetReset.byDuration(d)`	start from the first offset at/after `now − d` (KIP-1106)

import io.stoatflow.core.topology.AutoOffsetReset

// Historical/reference data — read everything from the start
builder.stream(
    "events",
    Consumed.with(Serdes.String(), eventSerde).withOffsetResetPolicy(AutoOffsetReset.Earliest),
)

// Real-time commands — only new records
builder.stream(
    "commands",
    Consumed.with(Serdes.String(), commandSerde).withOffsetResetPolicy(AutoOffsetReset.Latest),
)

import io.stoatflow.core.topology.AutoOffsetReset;

// Historical/reference data — read everything from the start
builder.stream(
    "events",
    Consumed.with(Serdes.String(), eventSerde).withOffsetResetPolicy(AutoOffsetReset.earliest()));

// Real-time commands — only new records
builder.stream(
    "commands",
    Consumed.with(Serdes.String(), commandSerde).withOffsetResetPolicy(AutoOffsetReset.latest()));

All sources reading the same topic must agree on the offset-reset policy; conflicting policies are rejected when the topology is built.

Timestamp extraction and watermarks

Event time and watermarks are configured per source through a WatermarkStrategy, set with Consumed.withWatermarkStrategy(...). The strategy does two jobs: it extracts the event timestamp from each record (via withTimestampAssigner), and it generates watermarks. With no strategy set, the source uses the application's global watermark strategy.

import io.stoatflow.core.watermark.WatermarkStrategy
import java.time.Duration

val strategy =
    WatermarkStrategy
        .forBoundedOutOfOrderness<String, Order>(Duration.ofSeconds(30))
        .withTimestampAssigner { _, value, _ -> value?.eventTime ?: 0L }
        .withIdleness(Duration.ofMinutes(2))

builder.stream(
    "orders",
    Consumed.with(Serdes.String(), orderSerde).withWatermarkStrategy(strategy),
)

import io.stoatflow.core.watermark.WatermarkStrategy;
import java.time.Duration;

WatermarkStrategy<String, Order> strategy =
    WatermarkStrategy
        .<String, Order>forBoundedOutOfOrderness(Duration.ofSeconds(30))
        .withTimestampAssigner((key, value) -> value != null ? value.eventTime() : 0L)
        .withIdleness(Duration.ofMinutes(2));

builder.stream(
    "orders",
    Consumed.with(Serdes.String(), orderSerde).withWatermarkStrategy(strategy));

The factory methods are WatermarkStrategy.forBoundedOutOfOrderness(maxOutOfOrderness), WatermarkStrategy.forMonotonousTimestamps(), and WatermarkStrategy.noWatermarks(). In Java, withTimestampAssigner also accepts a value-only Function<V, Long> or a key-and-value BiFunction<K, V, Long>, as shown above. For the full event-time model — watermarks, idleness, late-record handling — see Event time and watermarks.

Writing a sink — `to()` and `Produced`

KStream.to(topic, produced) writes the stream to a Kafka topic. It's a terminal operation — it returns nothing. Produced<K, V> configures the write side; when a serde is omitted, the sink falls back to the default serde.

import io.stoatflow.core.topology.Produced

stream.to(
    "output",
    Produced.with(Serdes.String(), Serdes.Long()),
)

// Named sink with serdes — readable in the topology graph
stream.to(
    "word-counts",
    Produced.`as`<String, Long>("sink")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long()),
)

import io.stoatflow.core.topology.Produced;

stream.to(
    "output",
    Produced.with(Serdes.String(), Serdes.Long()));

// Named sink with serdes — readable in the topology graph
stream.to(
    "word-counts",
    Produced.<String, Long>as("sink")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long()));

Produced factories mirror Consumed: Produced.with(keySerde, valueSerde), Produced.keySerde(serde), Produced.valueSerde(serde), Produced.streamPartitioner(partitioner), and Produced.as("name"). The chainable builders are .withKeySerde, .withValueSerde, .withStreamPartitioner, and .withName.

Custom partitioning

By default the producer partitions by key. Supply a StreamPartitioner<K, V> to control the target partition yourself — it receives the topic, key, value, and partition count, and returns an Optional<Set<Integer>> of 0-indexed partitions (KIP-837): Optional.empty() falls back to the default partitioner, a single-element set routes to that partition, an empty set drops the record, and multiple elements multicast (sink path only).

import io.stoatflow.core.topology.StreamPartitioner
import java.util.Optional

stream.to(
    "output",
    Produced.with(Serdes.String(), valueSerde)
        .withStreamPartitioner(
            StreamPartitioner { _, key, _, numPartitions ->
                Optional.of(setOf(Math.floorMod(key.hashCode(), numPartitions)))
            },
        ),
)

import io.stoatflow.core.topology.StreamPartitioner;
import java.util.Optional;
import java.util.Set;

stream.to(
    "output",
    Produced.with(Serdes.String(), valueSerde)
        .withStreamPartitioner(
            (StreamPartitioner<String, Value>) (topic, key, value, numPartitions) ->
                Optional.of(Set.of(Math.floorMod(key.hashCode(), numPartitions)))));

Dynamic topic routing

To pick the destination topic per record, pass a TopicNameExtractor<K, V> instead of a topic name. The extractor receives the key, value, and a RecordContext (source topic, partition, offset, timestamp, headers) and returns the topic name.

import io.stoatflow.core.topology.TopicNameExtractor

stream.to(
    TopicNameExtractor { _, value, _ -> "orders-${value.region}" },
    Produced.with(Serdes.String(), orderSerde),
)

import io.stoatflow.core.topology.TopicNameExtractor;

stream.to(
    (TopicNameExtractor<String, Order>) (key, value, ctx) -> "orders-" + value.region(),
    Produced.with(Serdes.String(), orderSerde));

For records that originate inside the topology (for example, results emitted as a window closes), the RecordContext source fields may be unknown: topic is null, and partition / offset are -1 (RecordContext.UNKNOWN_PARTITION / UNKNOWN_OFFSET). The timestamp is always available. Handle those cases when your routing depends on source metadata. Dynamic-sink topics are not part of the topology's static sink-topic set.

The fan-out rule

To process one source two (or more) ways, reuse the KStream reference — do not call builder.stream(...) twice on the same topic. A topic may only back one source node; a second subscription is rejected when the topology is built.

val orders = builder.stream("orders", Consumed.with(Serdes.String(), orderSerde))

// Branch 1
orders.filter { _, v -> v.isHighValue }.to("high-value-orders")

// Branch 2 — same reference, no second subscription
orders.mapValues { v -> v.summary() }.to("order-summaries")

KStream<String, Order> orders = builder.stream("orders", Consumed.with(Serdes.String(), orderSerde));

// Branch 1
orders.filter((k, v) -> v.isHighValue()).to("high-value-orders");

// Branch 2 — same reference, no second subscription
orders.mapValues(Order::summary).to("order-summaries");

The same rule applies to table, globalTable, and any other source. To merge two streams into one, use KStream.merge(...) (see the stock-tick-filter example); to split one stream into named branches by predicate, use split / branch. Fan-out and merge happen in-memory between processing lanes — no broker round-trip, no repartition topic. See Lanes and parallelism for the routing model.

Worked example

The runtime word count reads text-lines, counts by word, and writes word-counts — a String source and a Long sink:

builder
    .stream<String, String>("text-lines", Consumed.`as`("source"))
    .flatMapValues { line -> line.lowercase().split(Regex("\\s+")).filter { it.isNotBlank() } }
    .groupBy({ _, word -> word })
    .count(Materialized.`as`<String, Long, StateStore>("word-counts"))
    .toStream()
    .to(
        "word-counts",
        Produced.`as`<String, Long>("sink")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long()),
    )

Next steps

KStream and KTable — the operators between source and sink: map, filter, merge, split, table conversions.
Serdes — choosing and configuring serializers for keys and values.
State stores — Materialized and the store types behind table and the aggregations.
Scheduled sources — emit records on an interval or cron schedule without reading from Kafka.
Event time and watermarks — the full event-time model behind withWatermarkStrategy.

Testing topologies

Unit-test a StoatFlow topology in-memory with TopologyTestDriver — pipe records, advance time and watermarks, read state stores, and drive YAML config through StoatFlowTestDriver. No broker required.

KStream and KTable operations

Stateless stream transforms (map, filter, flatMap, selectKey, branch, merge) and KTable basics — with the rules for when a key change forces records onto a different lane.