پردازش جریانیStream Processing

When the input is a file (a sequence of bytes), the first step is usually parsing it into a sequence of records. In the streaming world a record is called an event, but it is practically the same thing: a small, self-contained, immutable entity detailing something that happened at a point in time — a user click, a purchase, a sensor temperature, a web-server log line. An event usually carries a timestamp too (wall-clock time; how it differs from a monotonic clock was covered in Chapter 9).

An event can be encoded (text, JSON, binary — Chapter 5), stored (file, table, document), and sent over the network. In batch, a file is written once and perhaps read by several jobs; in streaming, an event is produced once by a producer (publisher / sender) and perhaps processed by several consumers (subscribers / recipients). Same-kind events gather in a topic (stream) — the equivalent of a filename in a filesystem.

On the surface, a file or a database is enough to connect producer and consumer: the producer writes, the consumer polls periodically — exactly what a daily batch does. But as you move toward low latency, polling gets expensive: the more often you poll, the more requests return empty-handed. It is better for the consumer to be notified that a fresh event arrived. Relational databases have triggers, but they are limited and marginal; that is why specialized event-notification tools were built.

The simplest messenger is a direct channel like a Unix pipe or a TCP connection between producer and consumer — but those connect exactly one sender to one receiver, while a messenger must connect many producers to one topic and one topic to many consumers.

• Direct messaging (no middleman): UDP multicast in stock markets (low latency, app-level lost-packet recovery), brokerless libraries like ZeroMQ and nanomsg, metric agents like StatsD over UDP (counters are only right if every message arrives), and webhooks (the producer pushes to the consumer over HTTP/RPC). Their shared assumption: both sides are permanently online; an offline consumer loses messages, and if the producer itself crashes, its retry buffer dies with it.
• Message broker (queue): a server like a database, optimized for message streams. The producer writes to the broker; the broker delivers to the consumer. With data centralized in the broker, client churn and crashes are easily tolerated, and the durability question moves to the broker (in-memory only for some, on disk by configuration for others). Consumers are asynchronous: the producer only waits for the broker's buffering ack, not for final processing.

