Stream Processing
Deriving new streams from streams — stateful, changelog-backed windowed aggregation, and KEDA scaling workers on consumer lag rather than CPU, down to zero when the topic is quiet.
A consumer reacts to one event at a time. A stream processor does something more: it treats the stream as input to a standing computation and emits a new stream. The previous chapter established that the log is a replayable source of truth; this chapter builds standing computations on top of it.
Derive new streams from streams
A stream-processing app is a topology: sources (input topics), stateful processors (filter, join, window, aggregate), and sinks (output topics or stores). The output of one topology is just another stream that other services consume.
Reading the topology left to right makes the pattern concrete: two input topics
(orders and payments) are joined and bucketed into five-minute tumbling windows
keyed by order id; the windowed result is aggregated into per-merchant counts, sums,
and rates; and that aggregate is published as a brand-new revenue topic that a
dashboard — or any other service — can consume. Nothing here calls another service
synchronously; each stage only reads a stream and writes a stream.
The word that matters is stateful. Unlike a simple consumer, these processors keep local state — running totals, window buffers — and that state must survive restarts. The fault-tolerance trick is a changelog: the local state store is backed by a compacted Kafka topic, so if the pod dies, the state rebuilds by replaying the changelog. That is fault-tolerant local state with no external database in the hot path.
A windowed aggregation
Here is the canonical example — revenue per merchant, in tumbling five-minute windows — in each ecosystem. The JVM stacks use Kafka Streams; .NET uses Streamiz (a native Kafka Streams port); Python uses Faust; C++ keeps bounded state in-process and reaches for Flink only when the state outgrows a single worker.
// spring-kafka: the topology is built via a StreamsBuilder bean
@Configuration
@EnableKafkaStreams
public class RevenueTopology {
@Bean
public KStream<String, Order> revenueStream(StreamsBuilder b) {
KStream<String, Order> s = b.stream("order.placed",
Consumed.with(Serdes.String(), orderSerde));
s.groupBy((k, o) -> o.merchantId())
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
.aggregate(() -> 0.0, // local, changelog-backed
(mId, o, total) -> total + o.total(),
Materialized.with(Serdes.String(), Serdes.Double()))
.toStream()
.to("revenue.by-merchant"); // derived stream out
return s;
}
}
// quarkus-kafka-streams: the topology is a CDI-produced bean
@ApplicationScoped
public class RevenueTopology {
@Produces
public Topology build() {
StreamsBuilder b = new StreamsBuilder();
b.stream("order.placed", Consumed.with(Serdes.String(), orderSerde))
.groupBy((k, o) -> o.merchantId())
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
.aggregate(() -> 0.0, // local, changelog-backed
(mId, o, total) -> total + o.total(),
Materialized.as("revenue-by-merchant"))
.toStream()
.to("revenue.by-merchant");
return b.build();
}
}
// Streamiz.Kafka.Net (MIT) — a .NET port of Kafka Streams: same
// topology / state-store / changelog model, native C#, no JVM in the container.
var config = new StreamConfig<StringSerDes, StringSerDes>
{
ApplicationId = "revenue-aggregator",
BootstrapServers = "kafka:9092",
StateDir = "/var/streamiz/state", // local RocksDB, changelog-backed
};
var builder = new StreamBuilder();
builder.Stream<string, Order, StringSerDes, JsonSerDes<Order>>("order.placed")
.GroupBy((k, o, _) => o.MerchantId)
.WindowedBy(TumblingWindowOptions.Of(TimeSpan.FromMinutes(5), TimeSpan.Zero)) // no grace
.Aggregate(() => 0.0, // local, changelog-backed
(mId, o, total) => total + o.Total,
RocksDbWindows.As<string, double>("revenue-by-merchant")
.WithKeySerdes(new StringSerDes())
.WithValueSerdes(new DoubleSerDes()))
.ToStream()
.Map((w, t, _) => KeyValuePair.Create(w.Key, t)) // unwrap Windowed<string> to the merchant id
.To<StringSerDes, DoubleSerDes>("revenue.by-merchant"); // derived stream out
var stream = new KafkaStream(builder.Build(), config);
await stream.StartAsync();
import faust
from datetime import timedelta
app = faust.App("stream-processor", broker="kafka://my-cluster-kafka-bootstrap",
store="memory://") # state is rebuilt from the changelog topic
orders_topic = app.topic("order.placed", value_type=Order)
revenue_topic = app.topic("revenue.by-merchant", value_type=Revenue)
# table = local, changelog-backed state (a materialised view): tumbling windows,
# no grace (expires after one window), placed by the record's own timestamp
revenue = app.Table("revenue", default=float) \
.tumbling(300, expires=timedelta(seconds=300)) \
.relative_to_stream()
@app.agent(orders_topic)
async def aggregate(stream):
async for order in stream.group_by(Order.merchant_id): # repartition by merchant
revenue[order.merchant_id] += order.total # stateful, per key and window
start = int(faust.current_event().message.timestamp // 300) * 300
await revenue_topic.send(key=order.merchant_id, value=Revenue(
window_start=start, window_end=start + 300, merchant_id=order.merchant_id,
revenue=revenue[order.merchant_id].value())) # derived stream out
// In-process windowed aggregation: revenue-per-merchant, 5-minute tumbling.
// Stateful but bounded — fits in-process. For joins or large state: use Flink.
struct Window {
std::chrono::sys_seconds start;
std::unordered_map<std::string, double> revenue; // per-merchant
};
Window current{floor_to_5min(now()), {}};
consumer.subscribe({"order.placed"});
while (!stop_token.stop_requested()) {
for (auto& msg : consumer.poll(100ms)) {
Order o = deserialize(msg.value());
auto win_start = floor_to_5min(o.placed_at);
if (win_start != current.start) { // window rolled over
emit_window(current); // flush to revenue.by-merchant
current = Window{win_start, {}};
}
current.revenue[o.merchant_id] += o.total; // stateful, in-process
}
}
// goka: a group (the Kafka consumer group) with a changelog-backed table. Stage one re-keys
// each order by merchant; stage two folds it into the merchant's tumbling windows and emits
// every update to revenue.by-merchant. Window state lives in the group table, so a restart
// replays the changelog topic and picks up where it stopped.
graph := goka.DefineGroup("stream-processor",
goka.Input("order.placed", new(codec.Bytes), rekey),
goka.Output("stream-processor-by-merchant", rekeyedCodec{}),
goka.Input("stream-processor-by-merchant", rekeyedCodec{}, aggregate),
goka.Output("revenue.by-merchant", new(codec.Bytes)),
goka.Persist(stateCodec{}), // the group table
)
func aggregate(ctx goka.Context, msg any) {
in := msg.(*rekeyed) // {ts, total} for one order
state := loadState(ctx) // this merchant's windows, from the table
start := (in.TS / 1000 / windowSeconds) * windowSeconds // 5-minute floor in the book, WINDOW_SECONDS in the example
cur := state.add(start, in.Total) // count + running total for the window
ctx.SetValue(state) // goka writes the changelog
ctx.Emit(revenue, ctx.Key(), revenueJSON(ctx.Key(), start, cur)) // revenue.by-merchant
}
How the code works
Every tab is the same topology: read order.placed, group by merchant, bucket into
five-minute windows, fold each window into a running total, and write the result to
a derived revenue.by-merchant stream. The JVM, .NET, Python and Go versions (Kafka Streams, Streamiz, Faust, goka) get the
changelog-backed store for free — kill the pod and the window state rebuilds by
replaying the changelog topic. The C++ version keeps the window map in process,
which is fine while the state is bounded; once you need joins or state larger than a
worker, that is the signal to move to a dedicated engine like Flink rather than
hand-roll it.
Delivery guarantees in stream processing
The windowed aggregation has a silent assumption: what happens when the worker crashes mid-window, or when a message is delivered twice? Stream processing has three delivery guarantee levels, and picking the right one is a design decision:
At-most-once — process each message zero or one times. Commit the offset before processing. If the worker crashes after committing but before processing, the message is lost. Fast but lossy; acceptable for metrics that tolerate sampling.
At-least-once — process each message one or more times. Commit the offset after processing. If the worker crashes after processing but before committing, the message is redelivered and processed again. This is the default in most Kafka consumers and the approach this book prefers — paired with idempotent handlers (the same discipline from the event-driven and workflows chapters), reprocessing is a no-op and the combination is effectively once.
Exactly-once — process each message exactly once, even across crashes. Kafka’s
processing.guarantee=exactly_once_v2 wraps the consume-transform-produce cycle in
a transaction: the offset commit and the output record are written atomically, so a
crash either rolls back both or commits both. The cost is higher latency (transaction
coordination) and the requirement that both input and output live on the same Kafka
cluster.
# Kafka Streams config for exactly-once (JVM stacks)
processing.guarantee: exactly_once_v2 # atomic consume-transform-produce
# Requires: Kafka broker ≥ 2.5, transactional.id auto-assigned per task
For most services in this stack, at-least-once plus idempotency is the right default: it is simpler, works across cluster boundaries, and the idempotency discipline is already required by Jobs, CronJobs, and queue workers anyway. Exactly-once is worth the cost when the output of the stream processor is a derived fact (like the revenue aggregation) that other services treat as authoritative and where double-counting would be incorrect rather than merely redundant.
Beyond tumbling windows
The revenue example uses a tumbling window — fixed-size, non-overlapping buckets. Kafka Streams, Faust, and Streamiz support three other window types, each suited to a different access pattern:
Tumbling windows — fixed-size, non-overlapping. Every event lands in exactly one window. The revenue-per-merchant aggregation uses this: a clean five-minute bucket, no overlap.
Hopping (sliding) windows — fixed-size, but the window advances by a smaller interval. A five-minute window that advances every one minute produces overlapping windows, giving a smoothed view: each output reflects the last five minutes of data, updated every minute. Good for dashboards where a hard bucket boundary would cause a visible cliff.
Session windows — variable-length, defined by an inactivity gap. No events for the gap duration (say, 10 minutes) closes the window. This fits user-activity patterns: a user’s browsing session is a cluster of events with quiet gaps between them, and the session window captures that shape without you picking an arbitrary bucket size.
Late-arriving data — an event whose timestamp falls inside an already-closed window. Kafka Streams handles this with a grace period: windows stay open for an additional duration after their nominal close, accepting late events. Events that arrive after the grace period are dropped (or routed to a dead-letter topic). Flink extends this with watermarks — a system-wide notion of “how far behind event-time we expect to be” — which is essential for multi-source joins where sources have different latencies.
When to reach for Flink — in-process stream processing (Kafka Streams, Faust, Streamiz, goka, the C++ loop in this chapter’s example) works well for single-topic, bounded-state computations. Once you need stream-to-stream joins (correlate orders with payments by key and time), state larger than a single worker’s memory, or exactly-once across multiple Kafka clusters, a dedicated engine like Apache Flink is the right tool. Flink manages state snapshots, checkpointing, and watermarks at a scale that in-process libraries don’t attempt.
Scale on the signal that matters
A stream processor’s load is backlog, not CPU. The right scaling signal is consumer lag — how far behind the log the workers are. KEDA reads that lag and scales workers up, and crucially down to zero when there is no lag. For synchronous HTTP services the same KEDA closes the loop on request rate instead. One mechanism, two signals.
# event-driven: scale consumers on Kafka lag
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
spec:
scaleTargetRef: { name: payment-consumer }
minReplicaCount: 0 # scale to zero when there is no lag
maxReplicaCount: 20
triggers:
- type: kafka
metadata: { topic: order.placed, consumerGroup: payment, lagThreshold: "100" }
---
# synchronous: scale the HTTP service on concurrent requests (KEDA http-add-on)
kind: HTTPScaledObject
spec: { scaleTargetRef: { service: order-service }, replicas: { min: 0, max: 30 } }
The lagThreshold: "100" is the tuning knob — how many messages of backlog a
single worker is allowed before KEDA adds another. minReplicaCount: 0 is the cost
lever: true scale-to-zero when the topic is quiet, springing back the moment lag
builds. Picking the signal — lag for events, request rate for HTTP — is the whole
point; scaling a backlog-bound consumer on CPU would miss the load entirely.
Cross-check it yourself
Make lag visible, then make it move. Pause the payment-consumer, publish a burst
of order.placed with hey driving order-service, and watch the consumer-group
lag climb. Resume it and watch KEDA add replicas until the lag drains, then settle
back toward zero. The replica count tracking the lag curve — not CPU — is the
behaviour to confirm.
The code is in examples/06-stream-processing/. The run script there builds and
runs it; its README.md covers what it does and how to drive it.
Verification status: verified — examples/06-stream-processing/ passes 11/11 checks in each of the six languages on Docker Engine (2026-10-10)
(per-merchant revenue aggregation onto a derived Kafka topic, the stream-processor consumer group, consumer lag that builds while the processor is paused and drains on resume, and the burst appearing in the revenue stream).