Stateless Filter + Enrich
Per-record filter keeping European regions (eu-*), then enrich each kept row with a derived latency_ratio and an ok/breach SLA decision. No state — measures the raw I/O + serde + per-record compute ceiling (fed above each consumer's capacity so CPU saturates).
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")]
src --> s0
s0 --> s1
s1 -->|"at-least-once"| sinkBench characteristics
In-process computation — Per record: filter region LIKE 'eu-%', then derive latency_ratio = latency_ms / 80 and an ok/breach SLA flag. Stateless.
| product | delivery | processing | engine | scaling on one partition | decode |
|---|---|---|---|---|---|
| turbine | at-least-once (idempotent, acks=all) | batch (Arrow columnar; 100k rows / 500 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 100k rows / 500 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 100k rows / 500 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.
Drain ceiling (--prefill: static backlog, no producer running)
| product | backlog | drained/s | cpu_avg % | rows/s per core | rows/s per GB | µs CPU / record | output/s | run |
|---|---|---|---|---|---|---|---|---|
| turbine 0.3.5a1 | 120M | 868,640 | 281.3 | 308,762 | 629,067 | 3.24 | 273,407 | 2026-11-10 |
| arroyo 0.15.0 | 120M | 558,567 | 141.3 | 395,352 | 366,046 | 2.53 | 183,220 | 2026-10-27 |
| faststream 0.7.1 | 120M | 75,223 | 99.8 | 75,367 | 138,897 | 13.27 | 21,943 | 2026-10-27 |
| flink 2.2.1 | 120M | 549,936 | 223.0 | 246,641 | 382,122 | 4.05 | 171,977 | 2026-10-20 |
| quix 3.23.6 | 120M | 123,779 | 105.8 | 116,965 | 172,750 | 8.55 | 39,922 | 2026-10-20 |
| risingwave v3.1.0 | 120M | 355,484 | 126.4 | 281,168 | 404,187 | 3.56 | 111,089 | 2026-10-20 |
How to read this table
gendata wrote the backlog and exited BEFORE the consumer started, so the window measures a drain with no producer competing for the box — the product's consume ceiling free of gendata's CPU and 10 Hz burstiness. Not comparable with a live table (different regime); ⚠ exhausted = the backlog ran out mid-window, rate understated.
Resources (raw)
| product | cpu_avg % | cpu_p95 % | mem_avg MB | mem_max MB | samples | instrument |
|---|---|---|---|---|---|---|
| turbine 0.3.5a1 | 281.3 | 292.0 | 1,414.0 | 1,519.5 | 59 | /proc Δ, pss |
| arroyo 0.15.0 | 141.3 | 163.0 | 1,562.6 | 1,772.8 | 59 | /proc Δ, pss |
| faststream 0.7.1 | 99.8 | 101.1 | 554.6 | 655.7 | 59 | /proc Δ, pss |
| flink 2.2.1 | 223.0 | 242.7 | 1,473.7 | 1,509.4 | 59 | /proc Δ, pss |
| quix 3.23.6 | 105.8 | 111.7 | 733.7 | 740.0 | 59 | /proc Δ, pss |
| risingwave v3.1.0 | 126.4 | 137.4 | 900.6 | 944.2 | 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
arroyo — arroyo run executes the SQL pipeline locally (the release binary; Python only writes the SQL): its controller and one worker process, both in the sampled process tree, the worker running parallelism = the run's cores (3). A per-record job is chained into the one source subtask that reads the partition (no exchange, as in Flink's stateless SQL); librdkafka's consumer and producer threads take the rest. The Kafka source commits its offsets to the group at each checkpoint barrier (10 s, Arroyo's default) — the offset of the last record read, which the harness counts + 1 — so consumed/s is a real reading.
faststream — Pure-Python async (aiokafka): CPU-bound on one core, at about half of quix's per-core throughput (quix's librdkafka is a C extension). JSON is decoded with orjson — FastStream auto-detects it once it is installed — which MATCHES quix, whose default JSON deserializer is also orjson; on stdlib json its ceiling was ~22 % lower, so orjson both levels the comparison and lifts it. Benched with FastStream's batch=True subscriber (getmany) + fire-and-forget publish (no_confirm=True, so aiokafka batches the sends): the default per-record path awaits the broker ack on every message and is round-trip-bound (~8k/s, CPU idle), not how FastStream is run at throughput. FastStream has no built-in windowing, so it has no windowed bench: one would measure hand-rolled timer / state code, not FastStream.
flink — PyFlink submits a pure Table API / SQL job to a LOCAL MiniCluster whose parallelism is the run's cores — the source stays one subtask on the single Kafka partition and rebalances the records over the slots in-process, the bench's KPI (3 since 2026-10-20, 1 before). CPU/RSS are summed across the process tree INCLUDING the JVM child, so the figure carries JVM overhead (GC, checkpoints, network / produce threads). At-least-once checkpointing commits the consumed offset to the group each checkpoint, so consumed/s is a real reading like quix's. Benched version from flink/uv.lock; the Kafka SQL connector JAR is fetched from Maven on first run.
risingwave — Single-partition bound: RisingWave runs one source reader per Kafka partition, so on the bench's single partition its CPU stays far below the box's cores (sink coupled). It cannot trade its idle cores for more throughput; scaling needs more Kafka partitions (≈3× at 4), not more cores. consumed/s is read from RisingWave's source state (the Kafka offset its source has read; it commits no Kafka group offset); the sink is coupled to the dataflow (sink_decouple = false), so that offset is processed and produced, not merely buffered; CPU/RSS are the whole single-node container, re-created before each run and stopped after it.
Sanity — ✓ matches theory
Are we benching the same thing?
| product | selectivity (out/consumed) | expected | ✓ | breach_rate | enrichment | sample |
|---|---|---|---|---|---|---|
| turbine 0.3.5a1 | 0.3148 | 0.3055 | ✓ | 0.202 | ✓ | 500 |
| arroyo 0.15.0 | 0.3180 | 0.3250 | ✓ | 0.186 | ✓ | 500 |
| faststream 0.7.1 | 0.2917 | 0.3105 | ✓ | 0.180 | ✓ | 500 |
| flink 2.2.1 | 0.3118 | 0.3015 | ✓ | 0.200 | ✓ | 500 |
| quix 3.23.6 | 0.3216 | 0.3375 | ✓ | 0.198 | ✓ | 500 |
| risingwave v3.1.0 | 0.3125 | 0.3105 | ✓ | 0.178 | ✓ | 500 |
Verdict: ✓ matches theory — every row's selectivity is within 25% of the expected value (output ÷ consumed).
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 · drain | 0.3.5a1 | 1 | drain | — | 868,640 | 308,762 | 3.24 | 629,067 | - | +1.0% | +1.2% | 2026-11-10 |
| turbine · drain | 0.3.4a1 | 1 | drain | — | 858,095 | 305,769 | 3.27 | 606,962 | - | -4.7% | -6.5% ⚠ | 2026-11-10 |
| turbine · drain | 0.3.3a1 | 1 | drain | — | 917,847 | 320,840 | 3.12 | 655,815 | - | -1.5% | +0.3% | 2026-11-06 |
| turbine · drain | 0.3.2a1 | 1 | drain | — | 915,217 | 325,783 | 3.07 | 627,097 | - | +1.3% | -0.5% | 2026-11-04 |
| turbine · drain | 0.3.1a2 | 1 | drain | — | 920,019 | 321,570 | 3.11 | 640,596 | - | +2.4% | +0.6% | 2026-11-01 |
| turbine · drain | 0.3.1a1 | 1 | drain | — | 914,509 | 313,976 | 3.18 | 637,396 | - | +0.7% | +0.8% | 2026-10-28 |
| turbine · drain | 0.2.2a2 | 1 | drain | — | 907,031 | 311,682 | 3.21 | 640,773 | - | -2.5% | -2.7% | 2026-10-20 |
| turbine · drain | 0.2.2a1 | 1 | drain | — | 932,497 | 319,814 | 3.13 | 656,165 | - | -2.1% | +55.9% | 2026-10-19 |
| turbine · drain | 0.2.1a4 | 1 | drain | — | 598,255 | 326,625 | 3.06 | 417,118 | - | +0.9% | +2.5% | 2026-10-19 |
| turbine · drain | 0.2.1a3 | 1 | drain | — | 583,688 | 323,552 | 3.09 | 401,651 | - | +0.2% | -1.8% | 2026-10-19 |
| turbine · drain | 0.2.1a2 | 1 | drain | — | 594,525 | 322,946 | 3.10 | 425,849 | - | -17.8% ⚠ | +32.6% | 2026-10-19 |
| turbine · drain | 0.2.0a2 | 1 | drain | — | 448,465 | 392,949 | 2.54 | 366,332 | - | - | - | 2026-10-19 |
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@DE1, ovh:c3-32@SBG5, ovh:c3-32@UK1) · 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.