Brokers resemble databases (some even join two-phase commit via XA/JTA) but differ in practice: databases keep data until explicitly deleted, while some brokers auto-delete delivered messages (so they don't fit long-term storage); they assume short queues (deep buffering kills throughput); instead of a query language they offer topic-pattern subscriptions; and instead of point-in-time snapshot results they emit change notifications (a fresh message arrived — something databases lack). This is the traditional JMS/AMQP view (RabbitMQ, ActiveMQ and kin).

When several consumers read one topic, there are two patterns (Fig. 12-1):

• Load balancing: each message reaches one of the consumers; the broker assigns at will. For when processing each message is expensive and you want to parallelize by adding consumers. (In AMQP with several clients on one queue; in JMS with a shared subscription.)
• Fan-out: each message reaches all consumers; each listens to the same broadcast independently — the streaming equivalent of several batch jobs reading one input file. (In JMS with a topic subscription; in AMQP with an exchange binding.)

The two compose: with Kafka consumer groups, each message reaches one consumer within each group (load balancing inside the group) and all groups (fan-out across groups) — exactly what the bottom half of Diagram 1 shows.

If the producer sends faster than the consumer processes, there are three options: dropping messages, buffering them in a queue, or backpressure / flow control — stalling the producer until there is room. Unix pipes and TCP take the third path with a small fixed buffer: when full, the sender blocks until the receiver takes something. If messages are buffered, you must work out what happens as the queue grows: it won't fit in memory? It spills to disk? What does disk do to performance? And when the disk fills, then what?

Second question: if a node crashes or goes briefly offline, is any message lost? Like a database, durability is disk writes plus replication — and it costs. If you can occasionally lose messages, the same hardware buys higher throughput and lower latency. Whether loss is acceptable depends on the use case: for periodic sensor readings one lost dot hardly matters (a fresh value arrives minutes later — though beware that if many messages drop, metric inaccuracy goes unnoticed); but if you are counting events, every lost message means a wrong counter, so reliable delivery is required.

A consumer can crash at any moment: the broker delivered the message but it was never processed, or only half-processed. The fix is acknowledgment: the client must explicitly confirm a message is done so the broker removes it from the queue. If the connection closes without an ack or times out, the broker assumes the message was never processed and redelivers it to another consumer. (Subtlety: the message may be fully processed with only the ack lost on the network — handling that needs an atomic-commit protocol, unless the operation is idempotent or exactly-once isn't required.)

Redelivery doesn't just reorder; it can waste resources or wedge the flow forever. The classic scenario: a producer that serializes badly — say it drops a required key from a JSON object. The keyless message crashes the consumer, which restarts; no ack arrives, the broker resends, the next consumer dies too — and the loop runs forever. With a hard ordering guarantee no progress is possible at all; with relaxed ordering you progress but burn resources on messages that never get acked.

The fix is a dead letter queue (DLQ): instead of retrying forever, the message moves to a separate queue so consumers stay free. DLQs are normally monitored — each message in one is an error, and the operator decides: drop it forever, fix and reproduce it manually, or fix the consumer code. DLQs are common in most queueing systems, and log-based and stream-processing systems support them now too.

Sending a packet over the network or a request to a service is usually a transient act that leaves no permanent trace — and lightweight AMQP/JMS brokers inherited that transient mindset: even when they write a message to disk, they delete it soon after delivery. Databases and filesystems are the opposite: whatever is written is, as a rule, recorded forever until someone explicitly deletes it. That mindset difference hugely affects “derived data”: in batch you can repeat processing many times without harming the input, but in traditional messaging receiving a message with ack is destructive (the message leaves the broker) — so you can't rerun the same consumer and get the same result. And if you add a fresh consumer, it only gets messages after its registration; earlier messages are gone — unlike files and databases, where a late-arriving client still reads the past.

Why not have both: database-like durable storage plus messenger-like low-latency notification? That is the log-based broker idea. A log is just an append-only sequence of records on disk (the same one we saw in Chapter 4 storage engines, Chapter 6 replication and Chapter 10 consensus). The producer sends by appending to the log tail; the consumer receives by reading sequentially; and on reaching the tail it waits for fresh-message notification — exactly like tail -f in Unix.

For throughput beyond one disk, the log is sharded (Chapter 7): each shard is an independent log on its own machine, and a topic is a group of shards carrying same-kind messages (Fig. 12-3 — the bottom half of Diagram 1). Within each shard, which Kafka calls a partition, the broker assigns every message an ascending sequence number: the offset. Because a partition is append-only, messages inside one partition are fully ordered; but there is no ordering guarantee across partitions. Apache Kafka and Amazon Kinesis work this way (reaching millions of messages per second with fault tolerance through multi-machine sharding and replication).

• Trivial fan-out: several independent consumers read the log without affecting each other; reading never deletes from the log. For load balancing, instead of assigning individual messages the broker hands whole shards to consumer-group nodes, and each client reads its own shard single-threaded and in order.
• Parallelism ceiling = shard count: since one shard's messages reach one node, the cooperating nodes of a topic are at most its shard count. (One shard's work can be split, but that complicates offset management; one thread per shard is preferred, and you parallelize with more shards.)
• Head-of-line blocking: if one message is slow to process, the later messages of the same shard wait behind it.

So the rule of thumb: if messages are expensive and you want to parallelize them one by one and ordering matters little, a lightweight JMS/AMQP broker is better; if throughput is high, each message processes fast and ordering matters, the log-based approach shines — though the boundary between the two architectures is blurring (Kafka now supports lightweight JMS/AMQP-style consumer groups too). And since order survives only within a shard, every event that must stay ordered must go to the same shard — e.g. by partitioning on the user ID key.

Sequential shard consumption brings a gift: the broker needn't track per-message acks; periodically recording the consumer offset is enough. Every message below the offset is processed; larger ones unseen. That lighter bookkeeping plus batching and pipelining is what lifts log-based throughput. The price: if the consumer dies it resumes from the last recorded offset, not the last seen one — so it sees some messages twice. That offset is exactly the log sequence number of single-leader database replication (Chapter 6): the broker is like the leader database, the consumer like a follower.

In the three-way classification (drop / buffer / backpressure), the log-based approach is buffering with a large fixed buffer (bounded by disk space). If a consumer lags so far that its offset points at a deleted segment, those messages are lost — i.e. the broker effectively drops messages older than the buffer. You can monitor a consumer's lag behind the log tip and alert when it grows; since the buffer is big, a human operator gets time to fix the slow consumer and catch up before anything is dropped.

• Only that consumer suffers — everyone else's service is undisturbed. A big operational win: you can consume the production log for experimental development and debugging without fearing for production. When the consumer is off or crashed, its resource consumption stops — all that remains is its offset. (Unlike traditional brokers, where you must purge a dormant consumer's queues so they don't eat active consumers' memory.)
• Replay: in a traditional broker consumption is destructive, but in a log-based one it is like reading a file — a read-only operation that doesn't change the log. The only side effect is advancing the offset, which is the consumer's own business; so you can raise a copy of the consumer at yesterday's offset, write output elsewhere, and reprocess the past with different code as many times as you like — exactly like batch separating derived data from input with a repeatable transform.

