End-to-end Latency (iso-rate)
The stateless filter + enrich app of every product, fed the SAME modest rate every product keeps up with, with a probe consumer tailing the output topic: latency = record published − gendata's timestamp (p50 / p99 / p99.9). Prices each product's batching / commit / epoch model rather than its ceiling — the regime where a serving or alerting pipeline actually runs. Only meaningful in the keep-up regime.
Last updated: 2026-11-10 · same-machine, relative comparison.
flowchart LR
src[("bench-input<br/>1 partition")]
s0[\"<b>filter</b><br/>region = eu-*"/]
s1["<b>map</b><br/>latency_ratio, sla_status"]
sink[("bench-out")]
probe>"latency probe<br/>arrival − timestamp"]
src --> s0
s0 --> s1
s1 -->|"at-least-once"| sink
sink -.-> probeLatency (end-to-end — record published minus gendata's timestamp)
| product | regime | p50 ms | p90 ms | p99 ms | p99.9 ms | max ms | records | setup floor p50 / p99 ms | added p50 / p99 ms |
|---|---|---|---|---|---|---|---|---|---|
| turbine 0.3.5a1 | keep-up | 123.4 | 210.4 | 227.5 | 231.3 | 235.7 | 1433936 | 105.9 / 204.5 | 17.5 / 23.0 |
| arroyo 0.15.0 | keep-up | 153.6 | 236.4 | 266.7 | 274.1 | 277.4 | 1377567 | 105.8 / 204.6 | 47.8 / 62.1 |
| faststream 0.7.1 | keep-up | 134.9 | 219.7 | 238.5 | 244.5 | 255.4 | 1611623 | 106.1 / 204.4 | 28.8 / 34.1 |
| flink 2.2.1 | keep-up | 113.1 | 193.2 | 213.3 | 218.9 | 221.7 | 1111481 | 106.4 / 204.4 | 6.7 / 8.9 |
| quix 3.23.6 | keep-up | 122.6 | 205.0 | 223.8 | 229.4 | 232.9 | 1451282 | 105.9 / 204.8 | 16.7 / 19.0 |
| risingwave v3.1.0 | keep-up | 119.9 | 200.3 | 220.2 | 223.5 | 224.8 | 1451933 | 106.4 / 204.7 | 13.4 / 15.5 |
How to read this table
Measured by a probe consumer tailing the output topic (C client, 10 ms max fetch wait — the probe adds ≲10 ms, the same for every product). Only shown for keep-up runs: under saturation the figure is the backlog's age and grows without bound. setup floor is a harness producer writing one small record every 10 ms (acks=all, 5 ms linger) to its own topic during the same run, measured by the same probe: the broker's produce→visible latency at the products' output shape (on Redpanda, small produces surface on its flush cadence), which every product inherits. added = product − floor, the part the product is responsible for (fetch, batching, commit/epoch cadence, produce). The floor is a low-volume topic's (100 small records/s, surfacing on Redpanda's 100 ms flush.ms): an output topic carrying several MB/s — every record scored and republished — surfaces sooner: a percentile under the floor's reads below, added does not apply. Colours: each percentile against the best (lowest) of its column — green under 2×, yellow from 2×, orange from 5×, red from 10× (the ratio on hover).
Bench characteristics
In-process computation — Same as stateless (filter eu-*, derive latency_ratio + SLA flag); the output additionally carries gendata's timestamp for the probe.
| product | delivery | processing | engine | scaling on one partition | decode |
|---|---|---|---|---|---|
| turbine | at-least-once (idempotent, acks=all) | batch (Arrow columnar; 10k rows / 10 ms) | in-process · Rust core + free-threaded CPython 3.14t | intra-partition shards (parallelism=N, hash of partition_key, in-process) |
schema projection (6 of 19 fields, Arrow) |
| arroyo | at-least-once (checkpointed) | batch (vectorized SQL, Arrow / DataFusion; source batches of 10k rows / 10 ms) | in-process · Rust (arroyo run: controller + one worker process) |
none for a per-record job (chained into the subtask reading the partition) | schema projection (5 of 19 fields) |
| faststream | at-least-once (idempotent, acks=all; fire-and-forget publish) | batch I/O (getmany 10k rows / 10 ms), per-row compute | in-process · pure-Python async (aiokafka; orjson JSON codec) | none (one async consumer) | full record (19 of 19 fields, orjson dict) |
| flink | at-least-once (checkpointed) | batch (pipelined, JVM code-gen SQL) | in-process · JVM (local MiniCluster; Python only submits the job) | rebalance to N slots (parallelism.default, in-process local channels; source stays 1 subtask) |
schema projection (5 of 19 fields, Jackson); connector-default fetch |
| quix | at-least-once (idempotent, acks=all) | row (per-record StreamingDataFrame) | in-process · pure-Python over librdkafka (C client) | none (one consumer per partition, GIL) | full record (19 of 19 fields, orjson dict) |
| risingwave | at-least-once | batch (vectorized SQL) | external server · Rust (single-node: meta+compute+frontend) | hash exchange to N stream actors (SET streaming_parallelism, in-process) |
schema projection (5 of 19 fields) |
Input is a single Kafka partition (lz4). Each product may spread it over several cores through its own in-process mechanism (column scaling on one partition; what it actually used is cpu_avg); an in-process exchange is not a broker hop — the extra rec/rec column counts records that DID cross the broker on other topics. All products run at-least-once.
Consumer fetch: Turbine's core hard-codes librdkafka fetch.min.bytes=64 KB, fetch.wait.max.ms=10, max.partition.fetch.bytes=50 MB and a 500k-msg / ~2 GB prefetch queue (not adjustable), so the SAME client gets the SAME settings: quix and RisingWave (librdkafka) run with them, and so does FastStream (aiokafka, whose equivalent knobs lifted it ~14%). Flink runs at its Java connector's defaults: an A/B (2026-09-10) showed the identical numbers throttled the Java consumer (373k/s vs 543k/s at the same ~285k rows/s per core) — identical numbers are not identical behaviour on a different client. Memory at saturation is mostly this prefetch queue, for every librdkafka product.
Throughput / efficiency (live) — capped by the machine: gendata produces ~430k/s at most, so read capacity in the drain table
| product | offered/s | consumed/s | regime (lag drift) | cpu_avg % | rows/s per core | rows/s per GB | µs CPU / record | output/s | extra rec/rec | run |
|---|---|---|---|---|---|---|---|---|---|---|
| turbine 0.3.5a1 | 19,794 | 19,338 | keep-up (+2.4%) | 9.6 | 201,503 | 88,975 | 4.96 | 8,498 | 0 | 2026-11-10 |
| arroyo 0.15.0 | 19,760 | 19,174 | keep-up (+3.1%) | 7.1 | 268,789 | 40,187 | 3.72 | 7,109 | 0 | 2026-11-08 |
| faststream 0.7.1 | 19,788 | 19,366 | keep-up (+2.2%) | 33.6 | 57,604 | 233,506 | 17.36 | 9,390 | 0 | 2026-11-08 |
| flink 2.2.1 | 19,776 | 19,568 | keep-up (+1.1%) | 13.7 | 142,447 | 23,217 | 7.02 | 5,857 | 0 | 2026-11-08 |
| quix 3.23.6 | 19,770 | 19,572 | keep-up (+1.0%) | 17.7 | 110,828 | 200,229 | 9.02 | 7,553 | 0 | 2026-11-08 |
| risingwave v3.1.0 | 19,783 | 19,604 | keep-up (+0.9%) | 9.6 | 204,230 | 38,947 | 4.90 | 7,928 | 0 | 2026-11-08 |
How to read this table
consumed/s is what the product actually drained (its consumer-group offset or engine counter), offered/s what gendata produced; a dotted underline marks the median of a repeated batch — hover it for the run count and the min – max. regime (lag drift): keep-up = consumed ≈ offered with a flat backlog (a sustainable point); saturated = fed beyond capacity, so consumed/s IS the ceiling and latency is unbounded; draining = still eating warmup backlog (consumed > offered; CPU is catch-up, not steady state — lengthen the warmup, or raise the rate if the ceiling is what you want). The lag drift in parentheses = backlog change over the window ÷ records consumed — the evidence for the regime (≈ 0 at keep-up, positive when saturated, negative while draining). cpu_avg is the whole process tree, 100 = one core busy. rows/s per core = consumed ÷ (cpu_avg/100); rows/s per GB = consumed ÷ memory; extra rec/rec = records written to topics other than the input and output per record consumed (0 = no broker hop).
Resources (raw)
| product | cpu_avg % | cpu_p95 % | mem_avg MB | mem_max MB | samples | instrument |
|---|---|---|---|---|---|---|
| turbine 0.3.5a1 | 9.6 | 10.8 | 222.6 | 225.5 | 60 | /proc Δ, pss |
| arroyo 0.15.0 | 7.1 | 8.9 | 488.6 | 500.8 | 60 | /proc Δ, pss |
| faststream 0.7.1 | 33.6 | 35.5 | 84.9 | 86.7 | 60 | /proc Δ, pss |
| flink 2.2.1 | 13.7 | 22.6 | 863.0 | 896.9 | 59 | /proc Δ, pss |
| quix 3.23.6 | 17.7 | 18.7 | 100.1 | 100.1 | 60 | /proc Δ, pss |
| risingwave v3.1.0 | 9.6 | 10.8 | 515.4 | 520.8 | 59 | /proc Δ, rss |
How to read this table
CPU is summed over the whole process tree (JVM, UDF servers, the RisingWave container's processes) as utime+stime deltas per 1 s sample — 100 = one core. Memory is PSS (shared pages counted once) where readable, RSS otherwise (instrument). Rows marked ⚠ legacy ps were measured with the old ps %CPU lifetime average and are not comparable: re-run them.
Product notes
turbine — Batch formation is size-or-timeout; at this rate every batch closes on the timeout (BENCH_BATCH_TIMEOUT_MS, every batching product alike, 500 ms on the throughput workloads) — its cost in throughput is what the stateless page measures at 500.
arroyo — Arroyo's source flushes a batch on the bench's level batch size or latency timeout (source-batch-size / source-batch-linger; its defaults are 512 / 100 ms): at this rate every batch closes on the linger, the floor of its latency, by design.
flink — Flink's Kafka sink writes as records arrive (no wait for a checkpoint under at-least-once); its network buffers flush on the bench's latency timeout (execution.buffer-timeout.interval; its default is 100 ms).
risingwave — Its at-least-once Kafka sink produces rows as their chunks flow through it: the 1 s barrier (barrier_interval_ms) checkpoints, it does not hold the output back (max ~225 ms on the c3, well under one epoch).
Sanity — ⚠ off theory
Are we benching the same thing?
| product | selectivity (out/consumed) | expected | ✓ | breach_rate | enrichment | sample |
|---|---|---|---|---|---|---|
| turbine 0.3.5a1 | 0.4222 | 0.3070 | ⚠ | 0.224 | ✓ | 500 |
| arroyo 0.15.0 | 0.3709 | 0.3415 | ✓ | 0.192 | ✓ | 500 |
| faststream 0.7.1 | 0.4849 | 0.4675 | ✓ | 0.182 | ✓ | 500 |
| flink 2.2.1 | 0.2994 | 0.2995 | ✓ | 0.178 | ✓ | 500 |
| quix 3.23.6 | 0.3861 | 0.3710 | ✓ | 0.192 | ✓ | 500 |
| risingwave v3.1.0 | 0.4041 | 0.4045 | ✓ | 0.190 | ✓ | 500 |
Verdict: ⚠️ off theory — selectivity more than 25% from the expected value for: turbine (0.422 vs 0.307). Either the filter differs or the consumed reading is off.
Turbine releases
| line | version | bench | regime | offered/s | consumed/s | rows/s per core | µs CPU / record | rows/s per GB | p99 ms | Δ efficiency | Δ capacity | run |
|---|---|---|---|---|---|---|---|---|---|---|---|---|
| turbine | 0.3.5a1 | 2 | keep-up | 19,794 | 19,338 | 201,503 | 4.96 | 88,975 | 227.5 | +5.7% | - | 2026-11-10 |
| turbine | 0.3.4a1 | 2 | keep-up | 19,783 | 19,295 | 190,687 | 5.24 | 81,693 | 229.1 | ≠ bench | ≠ bench | 2026-11-10 |
| turbine | 0.3.3a1 | 1 ⚠ deprecated | keep-up | 19,772 | 19,315 | 216,153 | 4.63 | 82,520 | 253.9 | -1.1% | - | 2026-11-06 |
| turbine | 0.3.2a1 | 1 ⚠ deprecated | keep-up | 19,797 | 19,288 | 218,502 | 4.58 | 76,829 | 247.5 | +10.6% | - | 2026-11-04 |
| turbine | 0.3.1a2 | 1 ⚠ deprecated | keep-up | 19,777 | 19,326 | 197,640 | 5.06 | 75,865 | 247.6 | -8.2% ⚠ | - | 2026-11-01 |
| turbine | 0.3.1a1 | 1 ⚠ deprecated | keep-up | 19,792 | 19,615 | 215,222 | 4.65 | 77,299 | 248.3 | +2.3% | - | 2026-10-28 |
| turbine | 0.2.2a2 | 1 ⚠ deprecated | keep-up | 19,773 | 19,322 | 210,387 | 4.75 | 86,374 | 247.2 | -5.1% ⚠ | - | 2026-10-20 |
| turbine | 0.2.2a1 | 1 ⚠ deprecated | keep-up | 19,780 | 19,329 | 221,588 | 4.51 | 86,201 | 255.1 | -0.0% | - | 2026-10-18 |
| turbine | 0.2.1a4 | 1 ⚠ deprecated | keep-up | 19,777 | 19,345 | 221,666 | 4.51 | 73,678 | 255.3 | -8.1% ⚠ | - | 2026-10-18 |
| turbine | 0.2.1a3 | 1 ⚠ deprecated | keep-up | 19,776 | 19,350 | 241,129 | 4.15 | 70,963 | 244.4 | +7.8% | - | 2026-10-18 |
| turbine | 0.2.1a2 | 1 ⚠ deprecated | keep-up | 19,788 | 19,362 | 223,729 | 4.47 | 76,838 | 261.0 | +16.1% | - | 2026-10-18 |
| turbine | 0.2.0a2 | 1 ⚠ deprecated | keep-up | 19,771 | 19,348 | 192,671 | 5.19 | 77,474 | 245.8 | - | - | 2026-10-18 |
How to read this table
The Turbine lines measured on several releases, newest first: a regression check, not part of the cross-product comparison above (which shows the pinned release only). Δ compares each release with the one below it: efficiency = rows/s per core (the inverse of µs CPU per record), capacity = consumed/s, only when both runs were fed beyond capacity (saturated or drain — at a keep-up point consumed/s is just the offered rate). ⚠ = worse by more than 5%, about the single-run noise: confirm with --repeat 3. Two releases measured in two sessions also differ by the host-to-host spread — same flavor, another physical host: −0.7 % consumed / −3.5 % rows/s per core in the c3 control of 2026-10-07 —, which leaves little headroom under the 5 %: re-measure both releases in ONE session (suite.py run … --version X) before acting on a ⚠. No delta is computed across a change of bench version (bench, from bench.yaml), host or run config; ⚠ deprecated rows were measured on an older bench definition — re-run that release with suite.py run turbine WORKLOAD --version X.
Host: ovh:c3-32 (ovh:c3-32@SBG5) · AMD EPYC-Milan Processor (2400 MHz) · 16 cores / 16 threads · 31345 MB · kernel 6.12.100+deb13-cloud-amd64.
Redpanda: rpk version: v26.2.2. Redpanda, gendata and the product under test share this host.