Internal consistency on Kafka: emitting a correct answer at every commit

TL;DR

• The test: a stream of $1 transfers between 10 accounts. credits − debits = balance per account; total = Σ balance. Money is only moved, so total must be 0 at every consistent cut.
• What everyone else does: the Flink Table API emits 37,978,385 total values of which 13,325 are correct — 0.035%; ksqlDB emits 340, of which only the last is right; our own Kafka Streams twin sends total to −1,619 … +1,792 and emits 160 impossible balances on a dataset where the only valid values are {−1, 0, +1}.
• What StoatFlow does: zero impossible balances in every mode, and under emit.mode = on-barrier total is exactly 0 at every committed cut — measured on a 5-node cluster at 2,000 tx/s, verified across churn and unique-key workloads.
• Why: the commit barrier is a genuine Chandy–Lamport cut aligned to Kafka offsets, and the cascade-join withholds it from a downstream operator until every contributing chain has pushed its epoch into it. That cross-stream synchronization is exactly what the other engines lack.
• The honest part: getting to perfect took one real bug fix (a barrier-cut race across sub-topology boundaries). We found it, root-caused it, fixed it, and re-ran the benchmark to zero.

This post is about a property Jamie Brandon named internal consistency, the fact that no engine with a Kafka Streams–shaped API has it, and the measured demonstration that StoatFlow does.

The money test

Take the simplest invariant in accounting: money is only moved, never created or destroyed. Model it as a stream.

// transactions: (from, to) — each moves $1
val tx = builder.stream<String, Transfer>("transactions")

val credits = tx.map { _, t -> KeyValue(t.to,   1L) }.groupByKey().reduce { a, b -> a + b }   // received
val debits  = tx.map { _, t -> KeyValue(t.from, 1L) }.groupByKey().reduce { a, b -> a + b }   // sent

val balance = credits.outerJoin(debits) { c, d -> (c ?: 0) - (d ?: 0) }   // per-account net
val total   = balance.groupBy { _, b -> KeyValue("ALL", b) }.reduce({ a, b -> a + b }, { a, b -> a - b })

balance for any account is always in {−1, 0, +1} for a ring of $1 transfers. And total, the sum over all accounts, is always 0 — every credit has a matching debit. That is the invariant. It is not an approximation that holds "eventually"; it holds at every point in time, because it is a conservation law.

So here is the question: when your streaming engine emits a total value, is it 0?

What "internal consistency" means

Brandon's definition: a system is internally consistent if every output it produces is the correct output for some subset of the inputs it has seen so far. Not the latest subset, necessarily — but a coherent one. Every emitted row is a photograph of a real moment, never a double-exposure that blends two.

This is strictly stronger than eventual consistency, and the two don't imply each other. An eventually-consistent system is allowed to emit nonsense in the meantime as long as it converges. An internally consistent system is never allowed to emit nonsense at all — every intermediate output is itself a valid answer.

For the money test, internal consistency means: every emitted total is 0. If you ever see total = 3, the engine has shown you a world in which three dollars materialized from nothing — a subset of the inputs that never existed. That output is not "a little stale". It is wrong.

Brandon ran the test across engines, and the counts are the argument.

Materialize (and the differential dataflow it is built on) passes: 599 balance updates, and total is emitted exactly once, because the correct answer never changes.

The Flink Table API emits 18,999,012 balance updates and 37,978,385 total values. 13,325 of them are 0 — 0.035%. In one run, 26.3% of the emissions were a single value, 256, which the author could not explain and which differs run to run.

ksqlDB emits 340 total values, and in Brandon's words, "none of the outputs are correct until the last." On his simplified dataset its per-account balances — which can only ever be −1, 0 or +1 — reach −135,051 … +2,848.

Kafka Streams he couldn't get to produce output for the query at all (KAFKA-12594), which is why we built our own twin below.

Why no Kafka-shaped engine has it

The failure isn't a bug in any of these engines. It's architectural, and it has the same root everywhere.