Brokers and databases were always seen as two separate categories, but log-based brokers brought database ideas into messaging — and the reverse works too: streaming ideas can be brought into databases. You can even make the event stream the system of record: instead of a data model mutated by updates and deletes, model every state change as an immutable event in an append-only log and derive readable materialized views from it — the event sourcing of Chapter 3. A log-based broker (one that never deletes old events) is well-built for this. But you needn't go as far as event sourcing; even with a mutable model an event stream helps the database — because every database write is itself an event that can be captured, stored and processed. The replication log (Chapter 6) is the leader's write-event stream, and the state-machine replication principle (Chapter 10) is the same: if each event is a write and each replica processes the same events in the same order, all reach the same final state.

No single system covers every need: an OLTP database for user requests, a cache for speed, a text index for search, a warehouse for analytics — each keeps its own copy in its own optimized shape, so they must stay in sync. With the warehouse, periodic ETL (full copy + bulk load, i.e. batch) usually does this. An occasionally used alternative is dual write: application code explicitly writes to each system — first the database, then the search index, then the cache invalidation. But dual writes have two serious problems, the first shown in Diagram 3.

The second dual-write problem is fault tolerance, not concurrency: one write succeeds and the other fails, leaving the two systems inconsistent. Getting “both succeed or both fail” is the atomic-commit problem, which is expensive to solve (Chapter 8's 2PC). With a single-leader database the leader would order the writes and state-machine replication would work; but in the figure above there is no single leader — the database has its own leader and the index its own (a situation like Chapter 6's multi-leader replication). How nice it would be if there were truly one leader — the database, say — with the index following it. Is that practical?

The trouble with most databases' replication logs was that they were treated as internal implementation detail, not public API — for decades there was no documented way to tap the change log. But recently change data capture (CDC) has been taken seriously: observing every change written to the database and extracting it in a form replicable to other systems — more attractive still when changes are immediately available as a stream. Just capture the changes and continuously apply them to the search index; if the change log is applied in the same order, the index data matches the database. The index and every other derived system are just consumers of the change stream — and Diagram 3's race is solved: although requests A and B reach the database concurrently, the database itself orders their execution and writes them in that order to the replication log; the index picks up and applies the same order. If another system (say, the warehouse) is needed later, just add another consumer to the same stream.

