Joins

A join combines records from two sources that share a key. StoatFlow's join DSL is Kafka Streams compatible: the same five join families — stream-stream, stream-table, table-table, foreign-key, and co-grouping — with the same ValueJoiner, JoinWindows, Joined, StreamJoined, and TableJoined configuration types. Java code written against the KS join API compiles against StoatFlow unchanged (see the Kafka Streams compatibility matrix).

One structural difference is worth stating up front: because StoatFlow holds global state in a single instance, the two sides of a join do not have to be co-partitioned. There is no requirement that both topics share a partition count. Re-keying for a foreign-key join happens in-memory between lanes, not through a repartition topic.

The joiner

Every join takes a joiner — the function that combines a left value and a right value into the result. Two interfaces:

For leftJoin the right value may be null (no match on the right). For outerJoin either side may be null, but never both. Write the joiner to tolerate the nulls the join type allows.

Stream-stream joins

A stream-stream join matches records from two streams that arrive close together in event time. Both sides are buffered in window stores, and a record from one stream joins any record from the other whose timestamp falls inside the configured JoinWindows. The join is symmetric — a late-arriving record on either side can produce a match.

JoinWindows defines the time bounds. ofTimeDifferenceWithNoGrace(d) is a symmetric ±d window; before(d) / after(d) make it asymmetric; ofTimeDifferenceAndGrace(d, grace) adds a grace period for late records (see Event time and watermarks).

import io.stoatflow.core.topology.JoinWindows
import io.stoatflow.core.topology.StreamJoined
import org.apache.kafka.common.serialization.Serdes
import java.time.Duration

// Inner: emit only when both an order and a payment land within 5 minutes
val matched =
    orders.join(
        payments,
        { order, payment -> OrderPayment(order, payment) },
        JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
    )

// With serdes + a named, materialized join (recommended in production)
val matchedConfigured =
    orders.join(
        payments,
        { order, payment -> OrderPayment(order, payment) },
        JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
        StreamJoined.with(Serdes.String(), orderSerde, paymentSerde)
            .withStoreName("order-payment-join"),
    )

import io.stoatflow.core.topology.JoinWindows;
import io.stoatflow.core.topology.StreamJoined;
import org.apache.kafka.common.serialization.Serdes;
import java.time.Duration;

// Inner: emit only when both an order and a payment land within 5 minutes
var matched =
    orders.join(
        payments,
        (order, payment) -> new OrderPayment(order, payment),
        JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)));

// With serdes + a named, materialized join (recommended in production)
var matchedConfigured =
    orders.join(
        payments,
        (order, payment) -> new OrderPayment(order, payment),
        JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
        StreamJoined.with(Serdes.String(), orderSerde, paymentSerde)
            .withStoreName("order-payment-join"));

join, leftJoin, and outerJoin are all available, each in value-only and key-aware forms:

• join (inner) — emit only when both sides match within the window.
• leftJoin — emit for every left record; the right value is null when nothing matched by the time the window closes.
• outerJoin — emit for every record on either side; the unmatched side is null when the window closes.

Both sides may come from the same source: a self-join (events.join(events, ...)), or a stream joined with a processed copy of itself (events.join(events.mapValues(...), ...)). Every record is buffered on both sides and matched as each, as in Kafka Streams.

Configuring the join with StreamJoined

StreamJoined configures the two window stores that back a stream-stream join, plus naming and serdes:

A key-aware joiner uses the same calls — pass a ValueJoinerWithKey instead:

import io.stoatflow.core.topology.ValueJoinerWithKey

val labelled =
    orders.join(
        payments,
        ValueJoinerWithKey<String, Order, Payment, String> { customerId, order, payment ->
            "$customerId: ${order.total + payment.amount}"
        },
        JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
    )

import io.stoatflow.core.topology.ValueJoinerWithKey;

var labelled =
    orders.join(
        payments,
        (ValueJoinerWithKey<String, Order, Payment, String>)
            (customerId, order, payment) ->
                customerId + ": " + (order.total() + payment.amount()),
        JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)));

Stream-table joins

A stream-table join is an enrichment lookup: each stream record is joined against the current value for its key in a KTable. This is a point-in-time lookup — unlike a table-table join, updates to the table do not re-trigger the join. Only stream records produce output. The table must be materialized (created via builder.table(...) or stream.toTable(...) with a store).

The stream key is preserved, so the join stays on the same lane — no re-keying.

import io.stoatflow.core.topology.Joined

// Inner: enrich each click with the user record, drop clicks with no user
val enriched =
    clicks.join(users) { click, user -> EnrichedClick(click, user) }

// Left: keep every click; user is null when the table has no entry for the key
val enrichedAll =
    clicks.leftJoin(
        users,
        Joined.`as`("click-user-join"),
    ) { click, user -> EnrichedClick(click, user) }

import io.stoatflow.core.topology.Joined;

// Inner: enrich each click with the user record, drop clicks with no user
var enriched =
    clicks.join(users, (click, user) -> new EnrichedClick(click, user));