balance = credits ⋈ debits is a join of two aggregations that fan out from one source. In Kafka Streams (and anything shaped like it), the re-key sends the two halves of a single transfer down two separate repartition topics, consumed independently, each feeding an aggregation with its own state store and its own record cache that evicts and emits an intermediate whenever it fills. The join is driven by whichever half arrives — so when a transaction updates both, the join will, sooner or later, read the new credit against the old debit — one side at a different logical time than the other. The result is a balance that never existed, and a total that isn't 0.

There is no cut that spans the topology. Each operator has its own notion of "now". The record cache consolidates the volume of output — Kafka Streams emits far fewer rows than Flink — but each row it does emit is still a blend of two moments. Consolidated garbage is still garbage.

Our own Kafka Streams twin of the benchmark reproduces exactly this: 160 impossible balances (individual values reaching 180 where only {−1, 0, +1} is possible), and total wandering to −1,619 … +1,792. The same phenomenon the article finds in ksqlDB, whose balances reach 135,051 on the same impossible quantity — less extreme here only because our churn rate is lower.

The load-bearing insight: exactly-once is not internal consistency

It is tempting to think exactly-once processing fixes this. It does not, and the reason is worth internalizing:

Flink's EXACTLY_ONCE sink atomically commits all 19 million garbage rows. Transactional atomicity of an output batch says nothing about whether the contents of that batch are a coherent view of anything. Exactly-once guarantees you won't see a row twice or lose one. It says nothing about whether any individual row was ever true. They are orthogonal properties, and every Kafka-shaped engine ships the first without the second.

The fair caveat: this is a property of the DSL, not of the engine

Brandon's article carries an update worth reading before you conclude Flink can't do this. Vasia Kalavri showed him an internally consistent implementation of the same example in Flink's DataStream API, built out of custom operators with ProcessFunction — balances only ever −1, 0 or +1, and total always 0. His own conclusion: "given that the datastream api gives you the tools to build consistent systems, I'm curious as to why the table api doesn't use them."

So Flink's checkpoint barriers can carry a consistent cut. What doesn't carry it is the declarative layer — the Table API, Flink SQL, the Kafka Streams DSL, ksqlDB. Every one of them forwards eagerly, and getting a coherent answer out of them means dropping a level and hand-compiling your topology into operators you write yourself.

That is the honest shape of the claim, and it's the one we're making: not that nobody else can be internally consistent, but that nobody else gives it to you from the DSL, on Kafka, as a flag you turn on. Our advantage over Flink here is narrower than our advantage over Kafka Streams, and it is worth being precise about which one it is.

Why StoatFlow is different

StoatFlow is a single-instance engine: one replica per application, all state global, parallelism from Project Loom virtual threads rather than from partitioned tasks. That architecture turns out to be exactly what internal consistency needs, for four independent reasons:

1. The commit barrier is a genuine consistent cut. StoatFlow processes in epochs delimited by a barrier that flows through the whole topology; when it commits, it commits a per-partition offset prefix — a Chandy–Lamport cut of the input, snapshotted on a single thread between the last dispatched record and the barrier broadcast. There is one "now" for the entire topology, not one per operator.
2. The cascade-join synchronizes across streams. When two chains fan out from a source and merge at a join, StoatFlow withholds the barrier from the join's sub-topology until both contributing chains have pushed their epoch-N output into it. The join never processes its barrier — never commits — having seen the new credit but not yet the new debit. This cross-stream synchronization is precisely the primitive the article says nobody has. We didn't build it for this; we built it for exactly-once, and it happens to be exactly what internal consistency requires.

3. Watermarks are inserted at the edge. One global watermark, the minimum across non-idle partitions, broadcast identically to every sub-topology. Brandon's fourth failure mode — per-operator watermarks that drop different data at different operators — cannot occur here by construction.
4. The cache never evicts. When StoatFlow's state cache fills, it does not flush an eldest entry and emit an intermediate (the Kafka Streams behavior that produces mid-epoch blends). It cuts the epoch short — fires an early barrier — and commits a coherent cut. Under memory pressure it commits sooner; it never leaks a garbage intermediate. This is what turns barrier-gated emission from best-effort into a guarantee.