In effect CDC makes the source database the leader and everything else a follower, and the log-based broker is well-built for carrying change events because it preserves order. The logical (row-based) replication log does this job with all its challenges (schema change, modelling updates); the open-source Debezium project (with MySQL, PostgreSQL, Oracle connectors and more) and the Kafka Connect framework paved the way; Maxwell for MySQL by parsing the binlog and GoldenGate for Oracle are similar. Like brokers, CDC is usually asynchronous: the record database doesn't wait for the change to be applied downstream — the upside is a slow consumer never bothers the record system; the downside is that all Chapter-6 replication-lag issues apply here too.

• Initial snapshot: if you had the log of all historical changes, replaying it would build the whole state; but keeping all of history is usually too expensive and the log must be shortened. So bringing up a fresh derived system (say, a new text index) needs a consistent snapshot (Chapter 6) — and that snapshot must correspond to a known position/offset in the change log so you know where onward to apply changes from. Some CDC tools have this built in, some left it manual; Debezium offers gradual snapshots via Netflix's DBLog watermarking algorithm.
• Log compaction: the same idea as Chapter 4 storage engines — the engine periodically finds same-key records, drops duplicates and keeps only each key's latest update (Fig. 12-6 with the cat videos!). An update with a special null value (tombstone) means key deletion. A compacted log's disk footprint depends only on current content, not on the number of historical writes. The same idea works in CDC: if every change has a primary key and each update supersedes the previous one, keeping only each key's latest write suffices — and a fresh consumer can scan the whole log from offset zero and take a full database copy without needing a fresh snapshot from the source. Kafka has this feature, letting the broker also play durable storage, not just transient messaging.
• Outbox pattern: with CDC the database's internal schema effectively becomes public API (dropping a column breaks downstream consumers — and since CDC is a stream, its consumer may be a production service, causing customer-facing outages; data contracts exist for this). The way to separate internal from external schema is an outbox table with its own schema that CDC reads instead of the domain model. This is dual-write too — but both writes are in one system (the database) and in one transaction, so it lacks the previous section's challenges. Its price: maintaining the conversion between the two schemas and higher write volume in the database.
• CDC vs event sourcing: both log all changes but at different abstraction levels: in CDC the app works with a mutable database and the log is extracted from a low level (parsing the replication log) — so the extracted order is exactly the real write order and Diagram 3's race never arises. In event sourcing the app logic is explicitly built on immutable application-level events; events state the user's intent, not update mechanics, so a later event usually doesn't override the earlier one and rebuilding the final state needs the whole history (CDC-style compaction is impossible). The choice is situational: event sourcing is a big change (Chapter 3), while CDC can be added to an existing database with minimal change — the writing app may never even know CDC is running.

In Chapter 11 we saw batch profit from input immutability; the same immutability principle powers event sourcing and CDC. We usually think of the database as holding current state (optimized for reading), but any state that changes is the product of events that mutated it over time: the list of free seats is the product of processed reservations; an account balance the product of credits and debits. No matter how much is done and redone, the fact remains that those events happened. The key idea: mutable state and the append-only log of immutable events are not in conflict; they are two sides of one coin. The log of all changes is the evolution of state over time — mathematically inclined: application state is the integral of the event stream over time, and the change stream is the time-derivative of state (Fig. 12-7). As Gray and Reuter put it (1992): “Fundamentally, there is no need to keep a database; the log has all the information. The only reason for storing a database (i.e. the current end of the log) is retrieval efficiency.” And log compaction is the bridge between the two pictures: only each record's latest version remains.

There are two shadows. First, log consumers are usually asynchronous: the user writes and reads from a derived view, not yet seeing their own write (the Chapter-6 “read your own writes”). Synchronizing it needs distributed transactions or waiting, usually impractical — so views update asynchronously. Second, the limits of immutability: some workloads heavily update/delete a small dataset (immutable history explodes), and sometimes law (like GDPR) or leaked sensitive data demands real deletion — i.e. rewriting history and pretending the data was never written (in Datomic called excision). Real deletion is surprisingly hard (copies everywhere, backups deliberately immutable); one way is crypto-shredding: keep data encrypted and forget the key to delete — but that only moves the problem (the key store becomes mutable) and you must decide upfront which data is encrypted with which key.

