The self-join that emitted nothing

Join a stream with itself on Kafka Streams — s.join(s) — and pipe two records with the same key, a then b, inside the join window. Kafka Streams emits four results: a+a, b+a, a+b, b+b. StoatFlow emitted none. No exception, no log line, no metric moved; the output topic stayed empty.

The cause was one decision in how StoatFlow's multi-input operators worked out where a record came from: a stream–stream join picked its side, and a cogroup its aggregator, by the name of the record's source node. A self-join's records all come from one source, so every record was processed as the left side and the right side's store stayed empty. The same rule broke chained joins, joins over a merge, records forwarded from timers and window closes, and a cogroup of two groupings of one stream. This post lists the shapes, shows how Kafka Streams' own test driver served as the oracle that found them, and describes the fix: the input travels with the edge.

TL;DR
What: a join chose left or right, and a cogroup its aggregator, by the record's source node name — not by the edge the record arrived on.
Symptom: a self-join emitted nothing; chained and merged legs lost one side's records; records from timers, punctuators and window closes were dropped before a join. Silently.
How we found every shape: the same topology built and run on both Kafka Streams' TopologyTestDriver and StoatFlow's, outputs compared. On the old code every case but the control failed.
The fix: input i of a node is position i among its parents, and that input number travels with the edge — inside a sub-topology and across a lane hop.
Docs: Joins · Testing.

How a join knew its side

In Kafka Streams each side of a stream–stream join is its own processor: the left stream feeds a "this side" processor, the right stream feeds an "other side" processor, and a self-join wires one stream into both. A record runs through each side because it travels along each edge.

StoatFlow's join was one processor with two inputs, and it decided which input a record was on by comparing the record's source node name with the names it had resolved for its left and right parents. A stream derived from a source — through mapValues, selectKey or a merge — keeps its parent's source name. So the decision went wrong wherever two inputs shared a source, or a record's source name was not one of the two the join expected. The four cogroup operators did the same with a map from source name to aggregator, and skipped a record whose name was missing.

Every shape it broke

Shape	Kafka Streams	StoatFlow before the fix
`s.join(s)`	`a+a`, `b+a`, `a+b`, `b+b`	inner join: nothing; left and outer joins: only `(left, null)`
`s.selectKey(f1).join(s.selectKey(f2))`	both sides joined	as above — both legs carry `s`'s source name
`a.join(b).join(c)`	every first-join result joins `c`	results triggered by a `b` record dropped: the second join's left side was named after `a`
`s1.merge(s2).join(s3)`	`one+R`, `two+R`	every `s2` record dropped
a window close, timer or punctuator forwarding into a join	joined	dropped — such records carry no source the join recognises
two cogrouped groupings of one stream	`[Aa]`, `[Aa][Ba]`, `[Aa][Ba][Ab]`, `[Aa][Ba][Ab][Bb]`	one aggregator overwrote the other
`s.merge(s.mapValues(f))`	each record once, plus its mapped copy	the mapped copy never ran — only one chain was kept per source

In StoatFlow's own TopologyTestDriver it went further: every record reaching a join or cogroup through any node other than its source was dropped, so even stream.mapValues(…).join(…) produced nothing there. The last row is the odd one out: the test driver was not affected, and only an integration test on a real broker showed it.

Kafka Streams as the oracle

The tempting way to test a fix like this is to write down the expected output by hand. For joins that is error-prone — window semantics, the order in which a self-join emits, which side a merged record lands on. So the tests do not write it down. They build each topology twice, once with Kafka Streams' StreamsBuilder and once with StoatFlow's, pipe the same input into Kafka Streams' TopologyTestDriver and StoatFlow's, and compare the output values in order:

@Test
fun `self-join runs every record as both sides`() {
    val expected =
        assertMatchesKs(
            ab, // two records, same key, values a then b, inside the join window
            { b -> // Kafka Streams' StreamsBuilder
                val s = b.stream<String, String>("in")
                s.join(s, { l, r -> "$l+$r" }, KsJoinWindows.ofTimeDifferenceWithNoGrace(window)).to("out")
            },
            { b -> // StoatFlow's StreamsBuilder
                val s = b.stream<String, String>("in")
                s.join(s, { l, r -> "$l+$r" }, JoinWindows.ofTimeDifferenceWithNoGrace(window)).to("out")
            },
        )
    expected shouldContainExactly listOf("a+a", "b+a", "a+b", "b+b")   // pins the oracle itself
}

Three details keep the comparison honest. Kafka Streams' record cache is set to zero, so each cogroup update is emitted rather than folded into one per input record. Order is relaxed in two places only: a source joined with a processed copy of itself, where the fan-out's two branches may run in either order, and one Processor API fan-in case. And where the test also states an expected list, it checks Kafka Streams against it, so a misunderstanding of Kafka Streams' behaviour fails the test too.

The suite has 17 cases, 15 of them compared with Kafka Streams. The other two check hand-written values: a window close feeding a join, which StoatFlow closes on the watermark — a clock Kafka Streams' driver does not have — and a scheduled source as a cogroup leg, which Kafka Streams does not have at all. In the first red run on the old code, every case but the control failed, 12 of them; four more were added during review. A broker integration test runs the main shapes end to end — the self-join, two re-keyed legs, a merged leg, the cogroup of two groupings, a window close into a join, a source merged or joined with a processed copy of itself — plus the Processor API fan-in shapes.

The fix: the edge carries the input

A node's inputs are now defined by its parents: input i is position i in the node's parent list at build time, the same rule Kafka Streams' wiring implies. The input number travels with the edge. Inside a sub-topology, each forward to a child carries the input that child receives on; when a record crosses to another lane, the input travels on the record itself. The join maps input 0 to left and 1 to right; a cogroup holds its aggregators in input order.

What that means in practice:

A self-join runs each record as left, then as right, as Kafka Streams does, so s.join(s) emits all four results.
A Processor API node that lists the same parent twice runs twice — every edge keeps its slot, as in Kafka Streams; a forward by name to such a child runs it once, also as in Kafka Streams. KStream.merge de-duplicates its parents, so s.merge(s) emits once.
Records from callbacks join normally, because the input comes from the edge, not from the record's origin.
Every source dispatches into exactly one chain. A source whose records also feed a node beyond its own first sub-topology reaches it through an ordinary lane hop, so s.merge(s.mapValues(f)) runs both legs.
An unknown input is fatal. The compiler checks every edge's input number. A mismatch at runtime — which would mean the compiled metadata drifted — fails loudly instead of being dropped like before.

Tradeoffs and limits

The oracle only covers what Kafka Streams can express. Two of the cases cannot run on Kafka Streams' driver and check hand-written values; StoatFlow-specific behaviour — lanes, one global watermark — needs broker tests and invariants, not a reference driver. The comparison also checks values, not record timestamps.

Some fan-in shapes now pay a lane hop. A node whose parents' chains cannot be joined, or whose parents sit at different depths of one chain, moves into its own sub-topology, so every edge into it crosses to a lane. The hop is an in-memory hand-off with no repartition topic, but it is still work the old, wrong routing skipped. A source whose only edge leads into a merge that a processed leg also feeds gets a sub-topology of its own that only forwards: its own lanes, and a key hash and a queue hop per record. Consumed.withNumberOfLanes(...) on the source sizes it down.

None of it was visible. None of these shapes threw, logged or counted anything; an empty output topic looks the same as no input. The comparison against Kafka Streams is now a permanent part of the build, and an input the compiler cannot account for stops the application rather than dropping the record.

Read on:

Joins — stream–stream, stream–table and table–table joins in StoatFlow.
Testing — the TopologyTestDriver, and what it cannot show.
Kafka Streams compatibility matrix — where StoatFlow matches Kafka Streams, and where it deliberately does not.
Lanes and parallelism — the in-memory lane hop that replaces a repartition topic.