What was missing was small by comparison: like every Kafka-shaped engine, every non-monotonic operator forwarded its output eagerly — one row per input change. So we added the gate: an opt-in emit.mode = on-barrier (and a per-operator Suppressed.untilBarrier()) that holds a KTable operator's emissions and releases a single consolidated value per consistent cut.

The measured result

We ported Brandon's benchmark verbatim — 10 accounts, $1 transfers, the credits/debits/balance/total topology above — and ran it on a 5-node cluster at 2,000 transactions/second under exactly-once, with a read_committed verifier that flags any balance outside [−1, 1] and any total ≠ 0.

Per-account balances — StoatFlow is consistent even without the valve:

Even in EAGER mode — one emission per change, the same eager forwarding Kafka Streams uses — StoatFlow emits zero impossible balances — none in 1.67 million emissions — and Σ balance never leaves [0, +5]. That residual +5 isn't an inconsistency: it's the verifier observing an epoch's ~10 per-account updates one at a time. At every committed cut, total is exactly 0. Kafka Streams, on the identical workload, emits 160 balances that cannot exist.

The two engines from the article aren't columns here, because Brandon doesn't report per-account balance statistics in a form that lines up with ours — and his dataset isn't ours, so putting his numbers in the same row would invite a comparison neither of us measured. For the record: the Flink Table API's balance histogram runs to 33, and on his simplified dataset ksqlDB's balances reach −135,051 … +2,848. Both on a quantity whose only legal values are −1, 0 and +1.

The total view — the article's headline test. Now consolidate to one total per cut with emit.mode = on-barrier:

Two of those columns are Brandon's, not ours, and they carry a gap: neither the Flink nor the ksqlDB total has a published minimum and maximum. He reports Flink's qualitatively — the error "seems to be increasing over time", oscillating "between a number of stable attractors" — and ksqlDB's not at all. What both do publish is how many values came out and how many of them were right, which is the row that matters most anyway.

Under on-barrier, every emitted total is exactly 0 — verified across both a high-churn workload and a unique-key (worst-case) workload, at ~one emission per epoch (a ~680× reduction in output volume versus EAGER, each one a true consistent-cut value). This is the property the article says no Kafka-shaped engine has: total never leaves 0, on Kafka.

Kafka Streams, meanwhile, sends the same total view to −1,619 … +1,792. Its record cache consolidates the stream — few emissions — but the emissions it does make are garbage totals.

Those numbers span four orders of magnitude and end at zero, which is hard to see in a table. On a log axis, where every gridline is a 10× step, the shape of the result is the whole argument:

Read it as an error bar: the invariant says total = 0, so a bar's length is how wrong the engine got. Two decades separate Kafka Streams from StoatFlow's eager mode — but that is a difference of degree. The step that matters is the last one, from small to none.

Magnitude is only half of the question, though, and it's the half where the engines Brandon tested can't be plotted at all. The other half — how often an engine is right in the first place — is the one axis where every engine in this story has published numbers:

That chart is the reason both halves are here. Kafka Streams is right 91.6% of the time and StoatFlow's eager mode only 15.9% — but the eager mode's errors are never larger than +5, and Kafka Streams' reach ±1,792. Being wrong rarely and being wrong slightly are different virtues, and neither is the one you want. There is exactly one row that needs no such trade-off.

A single number per engine still understates it, because it suggests a system that is steadily a bit wrong. That is not the failure mode. Here is the same measurement over time, one engine per chart — same run, same 210-second window, same invariant. These are the three engines we measured ourselves; the Flink and ksqlDB figures above are Brandon's.

Kafka Streams: right almost always, and wildly wrong the rest of the time

175 of its 191 emissions read exactly 0. The other sixteen reach −1,619 and +1,792. That is the shape of the problem: not a drift you could bound and compensate for, but a correct-looking line that is briefly, unpredictably, wildly wrong.

The record cache is what makes it look so calm. It consolidates each commit interval back to a consistent 0, so the garbage totals surface only on the mid-interval evictions — which is precisely what makes this failure mode hard to catch in production. If you sampled this dashboard once a second you would call it healthy and be wrong about one sample in twelve.