So far we saw where streams come from (user events, sensors, database writes) and how they are carried. What remains: what you can do with a stream. Three broad ways: 1) write the events into a database/cache/index for others to query (the streaming equivalent of batch uses — especially when the stream consumer is that database's sole writer); 2) deliver events to humans (alert emails, pushes, realtime dashboards); 3) process one or more input streams into one or more output streams — possibly as a multi-stage pipeline. The rest of the chapter is about that third way: the snippet turning a stream into a derived stream (operator / job), kin to Unix processes and Chapter-11 MapReduce jobs — with the same dataflow pattern (reads input read-only, writes output append-only elsewhere) and the same sharding and parallelization. Simple operations like transform and filter work the same way.

• Complex event processing (CEP): the 1990s approach to stream analysis, for apps hunting a specific event pattern — like a regex finding character patterns in a string, CEP defines event patterns in a stream with a declarative language (SQL-like) or GUI, and the engine matches with internal state machines, publishing a complex event detailing the found pattern. The query/data ratio here is the inverse of a normal database: usually data is permanent and queries transient and forgotten; in a CEP engine queries are long-lived and stored, and each fresh event is checked for a match. Examples: Esper, Apama, StreamBase — and distributed processors like Flink and Spark have SQL on streams too.
• Stream analytics: a thin border with CEP, but less after specific sequences and more after aggregation and statistics over volumes: the rate of an event type, rolling averages, comparison with the previous window (trend detection or anomaly alerts). These stats are usually computed over a fixed time range (a window — §5.3). Probabilistic algorithms are sometimes used (Bloom, HyperLogLog for cardinality, percentile estimation) that give approximate answers with far less memory — but that is only optimization; there is nothing inherently approximate about stream processing. Frameworks: Storm, Spark Streaming, Flink, Samza, Beam, Kafka Streams (plus hosted services like Dataflow and Stream Analytics).
• Maintaining materialized views: the database's change stream keeps derived systems (cache, index, warehouse) fresh — and in event sourcing the app state is itself the materialized view. Unlike analytics, a bounded time window doesn't suffice: building a view may need all events (a window reaching back to the beginning of time); Kafka Streams and ksqlDB build exactly this on Kafka compaction. Databases have materialized views but usually refresh them with periodic batch or REFRESH MATERIALIZED VIEW — two flaws for streams: inefficiency (all data reprocessed every time) and staleness (source changes unseen until the next refresh). The general fix is incremental view maintenance (IVM): the SQL query compiles to incremental operators and only changed data is recomputed — so updates get far more frequent and fresher (Materialize, RisingWave, ClickHouse, Feldera: fresh events buffer in memory and periodically apply onto the disk view).
• Two smaller uses: search on a stream (like CEP but for single events against complex criteria — queries are stored and documents matched against them; Elasticsearch's percolator feature) and event-driven architecture vs RPC (Chapter 5): actor frameworks are more a concurrency mechanism than a data-management technique — their communication is transient and one-to-one (not durable and multi-subscriber), with arbitrary cycles (not the well-defined acyclic derivation pipeline); and many don't guarantee delivery on crash unless you add retries.

Since stream processing generalizes the data pipeline to incremental processing of unbounded datasets, batch's join need exists here too — but because a fresh event may arrive at any moment, joining on a stream is harder. Distinguish three kinds:

• Stream–stream join (window join): e.g. for each URL's click rate in search results, the “search” event and the “click” event with the same session id must meet. The click may never come (abandoned search) and its distance varies from seconds to days; network delay may even deliver the click before the search. So you pick a window (e.g. clicks up to one hour later attach to the search). Note: embedding search details in the click event is not a join — that only tells you about clicked ones, while an exact rate also needs clickless searches. Implementation: the processor keeps state (e.g. all last-hour events by session); with each event it adds to the index and checks the other side's index for the same session; on match it emits, and if a search expires clickless it emits “unclicked”.
• Stream–table join (stream enrichment): the same join as Fig. 11-2 (activity events + profile table), but continuous: input is the activity-event stream keyed by user id, output the same events enriched with profile info. Remote lookup is slow and dangerous for the database (Chapter 11); the fix is the database's local copy in the processor (hash join — in-memory hash table if small, local disk index otherwise). Difference from batch: no fixed point-in-time snapshot, since the database changes over the processor's lifetime — so the local copy must be kept fresh with CDC: the processor subscribes to both the profile changelog and the activity stream; on each profile create/change it mutates the local copy. So it is effectively a two-stream join (activity events + profile updates) — except the table side has a window to “the beginning of time” (conceptually infinite) with fresh versions replacing old ones, while the stream side may keep no window at all.
• Table–table join (materialized view maintenance): e.g. a social network's timeline cache — instead of expensive reads (walking all followees and merging), you materialize each user's inbox: on each fresh post, append to all followers' timelines; on post/account deletion, remove from all; on a fresh follow, add recent posts; on unfollow, remove. The processor takes the post stream and the follow stream and keeps the follower database to know which timelines each post updates — i.e. the materialized view of a two-table join (posts JOIN follows grouped by follower). The neat view: if you see the stream as derived from the table (Fig. 12-7) and join as table multiplication, the materialized join's change stream obeys the product-derivative rule: (u·v)′ = u′v + uv′ — each post change joins the latest followers and vice versa.

All three joins share one point: the processor builds state from one input and queries it with the other's records — and the order of state-holding events matters (follow first then unfollow, or the reverse?). In a sharded log like Kafka, order survives only within a shard, not across streams and shards — so if two streams' events happen close together, in what order are they processed? E.g. which activity events does a profile update join with: those processed before it (old profile) or after (new)? If cross-stream order is undefined, the join becomes nondeterministic: rerunning on the same input needn't give the same result. Warehouses call this slowly changing dimension, fixed with a unique ID per version (e.g. each tax-rate change gets a fresh ID and the invoice carries the sale-time rate's ID) — but at the price of keeping all versions and making compaction impossible; the other way is denormalizing (putting the rate in the sale event itself).

Once the event timestamp is known, the next step is defining a window over a time range for aggregation (counts, averages). Four common kinds:

• Tumbling: fixed length, each event in exactly one window (e.g. all 10:03-minute events in one window).
• Hopping: fixed length but consecutive windows overlap (e.g. a 5-minute window with a one-minute hop: 10:03–10:07, then 10:04–10:08).
• Sliding: all events within a given distance of each other (e.g. two events 4.5 minutes apart share a 5-minute window, though no fixed boundary would hold them together).
• Session: no fixed duration — all nearby events of one user, ending when the user has been idle a while (e.g. 30 minutes); common in website analytics.

Window operations usually keep temporary state: sometimes fixed-size (a counter is just one number however big the window grows), sometimes an event buffer until the window ends (sliding windows or stream joins) — so you must make sure the machines have room for the state.

Stream processors deal with time a lot (e.g. “the average of the last five minutes”), and what “the last five minutes” means is surprisingly subtle. In batch, tasks chew through history fast, and when time separation is needed they look at each event's embedded timestamp — looking at the executing machine's clock is meaningless, since run time has nothing to do with occurrence time (a minutes-long batch may read a year of history; the timeline that matters is that year, not those minutes). And that use of the event timestamp makes processing deterministic: rerunning on the same input gives the same result.

But many streaming frameworks window by the processing machine's local clock (processing time): simple, and reasonable when the produce-to-process gap is negligible — but it breaks with any significant lag. Processing lags for a thousand reasons: queues, network failures, contention, consumer restarts, or reprocessing the past after a bugfix. Delay even makes order unpredictable: the user requests server A first then server B, but B's event reaches the broker sooner — the processor sees B first, then A, against occurrence order. The analogy is the Star Wars films: episode IV came in 1977, episode I in 1999; watching in release order (processing time) clashes with story order (episode number = event timestamp) — humans cope, but a streaming algorithm must be explicitly written for it.

The “whose clock” problem runs deeper: an offline mobile app buffers events on-device and sends them hours later (extreme stragglers); the timestamp should be the device-clock interaction time, but the user-controlled device clock is untrustworthy (wrong deliberately or accidentally). The server's arrival time (your own clock) is more precise but less meaningful. One way out is recording three timestamps: occurrence time (device clock), send time (device clock), receive time (server clock — subtracting the second from the third estimates the device skew and corrects the timestamp). This problem isn't stream-only; batch has it too, only in streams you feel time passing more and notice it.

The subtle problem of defining windows with event time: you're never sure whether all of a window's events arrived or more are on the way. E.g. you count events in one-minute windows; minute-37's events are counted and now most inputs are minutes 38 and 39 — when do you declare minute 37's window done and emit its counter? You can time out (no fresh event for that window for a while, so declare it ready), but some events may be buffered on another machine (network partition). You must cope with those late stragglers arriving after the announcement; two broad ways: ignore them (normally a small fraction — metric the dropped count and alert if it grows) or publish a correction (the window's updated value including the laggards, possibly retracting the previous output). Sometimes a special message says “from now on no message with timestamp before t will come” and the consumer triggers the window with it — but with producers on different machines each producer's threshold must be tracked separately, and adding/removing producers gets harder.

In Chapter 11 we saw batch tolerate failure simply: the broken task restarts from scratch on another machine and its partial output is discarded; since input is immutable and each task's output goes to a separate file that appears only on successful completion. The result looks as if no failure happened — as if each input record was processed exactly once, although restarts mean some records were processed several times with only one visible effect. This is called exactly-once semantics (effectively-once is the more precise description). Streams have the same problem but no simpler: waiting for a task's “end” to reveal output is impossible, since the stream is infinite and never ends.

One way is cutting the stream into small blocks and treating each block like a miniature batch: microbatching (in Spark Streaming; batch size usually about a second — an efficiency tradeoff: smaller batches mean more scheduling and coordination overhead, bigger batches mean more latency before results show). Microbatching also builds a tumbling window the size of the batch (in processing time, not event timestamps); jobs wanting bigger windows must explicitly carry state from one microbatch to the next. The different approach in Flink: periodically producing a rolling checkpoint of state and writing it to durable storage; if an operator crashes it restarts from the freshest checkpoint and discards output between the last checkpoint and the crash. Checkpoints are triggered by barriers in the message stream — like microbatch boundaries but without imposing a particular window size. Within a stream processor's boundaries, both approaches give batch's exactly-once.

But the moment output leaves the processor (a database write, a publish to an external broker, sending an email), the framework can no longer discard the broken microbatch's output — a restart makes the external effect happen twice, and microbatch/checkpoint alone can't stop it.

For exactly-once appearance under failure, you must ensure all outputs and side effects of processing one event persist if and only if processing succeeded: messages to downstream operators or external systems (even emails and pushes), database writes, operator state changes, and input-message acknowledgment (including advancing the offset in a log-based broker) — all must happen atomically: all or none. Sound familiar? It is Chapter 8's “exactly-once message processing” with distributed transactions and 2PC. But in a narrower environment such an atomic commit can be implemented efficiently (Google's Dataflow, VoltDB, Kafka): unlike XA, which transacts across heterogeneous technologies, these keep the transaction internal — managing both state changes and messaging inside the framework — and the protocol overhead is amortized by processing several input messages in one transaction.

The other way to discard a broken task's partial output is relying on idempotence (Chapter 5): an operation whose repeated execution has the same effect as running once — like deleting a key in a key-value store (deleting again has no effect) versus incrementing a counter (rerunning means incrementing twice). Even non-idempotent operations can usually be made idempotent with a little metadata: when consuming from Kafka each message has a durable ascending offset; when writing to an external database you can store, alongside the value, the offset of the message that triggered the last write — so you know the update was already applied and skip redoing it (Storm's Trident state management is the same idea). Relying on idempotence needs assumptions: restarts must replay the same messages in the same order (a log-based broker does exactly that), processing must be deterministic, and no other node may concurrently update the same value — on failover from one node to another, fencing may be needed (Chapter 9) so the presumed-dead-but-alive node doesn't interfere. With all these constraints, idempotent operations reach exactly-once at low overhead.