// Left: keep every click; user is null when the table has no entry for the key
var enrichedAll =
    clicks.leftJoin(
        users,
        (click, user) -> new EnrichedClick(click, user),
        Joined.as("click-user-join"));

Joined.as(name) names the join processor — that's its only effective setting. For Kafka Streams source compatibility Joined also accepts serdes (withKeySerde / withValueSerde / withOtherValueSerde) and withGracePeriod(...), but these are advisory in StoatFlow: they're stored so KS code compiles, while serde resolution actually comes from Consumed/Grouped upstream (a documented divergence — see the compatibility matrix). Key-aware joiners (ValueJoinerWithKey) are available for both join and leftJoin. There is no outerJoin for stream-table joins — an outer join has no stream record to anchor an unmatched table entry to.

Joining a GlobalKTable

A GlobalKTable join looks the table up by a key you derive from each stream record, instead of by the stream key. A keySelector maps (key, value) to the lookup key; returning null means "no lookup". The output keeps the stream key.

To use either key in the result, pass a ValueJoinerWithStreamAndMappedKey: it receives the stream record's key, the mapped lookup key, and both values (keys first, as in Kafka Streams 4.4):

// Each order value carries the product id it refers to; the output stays keyed by the order id
val enriched =
    orders.join(products, { _, order -> order.productId }) { orderId, productId, order, product ->
        EnrichedOrder(orderId, productId, order, product)
    }

// Left join: product is null when there is no match; productId is null when the selector returned null
val enrichedAll =
    orders.leftJoin(products, { _, order -> order.productId }) { orderId, productId, order, product ->
        EnrichedOrder(orderId, productId, order, product)
    }

import io.stoatflow.core.topology.Named;

// Each order value carries the product id it refers to; the output stays keyed by the order id
var enriched =
    orders.join(
        products,
        (orderId, order) -> order.productId(),
        (orderId, productId, order, product) -> new EnrichedOrder(orderId, productId, order, product));

// Left join: product is null when there is no match; productId is null when the selector returned null
var enrichedAll =
    orders.leftJoin(
        products,
        (orderId, order) -> order.productId(),
        (orderId, productId, order, product) -> new EnrichedOrder(orderId, productId, order, product),
        Named.as("order-product-join"));

To name the join processor from Kotlin, pass a Named before the trailing joiner — orders.leftJoin(products, { _, order -> order.productId }, Named.`as`("enrich")) { … }; from Java, as the last argument.

A plain ValueJoiner works too. The ValueJoinerWithKey overloads compile, but they are deprecated (as in Kafka Streams 4.4, KIP-1340): their key is the stream key, not the lookup key the keySelector produced, so it does not necessarily match the table record being joined. Move to ValueJoinerWithStreamAndMappedKey when you need a key.

An inner join drops a record whose value is null or whose keySelector returns null. A left join drops only a null value; with a null lookup key it still calls the joiner, as (streamKey, null, value, null).

Under exactly_once the lookup reads the global table as of the stream record's own commit, like every other join (see Joins under exactly-once). That is a stronger guarantee than Kafka Streams gives, where a separate global thread updates the table with no consistent cut against the stream tasks. The API is the same.

Table-table joins

A table-table join matches the two tables by key: for each key present in both, it emits the joined value, and it re-emits whenever either side updates. The result is itself a KTable, so it carries changelog (tombstone) semantics — a deletion on the relevant side deletes the join result.

Both tables must be materialized. The three join types follow KS semantics exactly:

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

// Inner join on equal keys
val joined =
    accounts.join(balances) { account, balance -> AccountBalance(account, balance) }

// Outer join, named and materialized into its own store
val outer =
    accounts.outerJoin(
        balances,
        { account, balance -> AccountBalance(account, balance) },
        Named.`as`("account-balance-join"),
        Materialized.`as`<String, AccountBalance, StateStore>("account-balance"),
    )

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.Materialized;
import io.stoatflow.core.topology.Named;

// Inner join on equal keys
var joined =
    accounts.join(balances, (account, balance) -> new AccountBalance(account, balance));

// Outer join, named and materialized into its own store
var outer =
    accounts.outerJoin(
        balances,
        (account, balance) -> new AccountBalance(account, balance),
        Named.as("account-balance-join"),
        Materialized.<String, AccountBalance, StateStore>as("account-balance"));

Each of join / leftJoin / outerJoin has overloads taking an optional Named and an optional Materialized — name the node, materialize the result, or both.

Foreign-key joins

A foreign-key (FK) join joins two tables on a key extracted from the left table's value rather than on the record key. This is the table-table equivalent of a relational JOIN ... ON left.fk = right.id. The right table is keyed by the foreign key; the result keeps the left table's key.

The signature is left.join(right, foreignKeyExtractor, joiner, tableJoined, ...):

• foreignKeyExtractor derives the right table's key from the left value — (V) -> KO? (value only) or (K, V) -> KO? (key + value). A null foreign key means there is no foreign row to join: an inner join produces no result for it, a left join produces joiner(left, null), as in Kafka Streams.