StoatFlow EAGER: wrong constantly, and never by more than +5

Read the axis before the shape: this chart is zoomed 333× against the one above, because at Kafka Streams' amplitude StoatFlow's eager mode is indistinguishable from the zero line. Its +5 would be a fortieth of a pixel up there.

On that axis, EAGER looks worse than Kafka Streams, and on one measure it is: 84.1% of its 1.67 million emissions are non-zero, against Kafka Streams' 8.4%. But the band never leaves [0, +5], and it never goes negative. That residual isn't an inconsistency in the engine — it's the verifier watching an epoch's ~10 per-account updates land one at a time. At every committed cut, total is already exactly 0. Wrong often and wrong slightly is a different failure from wrong rarely and wrong by 1,792, and neither is the one you want.

StoatFlow ON_BARRIER: exactly 0, at every emission

Same zoom, same axis, same 210 seconds as the chart above — so the empty space here is precisely the space EAGER's band fills. Hold the two side by side and that gap is the whole feature.

There is no line to trace because there is no variation to show. All 210 emissions are 0, one per commit barrier, ~1.00 s apart, each one a true consistent-cut value. Not a small error. The absence of one.

The honest part: the bug we had to fix

The first time we ran the on-barrier benchmark, it wasn't perfect. total was 0 at 406 of 412 emissions on the churn workload — and at 390 of 412 on unique keys. A small residue of tiny (±2…±4) violations survived the valve: 6 on churn, 22 on unique keys.

That residue was easy to hand-wave away as a startup transient. It wasn't. The full-topic dump showed the violations were scattered through the run, all positive (a credit reflected without its matching debit), and — the tell — the slower workload had more of them. That ruled out startup and churn and pointed at something structural.

It was: the commit barrier was a consistent cut within a sub-topology, but not across a sub-topology boundary. The total view sits two aggregation levels below the fan-out-merge join, behind a repartition. When the barrier cascaded across that boundary on a detached thread, an upstream lane could resume the instant it reported the barrier and forward a next-epoch record that overtook the barrier into the downstream cut — leaking a sliver of the following epoch into the committed transaction. Under crash recovery those leaked records would even double-count.

The fix (we call it receiver-side epoch hold-back) tags every record that crosses a sub-topology boundary with the epoch it belongs to; the receiving side briefly holds back any record that ran ahead of the barrier until the barrier arrives there too. No pauses, no coordinator changes — the cut is realigned across the boundary by construction.

We re-ran the benchmark on the fixed build. Churn: 6 → 0. Unique keys: 22 → 0. Every one of the ~412 committed total emissions is now exactly 0, on both workloads, at 2,000 tx/s. We're writing this down because "near-perfect" is not the claim — perfect, and here's the run that proves it is.

How to turn it on

Internal consistency is opt-in and per-operator, because eager emission is the right default for latency-sensitive views and matches Kafka Streams exactly. When you want a coherent view, gate it:

// Per operator:
total.suppress(Suppressed.untilBarrier())   // emit one consolidated value per consistent cut

// Or topology-wide, in application.yaml:
stoatflow:
  emit:
    mode: on-barrier

Under the hood this buffers a KTable operator's emissions per lane and releases them at the commit barrier — riding the same Kafka transaction as the offsets that produced them, so the consolidated emission is exactly-once and internally consistent. The buffer is bounded by the same early-barrier mechanism as everything else: it never grows without cutting the epoch.

Credits

This work started with Jamie Brandon's Internal consistency in streaming systems — the benchmark, the definition, and the framing are his, and the article is the clearest statement of the problem we know of. Read it.

The investigation was prompted by Ralph M. Debusmann, whose Berlin Buzzwords 2026 talk Kafi Streams – Complex Stream Processing Made Simple (recording) and the conversations that followed put the question of whether a Kafka-shaped engine could be internally consistent in front of us. The answer, it turns out, is yes — and it was mostly already there.

StoatFlow is a single-replica, Kafka Streams–compatible stream processor built on JDK 25 virtual threads. Internal-consistency mode is available now. Get in touch for early access, or head to Getting Started.