And state? Every stateful process (windowed aggregation, join tables and indexes) must recover state after failure: either keep state in a remote datastore and replicate it (but a remote query per message is slow), or keep state local and periodically replicate so a fresh task reads it after failure and continues loss-free (Flink snapshots periodically; Kafka Streams sends state changes to a separate compacted Kafka topic, CDC-like; VoltDB by redundantly processing each input on several nodes). Sometimes no replication is needed at all because state rebuilds from the input streams (replay a short-window aggregation fast; build a database's local replica from the compacted change stream) — everything depends on infrastructure efficiency, and no way is ideal for all situations.

Now we see all the chapter's choices in one tree: first which broker, then which delivery meaning. Note the tree's main path (start → q1 → q3 → out_log → q4 → out_idem) is today's common recommendation: if you want replay and throughput is high, a log-based broker; and since external effects are usually outside the framework, you effectively reach “practically exactly once” with idempotent writes.

• “Kafka guarantees global order” — no; only inside each partition. Two partitions' events can arrive in any order; if per-entity order is needed, partition-key it.
• “Ack means exactly once” — no; ack means at-least-once (redelivery + lost acks = duplicates). Exactly-once also needs atomic commit or idempotence.
• “Add consumers to go faster” — only up to the partition count; beyond that nodes idle. Raise partitions first.
• “Delete from the log like a queue” — log consumption is read-only; manual deletion breaks everyone else's replay. Retention is managed by retention/compaction, not manual deletes.
• “Dual writes are fixed with try/catch” — no; the ordering race (Fig. 12-4) happens with zero errors and try/catch can't see it. Only a single leader + CDC.
• “CDC is the same as event sourcing” — no; CDC extracts the low-level log of a mutable model (real write order), sourcing writes high-level immutable events (user intent). Compaction doesn't work for the latter.
• “Close windows by the server clock” — under lag you manufacture bad data (Fig. 12-8). Window by event time + watermark; the processor clock only when lag is guaranteed negligible.
• “Always ignore laggards” — it depends: fine for a rate metric, but billing needs corrections (retracting the old). Pick the policy explicitly and metric it.
• “Microbatch means end-to-end exactly-once” — only inside the framework; external effects (database, email) duplicate on retry unless atomic or idempotent.
• “A stream join is like a batch join” — no; input never ends so you need windows and state, cross-stream order is undefined (nondeterminism), and expiry is itself an output (“unclicked”).

• Recite the drop/buffer/backpressure trio with the pipe/TCP example.
• Load balancing (one) vs fan-out (all) + the consumer-group combination.
• Draw the m4-before-m3 mechanism (load balancing + redelivery) from memory.
• Partition/offset: order inside vs across + parallelism ceiling + post-resume duplicates.
• The A/B dual-write race and the CDC fix (single leader, single order).
• Compaction (latest per key + tombstone) and outbox (single-transaction dual write).
• The duality: state = event integral, stream = state derivative.
• The three joins (bounded / infinite / two-sided windows) + the (u·v)′ rule.
• Event time vs processing time + the artificial-spike figure + watermarks and the two laggard policies.
• The four windows (tumbling/hopping/sliding/session) with one example each.
• Exactly-once = microbatch/checkpoint inside + atomic commit/idempotence outside (effectively-once).