Joins

Combine two streams or tables on a shared key — stream-stream windowed joins, stream-table enrichment lookups, table-table joins, foreign-key joins, and co-grouping.

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:

Interface	Signature	When
`ValueJoiner<V1, V2, VR>`	`(left, right) -> result`	The result depends only on the two values.
`ValueJoinerWithKey<K, V1, V2, VR>`	`(key, left, right) -> result`	The result also needs the join key. The key is read-only.

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.

Unmatched leftJoin / outerJoin results are emitted when the window closes, which is driven by the watermark — not on the bare arrival of the left record. A join window with no grace closes as soon as the watermark passes the window end.

Configuring the join with StreamJoined

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

Method	Effect
`StreamJoined.as(name)`	Name the join processor.
`StreamJoined.with(keySerde, leftValueSerde, rightValueSerde)`	Set the serdes for both sides.
`.withStoreName(base)`	Derive store names `{base}-left-store` and `{base}-right-store`.
`.withThisStoreSupplier(...)` / `.withOtherStoreSupplier(...)`	Supply explicit window stores ("this" = the stream `join` is called on; "other" = the argument).
`.withDslStoreSuppliers(StoreType.IN_MEMORY)`	Choose in-memory or RocksDB for both stores (takes precedence over explicit suppliers).
`.withLoggingEnabled(...)` / `.withLoggingDisabled()`	Force changelog logging on/off for the join stores.

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.

Worked example. The news-feed-subscription-processor example enriches a windowed result stream with subscriber and user detail held in a KTable, using exactly this shape — subscriptionResults.join(subscriptionAndUserTable, joiner, Joined.as("...")) — to build the final notification before writing to the sink.

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:

Method	Emits	Result deleted when
`join` (inner)	only when both keys are present	either side becomes a tombstone
`leftJoin`	for every left update; right may be `null`	the left side becomes a tombstone
`outerJoin`	for every update to either side; either may be `null`	both sides become tombstones

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 no join is possible and the record is ignored.
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.

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

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

You have	You want	Use
Two event streams, matches close in time	Correlate within a time window	stream-stream (`JoinWindows`)
An event stream + a lookup table	Enrich each event with current table state	stream-table
Two changelog tables, same key	A joined table that updates with either side	table-table
Two tables joined on a value field	Relational-style FK lookup	foreign-key
Several grouped streams, one key	Fold all of them into one aggregate	co-grouping

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

Windowing

Windowed aggregations in StoatFlow — tumbling, hopping, sliding, and session windows; emit strategies, suppression, Windowed keys, and WindowedSerdes.

The Processor API

Write custom record-by-record logic with Processor and FixedKeyProcessor — forward, access state, schedule punctuators, and register key-based timers.