When a join result goes away — the left row is deleted, its foreign key changes to a row that doesn't exist or to null, or the foreign row is deleted — the join forwards a (key, null) tombstone, whether or not the join is materialized, exactly as Kafka Streams does. Like Kafka Streams, an inner join also sends one for a key whose old foreign row never existed — a harmless no-op delete downstream; a foreign-row delete tombstones every order that referenced it.

• joiner combines the left value with the matched right value.
• TableJoined.as(name) names the join processors (its only setting).

Both inner (join) and left (leftJoin) FK joins are supported; outer is not — there is no left key to anchor an unmatched right row to. The FK join re-emits when the left value changes (re-extract the FK, look up the right table) and when the right value changes (re-evaluate every left row that points at it).

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

// orders keyed by orderId, customers keyed by customerId.
// Extract customerId from each order, look it up in customers.
val enrichedOrders =
    orders.join(
        customers,
        { order -> order.customerId },           // foreign-key extractor (left value -> right key)
        { order, customer -> EnrichedOrder(order, customer) },
        TableJoined.`as`("order-customer-join"),
        Materialized.`as`<String, EnrichedOrder, StateStore>("enriched-orders")
            .withKeySerde(Serdes.String())
            .withValueSerde(enrichedOrderSerde),
    )

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.Materialized;
import io.stoatflow.core.topology.TableJoined;

// orders keyed by orderId, customers keyed by customerId.
// Extract customerId from each order, look it up in customers.
var enrichedOrders =
    orders.join(
        customers,
        order -> order.customerId(),             // foreign-key extractor (left value -> right key)
        (order, customer) -> new EnrichedOrder(order, customer),
        TableJoined.as("order-customer-join"),
        Materialized.<String, EnrichedOrder, StateStore>as("enriched-orders")
            .withKeySerde(Serdes.String())
            .withValueSerde(enrichedOrderSerde));

The Java foreign-key overloads take java.util.function.Function<V, KO> (or BiFunction<K, V, KO> for the key-aware form) for the extractor — the same shape as the current Kafka Streams FK-join API — and ValueJoiner<V, VO, VR> for the joiner, so a method reference like Order::customerId works as the extractor and a constructor reference like EnrichedOrder::new works as the joiner.

Joins under exactly-once

Several joins read a table that another lane writes: both sides of a table-table join and of a foreign-key join, a stream-table join after a re-key, and a stream–GlobalKTable join. Under exactly_once (the default) each lane reads the table as of its own commit — never a value another lane has already written for the next, still-uncommitted commit — so a join never commits a result built on an uncommitted value. Under at_least_once a join reads the latest value.

That needs a store that can answer "as of this commit". The default stores can — RocksDB and in-memory key-value stores batch their writes per commit — and so can the RocksDB versioned store, Stores.persistentVersionedKeyValueStore(...). Two stores cannot, because they write directly: Stores.lruMap(...) and Stores.inMemoryVersionedKeyValueStore(...).

Co-grouping

Co-grouping aggregates multiple grouped streams of different value types into one result, each stream contributing through its own aggregator. It's the join-like answer to "fold orders and payments for the same customer into one summary" — without windowing the two sides against each other.

Start a cogroup from a KGroupedStream, chain .cogroup(other) { ... } for each additional stream, then .aggregate(initializer):

import io.stoatflow.core.topology.Materialized

val orders = orderStream.groupByKey()       // KGroupedStream<String, Order>
val payments = paymentStream.groupByKey()   // KGroupedStream<String, Payment>

val summary: KTable<String, CustomerSummary> =
    orders
        .cogroup { _, order, agg -> agg.addOrder(order) }
        .cogroup(payments) { _, payment, agg -> agg.addPayment(payment) }
        .aggregate(
            { CustomerSummary() },
            Materialized.`as`("customer-summary"),
        )

import io.stoatflow.core.topology.Materialized;

var orders = orderStream.groupByKey();       // KGroupedStream<String, Order>
var payments = paymentStream.groupByKey();   // KGroupedStream<String, Payment>

KTable<String, CustomerSummary> summary =
    orders
        .cogroup((key, order, agg) -> agg.addOrder(order))
        .cogroup(payments, (key, payment, agg) -> agg.addPayment(payment))
        .aggregate(
            CustomerSummary::new,
            Materialized.as("customer-summary"));

The co-grouped streams may share a source — orders.cogroup { ... }.cogroup(orders.mapValues { ... }.groupByKey()) { ... } folds every record through each aggregator. Co-grouping the same KGroupedStream a second time replaces its aggregator rather than adding one, as in Kafka Streams.

aggregate has overloads taking an optional Named and an optional Materialized. Co-grouping can also be windowed — call .windowedBy(TimeWindows...) or .windowedBy(SessionWindows...) before aggregate; see Windowing. For the non-co-grouped single-stream folds, see Aggregations.

Choosing a join

Where to go next

• Aggregations — group and fold a single stream or table
• Windowing — windowed aggregations and windowed co-grouping
• Serdes — the serializers join stores need for keys and both value sides
• State stores — store types behind join state
• Event time and watermarks — what drives stream-stream window closing