Sources and sinks

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 a new topic matching the pattern is picked up within the consumer's metadata.max.age.ms (StoatFlow defaults it to 30 s for a topology with a pattern source; Kafka's default is 5 minutes) — as an incremental assignment, with no restart.

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

How pattern sources behave

• Several patterns are fine; each source declares one pattern. A topology may declare any number of stream(Pattern) sources (plus literal topics) — they are combined into the one subscription a Kafka consumer accepts, as in Kafka Streams. Two sources on the same pattern are rejected at build time; a literal source topic that a pattern also matches is rejected at build time, naming both; a topic matched by two different patterns is rejected loudly when it is assigned (the application terminates naming the topic and both sources — narrow the patterns or rename the topic, exactly the Kafka Streams rule).
• matches(), not find() — the pattern must match the whole topic name, the way the Kafka consumer itself evaluates a subscription; a pattern's own flags (CASE_INSENSITIVE, LITERAL, …) are honoured. Literal topics next to a pattern are quoted into the combined subscription, so my.topic does not also subscribe myXtopic (Kafka Streams appends them unquoted). Kafka's internal topics (__consumer_offsets, …) are never matched, even by .*.
• Matching nothing is not an error. If no topic matches at startup the application starts, logs a warning naming the pattern, reports stoatflow.source.pattern.matched.topics{source="…"} as 0, and picks up the first matching topic when it appears.
• Joins, cogroups, the test driver, Consumed.withOffsetResetPolicy and a per-source WatermarkStrategy all work on a pattern source — the resolution rule is shared between the engine and TopologyTestDriver.
• Deleting a matched topic is handled for you. StoatFlow recognises that the topic left the subscription (rather than that a second instance joined), restarts its engine in place, and carries on with the remaining topics — typically well under a second, with no pod restart. The in-flight batch is discarded rather than committed, so the surviving topics resume from their last committed offsets and reprocess anything read but not yet committed — exactly-once still holds for them. Records read from the deleted topic but not yet committed are simply gone; the topic no longer exists to re-read them from, which is also what Kafka Streams' task close implies. A partition-count increase is picked up live. Under hot-standby HA the same change is absorbed by the active's own in-place restart.
• Deleting a literal source topic still stops the application, which is what Kafka Streams does too (MissingSourceTopicException). Only pattern-matched topics are expected to come and go.
• A topic that briefly becomes unreachable — a broker rolling restart, a controller failover — can momentarily look like a deletion to any Kafka consumer, Kafka Streams included. StoatFlow answers it the same way as a real deletion, and because the in-flight epoch is discarded rather than committed, nothing is duplicated when the topic reappears.

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, and new StoatFlow code should prefer table(...).

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.

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

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.

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.

Aligning sources — withWatermarkAlignment

Sources that share an alignment group can be kept from drifting apart in event time with Flink's withWatermarkAlignment(group, maxAllowedWatermarkDrift[, updateInterval]):

fun aligned(): WatermarkStrategy<String, Order> =
    WatermarkStrategy
        .forBoundedOutOfOrderness<String, Order>(Duration.ofSeconds(5)) // the same bound on every source of the group
        .withIdleness(Duration.ofMinutes(1)) // required with alignment
        .withWatermarkAlignment("orders-and-payments", Duration.ofSeconds(30))

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

WatermarkStrategy<String, Order> aligned =
    WatermarkStrategy
        .<String, Order>forBoundedOutOfOrderness(Duration.ofSeconds(5)) // the same bound on every source of the group
        .withIdleness(Duration.ofMinutes(1)) // required with alignment
        .withWatermarkAlignment("orders-and-payments", Duration.ofSeconds(30));

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

A partition more than the drift ahead of a lagging partition of its group is held: it keeps being fetched, and its records go only in timestamp order. When its next record is the oldest buffered one and a lagging partition has nothing buffered, the whole application waits — nothing may overtake the oldest record — until that partition sends a record or is declared idle. Size the drift to at least the sparse source's usual gap between records plus the difference in out-of-orderness bounds (drift ≥ G + b(sparse) − b(held)), or the application stops for most of every gap. Two build-time rules guard the configuration: alignment-idleness-off (an error by default) refuses an aligned source with no idleness timeout, and alignment-mixed-watermark-bounds (a warning) flags a group whose out-of-orderness bounds differ by more than the drift — see Topology validation. The full behaviour, the sizing rule and how to watch it are on Watermark alignment.

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

The interface has the Kafka Streams 4.4 shape (KIP-1321). Implement partitions(topic, key, value, headers, numPartitions) — StoatFlow always calls this one, with the record's headers:

import io.stoatflow.core.topology.StreamPartitioner
import org.apache.kafka.common.header.Headers
import java.util.Optional

/** Routes by a `tenant` header, else by key. */
class TenantPartitioner : StreamPartitioner<String, Value> {
    @Deprecated("Implement the headers overload")
    override fun partitions(topic: String, key: String, value: Value, numPartitions: Int): Optional<Set<Int>> =
        Optional.of(setOf(Math.floorMod(key.hashCode(), numPartitions)))

    override fun partitions(
        topic: String,
        key: String,
        value: Value,
        headers: Headers,
        numPartitions: Int,
    ): Optional<Set<Int>> {
        val tenant = headers.lastHeader("tenant") ?: return partitions(topic, key, value, numPartitions)
        return Optional.of(setOf(Math.floorMod(String(tenant.value()).hashCode(), numPartitions)))
    }
}

stream.to("output", Produced.with(Serdes.String(), valueSerde).withStreamPartitioner(TenantPartitioner()))

import io.stoatflow.core.topology.StreamPartitioner;
import java.util.Optional;
import java.util.Set;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;

/** Routes by a {@code tenant} header, else by key. */
final class TenantPartitioner implements StreamPartitioner<String, Value> {
    @Override
    @SuppressWarnings("removal")
    public Optional<Set<Integer>> partitions(String topic, String key, Value value, int numPartitions) {
        return Optional.of(Set.of(Math.floorMod(key.hashCode(), numPartitions)));
    }

    @Override
    public Optional<Set<Integer>> partitions(
            String topic, String key, Value value, Headers headers, int numPartitions) {
        Header tenant = headers.lastHeader("tenant");
        if (tenant == null) {
            return partitions(topic, key, value, numPartitions);
        }
        return Optional.of(Set.of(Math.floorMod(new String(tenant.value()).hashCode(), numPartitions)));
    }
}

stream.to("output", Produced.with(Serdes.String(), valueSerde).withStreamPartitioner(new TenantPartitioner()));

The 4-arg partitions(topic, key, value, numPartitions) is the interface's single abstract method, so a lambda still works — Kafka Streams 4.4 deprecated it for removal, and you will see that warning. A lambda cannot override the headers method, so it never sees the headers; use a class when you need them.

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

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

Partition choice must be a pure function of the record, headers included: a replayed epoch has to route every record where it went the first time. On repartition() the partitioner picks an in-memory lane rather than a Kafka partition — numPartitions is the lane count, multicast is rejected, and a partitioner feeding a stateful operator must send each key (with its headers) to one lane consistently.

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

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.