DDIA Ch 11 — Notes & Questions

Chapter 11: Stream Processing — Study Notes

Batch processing (Ch. 10) operates on bounded, fixed-size inputs. Stream processing
does the same continuously over unbounded (never-ending) inputs. The central abstraction
is the event: a small, self-contained, immutable record of something that happened at a
point in time, usually given a timestamp by a clock. Producers/publishers generate events;
related events are grouped into a topic or stream and read by consumers/subscribers.

Polling a datastore for new events is wasteful at low latency (most polls return nothing), so
consumers should be notified instead. Databases handle this poorly (triggers are limited),
so specialized tools exist.

1. Transmitting Event Streams

Messaging Systems

A messaging system pushes events from producers to consumers. Unlike a Unix pipe /
TCP connection (one sender, one receiver), it supports multiple producers and consumers
on a topic (publish/subscribe model). Two key questions differentiate systems:

1. What if producers outpace consumers? Three options:

Drop messages — acceptable for some metrics/sensors.
Buffer in a queue — must understand growth behavior (crash? spill to disk?).
Backpressure (flow control) — block the producer. Used by Unix pipes and TCP
(small fixed buffer).
2. What if nodes crash? Durability requires writing to disk and/or replication, which has
a cost. Dropping messages buys higher throughput/lower latency. Whether loss is
acceptable is application-dependent (occasional missing sensor reading is fine; lost
count events corrupt counters).

Batch processing gives a strong reliability guarantee: failed tasks retry

automatically, partial output is discarded — output looks as if no failure occurred.
Streaming aims to replicate this.
Direct Messaging (Producer → Consumer)

No intermediary node. Works in narrow situations but requires app code to handle message
loss; tolerates only limited faults (assumes producers/consumers are always online): - UDP
multicast — stock market feeds (low latency); app-level recovery of lost packets. -
Brokerless libraries — ZeroMQ, nanomsg (pub/sub over TCP/IP multicast). - StatsD /
Brubeck — UDP for metrics (counts only correct if all messages received → approximate). -
Webhooks — producer makes direct HTTP/RPC call to a consumer’s registered callback
URL.

Message Brokers

A message broker (message queue) is essentially a database optimized for message

streams. Runs as a server; producers and consumers are clients. Centralizing data in the
broker tolerates clients that come and go, and moves durability to the broker. Consumers
are typically asynchronous: the producer waits only for the broker to buffer the message,
not for processing.

Broker vs. database: - Databases keep data until explicitly deleted; brokers usually delete
a message once delivered — not suitable for long-term storage. - Brokers assume a small
working set (short queues); throughput degrades if many messages must be buffered. -
Databases support secondary indexes/queries; brokers support subscribing to topic subsets
matching a pattern. - Database query = point-in-time snapshot (no notification on later
change); broker = notifies clients of new data but no arbitrary queries.

Standards: JMS, AMQP. Implementations: RabbitMQ, ActiveMQ, HornetQ, Qpid, TIBCO,

IBM MQ, Azure Service Bus, Google Cloud Pub/Sub. Some support 2PC via XA/JTA.

Multiple Consumers — two patterns (often combined)

Load balancing — each message goes to one of the consumers; parallelizes

expensive processing. (AMQP: multiple clients on one queue; JMS: shared
subscription.)
Fan-out — each message delivered to all consumers; independent consumers tune
into the same broadcast. (JMS: topic subscriptions; AMQP: exchange bindings.)

Acknowledgments and Redelivery

To avoid loss, a consumer must acknowledge a message before the broker removes it. No
ack (closed/timed-out connection) → broker redelivers to another consumer. (Ack may have
been lost after full processing → needs an atomic commit protocol to handle.)
 Redelivery + load balancing reorders messages: an unacked message redelivered
later interleaves out of original order. Even when JMS/AMQP try to preserve order, this
combination breaks it. Fix: one queue per consumer (no load balancing). Reordering
matters only when messages have causal dependencies.

Partitioned Logs

Traditional messaging is built on a transient mindset — messages deleted after delivery.

Databases/filesystems are the opposite: writes persist. This matters for derived data — you
can rerun batch jobs on read-only input, but AMQP/JMS message receipt is destructive
(can’t reread). New consumers only see messages sent after they register.

Log-based message brokers combine durable log storage with low-latency notification: - A
log is an append-only sequence of records on disk (cf. log-structured storage, WAL,
replication logs). Producer appends; consumer reads sequentially, then waits for new
appends (like tail -f ). - Partition the log for throughput: partitions hosted on different
machines, read/written independently. A topic = group of partitions carrying the same
message type. - Within a partition, each message gets a monotonically increasing offset
(sequence number); messages within a partition are totally ordered. No ordering across
partitions.

Examples: Apache Kafka, Amazon Kinesis Streams, Twitter DistributedLog. Achieve

millions of messages/sec via partitioning + fault tolerance via replication.

Logs vs. traditional messaging: - Fan-out is trivial (reading doesn’t delete). Load
balancing assigns entire partitions to consumer-group nodes (coarse-grained). -
Downsides: parallelism capped at the number of partitions; a slow message causes head-
of-line blocking for its partition. - Rule of thumb: JMS/AMQP suits expensive per-message
processing where order is unimportant; the log-based approach suits high throughput,
fast per-message processing, and order-sensitive workloads.

Consumer offsets: Broker tracks only periodic consumer offsets (not per-message acks) —
everything below the offset is processed. Analogous to the log sequence number in single-
leader DB replication (broker = leader, consumer = follower). On consumer failure, another
node resumes at the last recorded offset — messages after that offset may be processed
twice.

Disk space: Log split into segments; old segments deleted/archived → effectively a
bounded circular/ring buffer on disk (can be large — back-of-envelope: 6 TB drive at 150
MB/s ≈ 11 hours at max write rate; typically days/weeks). Throughput stays constant
regardless of retention (everything written to disk anyway), unlike in-memory brokers that
slow down when spilling to disk.

Slow consumers: Log-based = buffering with a large fixed-size buffer. If a consumer falls
behind older than retained data, it misses messages (only that consumer is affected — big
operational advantage; you can safely tap a production log for dev/test/debug). Monitor
consumer lag and alert.

Replaying old messages: Consuming is read-only; only side effect is the offset moving
forward. The offset is under consumer control → you can replay (e.g., start a copy from
yesterday’s offset). This makes log-based messaging like batch processing: repeatable
transformations, easier experimentation and recovery.

2. Databases and Streams

A write to a database is an event that can be captured, stored, processed. A replication log
is a stream of write events. The state machine replication principle: if every replica
processes the same events in the same order (deterministically), all reach the same state —
just event streams.

Keeping Systems in Sync

No single system satisfies all needs (OLTP DB + cache + full-text index + data warehouse),
and copies must stay in sync. - ETL / batch — periodic full dumps; can be too slow. - Dual
writes — app writes to each system directly. Serious problems: - Race condition —
concurrent writes interleave differently across systems → permanent inconsistency, silently
(Figure 11-4). - Partial failure — one write succeeds, another fails → inconsistency (an
atomic-commit problem, expensive to solve). - Root cause: no single leader. Fix: make one
system the leader and others followers.

Change Data Capture (CDC)

CDC = observing all writes to a database and extracting them as a stream so they can be
replicated to other systems in the same order. Makes the source DB the leader and derived
systems (search index, cache, warehouse) followers — they are just consumers of the
change stream. A log-based broker is ideal (preserves order).

Implementation: triggers (fragile, high overhead) or parsing the replication log

(more robust; must handle schema changes).
 Examples: LinkedIn Databus, Facebook Wormhole, Yahoo Sherpa; Bottled Water
(PostgreSQL WAL), Maxwell/Debezium (MySQL binlog), Mongoriver (MongoDB
oplog), GoldenGate (Oracle).
Usually asynchronous → operationally nice but subject to replication lag.

Initial snapshot: Keeping all changes forever is often infeasible, so you start a new derived
system from a consistent snapshot that corresponds to a known offset in the change log,
then apply changes from there.

Log compaction: Storage engine keeps only the most recent value per key, discarding
overwritten versions; a tombstone marks deletion. Disk usage depends on current DB
contents, not write history. With CDC, lets you rebuild a derived system by scanning a
compacted topic from offset 0 — no new snapshot needed. Supported by Kafka.

API support for change streams: Increasingly first-class — RethinkDB query

subscriptions, Firebase/CouchDB change feeds, Meteor (MongoDB oplog), VoltDB
streaming export, Kafka Connect.

Event Sourcing

From the domain-driven design (DDD) community. Like CDC, stores all state changes as a
log of change events — but at a different abstraction level: - CDC — app mutates DB
freely; log extracted at low level (replication log). App is unaware. - Event sourcing — app
logic is built on immutable events at the application level; the event store is append-only,
updates/deletes discouraged. Events reflect user intent (“student cancelled enrollment”), not
low-level state changes.

Benefits: meaningful data model, easier app evolution, better debugging/auditing, guards
against bugs. Similar to the chronicle data model and the fact table in a star schema. Tools:
Event Store — but a plain DB or log broker works too.

Deriving current state: Users want current state, not history → replay/transform the event
log into a read-optimized materialized view (must be deterministic). Unlike CDC, log
compaction can’t work the same way — events express intent and don’t overwrite prior
events, so you need the full history. Use periodic snapshots as a performance optimization
only.

Commands vs. events: A request arrives as a command (may still fail validation). Once
validated/accepted it becomes an event — durable, immutable, a fact. Consumers cannot
reject an event. Validation must be synchronous before the event is generated (e.g.,
serializable transaction), or split into a tentative event + later confirmation event.
State, Streams, and Immutability

Mutable state and an append-only changelog of immutable events are two sides of the
same coin — state is always the result of a sequence of events. - Analogy: state = integral
of the event stream over time; change stream = derivative of state by time (Figure 11-
6). - Pat Helland: “The truth is the log. The database is a cache of a subset of the log.” Log
compaction bridges log and database state.

Advantages of immutable events: - Old idea — accountants’ append-only ledgers;

mistakes are fixed with a compensating transaction, never erased (auditability). - Easier
recovery from buggy code than destructive overwrites. - Captures more information than
current state (e.g., item added then removed from cart — useful for analytics).

Deriving several views from one log: Separate write form from read form → derive
multiple read-optimized representations from the same event log (Druid ingests from Kafka,
etc.). Makes app evolution easy: build a new view alongside old systems, then retire the old.
This is command query responsibility segregation (CQRS). Renders the
normalization/denormalization debate largely moot — denormalize freely in read views.

Concurrency control: Biggest downside — consumers are asynchronous, so a user may

not read their own write immediately (“reading your own writes” problem). Solutions:
synchronous read-view update (needs a transaction), or total-order-broadcast approach. But
event sourcing also simplifies concurrency: a self-contained event = a single atomic
append; if log and state are partitioned the same way, a single-threaded consumer needs no
write concurrency control (serial order removes nondeterminism).

Limitations of immutability: Feasibility depends on churn. High update/delete rates on a

small dataset → history grows huge; fragmentation, compaction/GC become critical. Also,
you may need to truly delete data (privacy regulations, leaks) — appending a “deleted”
event isn’t enough; you must rewrite history (Datomic excision, Fossil shunning). Truly
deleting data is surprisingly hard (copies in storage engines, SSDs, backups).

3. Processing Streams

Three things you can do with a stream: 1. Write to storage (DB, cache, search index) for
later querying — streaming equivalent of batch output. 2. Push to users — email/push
notifications, real-time dashboards (a human is the consumer). 3. Process input stream(s)
→ output stream(s) — pipelines of processing stages. Code = an operator or job (like
Unix processes / MapReduce; consumes read-only input, appends output).
Partitioning/parallelization patterns mirror MapReduce. Crucial difference: a stream never
ends → no sorting, no sort-merge joins; restarting from the beginning after a crash may be
infeasible for a long-running job.

Uses of Stream Processing

Originally monitoring (fraud detection, trading, manufacturing, military/intelligence).

Complex Event Processing (CEP) — search for event patterns (like a regex over an
event stream). Uses declarative queries (SQL-like) or a GUI; engine maintains a state
machine and emits a complex event on match. Roles reversed vs. a DB: queries are
stored long-term, data flows past them. Tools: Esper, IBM InfoSphere Streams,
Apama, TIBCO StreamBase, SQLstream.
Stream analytics — aggregations and statistical metrics over many events (rates,
rolling averages, percentiles), computed over windows. Often uses probabilistic
algorithms (Bloom filters, HyperLogLog for cardinality, percentile estimators) to save
memory — these are an optimization, not inherent inexactness. Frameworks: Storm,
Spark Streaming, Flink, Concord, Samza, Kafka Streams; hosted: Google Cloud
Dataflow, Azure Stream Analytics.
Maintaining materialized views — keep caches/indexes/warehouses (and event-
sourced app state) up to date. Needs a window stretching back to the beginning of
time (minus compacted obsolete events), unlike time-windowed analytics. Samza,
Kafka Streams support this.
Search on streams — store queries, run documents past them (e.g., media
monitoring, real-estate alerts). Elasticsearch percolator. Index the queries to scale.
Message passing & RPC — actor frameworks are related but not stream processors
(ephemeral, one-to-one, arbitrary/cyclic comms vs. durable, multi-subscriber, acyclic
pipelines). Some crossover (Storm distributed RPC).

Reasoning About Time

Event time (when it happened, from the event’s timestamp) vs. processing time
(machine’s local clock when processed). Batch uses event time naturally; many stream
frameworks default to processing time — fine only if processing lag is negligible.
Causes of delay/reordering: queueing, network faults, contention, consumer restarts,
reprocessing. Messages can arrive out of order (Star Wars release-order analogy:
episode number = event time, watch date = processing time).
Confusing the two → bad data: a restart that processes a backlog looks like an
anomalous request spike if measured by processing time (Figure 11-7).
 Knowing when a window is complete: You can never be sure all events for a
window have arrived. Straggler events arrive late. Two options:
1. Ignore stragglers (track/alert on dropped count).
2. Publish a correction (updated window value; may need to retract prior output).
A special marker message (“no more messages with timestamp < t”) can trigger
windows, but is tricky with multiple producers.
Whose clock? Device clocks (e.g., mobile app buffering offline) are untrustworthy;
server-receipt time is accurate but less meaningful. Fix: log three timestamps —
event time (device), send time (device), receive time (server) — then receive − send
estimates the device-clock offset to correct the event timestamp.

Types of windows: - Tumbling — fixed length, non-overlapping; every event in exactly one
window (round timestamp down). - Hopping — fixed length but overlapping (e.g., 5-min
window, 1-min hop) for smoothing. - Sliding — all events within an interval of each other
(buffer sorted by time, expire old). - Session — no fixed duration; groups events for one
user until an inactivity gap (e.g., 30 min). Common for website sessionization.

Stream Joins

New events arriving anytime makes joins harder than in batch. Three types:

Stream-stream join (window join) — e.g., join search events with click events by
session ID to compute click-through rate. Click may never come or arrive much later
(or even before the search). Choose a join window. Processor must maintain state
(recent events indexed by key) and emit matches (or “not clicked” on expiry).
Stream-table join (stream enrichment) — enrich activity events with DB data (e.g.,
user profile). Remote lookups are slow/risky → keep a local copy of the DB (in-
memory hash or local index), kept fresh via CDC subscription. Effectively a stream-
stream join where the table-changelog side uses an infinite window (newer records
overwrite older) and the stream side may use none.
Table-table join (materialized view maintenance) — both inputs are changelogs
(e.g., Twitter timeline cache: streams of tweets + follow relationships). Maintains a
materialized view of a join query; any change on one side joins with the latest state of
the other (the product rule (u·v)′ = u′v + uv′ ).

Time-dependence of joins: All three maintain state from one input and query it from the
other. Order matters (follow-then-unfollow ≠ unfollow-then-follow). Within a partition, order
is preserved; across streams/partitions, no ordering guarantee → joins become
nondeterministic (rerunning may give a different result due to different interleaving). In data
warehouses this is the slowly changing dimension (SCD) problem — fix by giving each
version a unique ID (e.g., tax rate at time of sale), but then log compaction is impossible
(all versions retained).

Fault Tolerance

Batch achieves exactly-once semantics (more precisely effectively-once) easily:

immutable input, separate output files, output visible only on task success. Streaming can’t
“wait until finished” — the stream is infinite.

Microbatching & checkpointing:

Microbatching (Spark Streaming) — break stream into ~1-second blocks, each
a mini batch. Implicitly a tumbling window by processing time. Smaller = more
overhead; larger = more delay.
Checkpointing (Apache Flink) — periodic rolling checkpoints of state to durable
storage, triggered by barriers in the stream; restart from last checkpoint, discard
output since then. No forced window size.
Both give exactly-once within the framework — but once output leaves the
processor (DB write, external broker, email), a failed-batch retry causes the side
effect twice.
Atomic commit revisited: All outputs and side effects (downstream messages, DB
writes, state changes, offset advance) must take effect atomically — all or none. A
restricted, efficient atomic-commit facility (unlike heterogeneous XA) is used by Google
Cloud Dataflow, VoltDB, and planned for Kafka; transaction overhead amortized
across several messages.
Idempotence: An idempotent operation can be applied multiple times with the same
effect as once (set key = value is idempotent; increment counter is not). Make
operations idempotent with metadata — e.g., store the Kafka offset with the written
value to detect/skip already-applied updates (Storm’s Trident). Assumptions: replay
same messages in same order, deterministic processing, no concurrent updates;
fencing may be needed during failover. Cheap path to exactly-once.
Rebuilding state after failure: Windowed aggregations and join tables/indexes need
recoverable state. Options:
Remote datastore + replication (per-message remote query is slow).
Local state replicated periodically — Flink snapshots to HDFS; Samza/Kafka
Streams replicate state changes to a log-compacted Kafka topic (CDC-style);
VoltDB redundantly processes each message on multiple nodes.
Sometimes rebuild from the input stream (replay a short window, or rebuild a
CDC-maintained local DB copy from the log-compacted change stream).
No universally ideal trade-off — depends on network vs. disk latency/bandwidth.
Key Takeaways

A stream is an unbounded sequence of immutable events; stream processing =

continuous batch processing. Message brokers/event logs are the streaming
equivalent of a filesystem.
Two broker styles: AMQP/JMS (per-message assignment + acks, destructive delete,
good for task queues where order doesn’t matter) vs. log-based (partition-level
assignment, ordered, retained on disk, replayable — good for high-throughput, order-
sensitive, derived-data streaming).
A partitioned log gives total ordering within a partition (via offsets), constant
throughput regardless of retention, and lets a slow/experimental consumer replay
history without disrupting others.
Dual writes cause silent inconsistency (race conditions, partial failures). The fix is a
single leader: CDC (low-level, app-unaware) or event sourcing (app-level immutable
events) turns derived systems into ordered consumers.
Log compaction (keep latest value per key + tombstones) lets a log hold a full DB
copy and bridges “log vs. database state” — but doesn’t apply to intent-based event-
sourced logs.
Immutability (state = integral of events; change stream = derivative of state) enables
auditability, multiple read views (CQRS), and easy recovery — bounded by churn and
true-deletion requirements.
Distinguish event time from processing time; using processing time during
backlogs/restarts produces phantom spikes. Handle straggler events by ignoring
(+alert) or publishing corrections.
Window types: tumbling (fixed, non-overlapping), hopping (fixed, overlapping),
sliding (interval-based), session (inactivity-gap, no fixed length).
Three stream joins: stream-stream (window join), stream-table (enrichment via a
CDC-maintained local copy, infinite window), table-table (materialized view of two
changelogs). Cross-stream ordering makes joins nondeterministic (the SCD
problem).
Achieve exactly-once / effectively-once via microbatching,
checkpointing/barriers, restricted atomic commit, or idempotent writes (often
keyed on the message offset); recover state by local replication or replay from the
input log.
Chapter 11: Stream Processing - Assessment
Questions

Section 1: Conceptual/Reasoning Questions

Question 1

Why is the distinction between “event time” and “processing time” a fundamental challenge
in stream processing? Provide a concrete example where confusing the two leads to
incorrect results.

Question 2

Explain why dual writes (having application code write to multiple systems such as a
database and a search index) are fundamentally flawed for keeping systems in sync. What
are the two distinct categories of problems that arise, and how does change data capture
(CDC) address them?

Question 3

The chapter draws an analogy between application state and an event stream using
calculus: “the application state is what you get when you integrate an event stream over
time, and a change stream is what you get when you differentiate the state by time.” Explain
the practical implications of this relationship. Why does it matter that the “log is the truth and
the database is a cache”?

Question 4

Why does log compaction work differently for change data capture (CDC) versus event
sourcing? What property of CDC events allows compaction, and what property of event-
sourced events prevents it?
Question 5

The chapter states that “exactly-once semantics” is misleading and “effectively-once” is

more accurate. Explain why. What is the actual guarantee being provided, and what are the
two main mechanisms (besides distributed transactions) that stream processors use to
achieve it?

Question 6

Why does the combination of load balancing and message redelivery in traditional message
brokers (AMQP/JMS) inevitably lead to message reordering? How does the log-based
approach solve this problem, and what trade-off does it introduce?

Section 2: Scenario-Based Problems

Question 7

You are building a real-time fraud detection system that needs to correlate a user’s
purchases across multiple payment methods within a 5-minute window. Events arrive out of
order from different payment processors, with network delays varying from 0 to 45 seconds.
Design your approach by addressing: - (a) Which type of window would you use and why? -
(b) How do you handle late-arriving events? - (c) What state must the stream processor
maintain? - (d) What trade-off do you make between latency and correctness?

Question 8

Your company has an OLTP database (PostgreSQL), a full-text search index

(Elasticsearch), a cache (Redis), and a data warehouse (Redshift). Currently, the application
code performs dual writes to all four systems. After reading this chapter, you propose a new
architecture. Describe: - (a) What architecture would you propose and why? - (b) How would
you handle the initial migration (bootstrapping derived systems)? - (c) How would you
handle the addition of a new derived system (e.g., a recommendation engine) six months
later? - (d) What happens if the Elasticsearch consumer falls behind by several hours?

Question 9
You are designing a mobile analytics pipeline. A mobile app buffers events locally when
offline and sends them in batches when connectivity is restored. Some devices have
incorrect system clocks. Events may arrive at the server hours or days after they occurred. -
(a) What three timestamps should you record per event, and how do you estimate the true
event time? - (b) If you use 1-hour tumbling windows for aggregation, how do you handle
events that arrive days late? - (c) What are the implications for your downstream consumers
if you publish corrections to already-closed windows?

Question 10

You need to implement a Twitter-like home timeline feature. When a user posts a tweet, it
should appear in the timelines of all their followers. When a user unfollows someone, that
person’s tweets should be removed. Describe: - (a) What type of stream join is this, and
why? - (b) What streams do you need as input? - (c) What state must the stream processor
maintain? - (d) How does this relate to the concept of maintaining a materialized view?

Question 11

A stream processor enriches click events with user profile data (a stream-table join). The
user profile database is updated roughly 1000 times per second across all users. Click
events arrive at 50,000 events per second. Discuss: - (a) Why is querying the remote
database directly for each click event problematic? - (b) How would you maintain a local
copy of the profile database in the stream processor? - (c) What happens if a profile update
and a click event for the same user arrive at nearly the same time? Is the join deterministic?
- (d) How does this relate to the concept of “slowly changing dimensions”?

Question 12

You are running Apache Flink for stream processing. Your job computes hourly revenue
aggregations and writes results to a PostgreSQL database. The Flink job crashes after
processing 45 minutes of data and writing partial results to PostgreSQL. - (a) Explain why
checkpointing alone does not solve this problem. - (b) How could you use idempotent writes
to achieve effectively-once semantics? - (c) What assumptions must hold for the idempotent
approach to work correctly?
Section 3: Compare & Contrast

Question 13

Compare log-based message brokers (e.g., Apache Kafka) with traditional message brokers
(e.g., RabbitMQ) across the following dimensions: - (a) Message ordering guarantees - (b)
Replay capability and consumer independence - (c) Handling of slow consumers - (d)
Parallelism model and its constraints - (e) When you would choose one over the other

Question 14

Compare change data capture (CDC) and event sourcing along the following dimensions: -
(a) Level of abstraction of events - (b) Application awareness of the event mechanism - (c)
Applicability of log compaction - (d) Relationship between the event log and current state -
(e) Initial snapshot requirements

Question 15

Compare the three types of stream joins (stream-stream, stream-table, table-table) in terms
of: - (a) The nature of the input streams - (b) The window semantics used - (c) The state that
must be maintained - (d) A concrete use case for each

Question 16

Compare microbatching (Spark Streaming) and checkpointing (Apache Flink) as fault-

tolerance mechanisms: - (a) How each achieves exactly-once semantics - (b) Latency
characteristics - (c) Relationship to windowing - (d) Behavior when output leaves the stream
processing framework

Section 4: Mathematical/Quantitative Questions

Question 17

A stream processor uses 1-minute tumbling windows based on event time and allows a late
event tolerance of 30 seconds (events arriving more than 30 seconds after their window
closes are dropped). Events arrive with a clock skew uniformly distributed between 0 and 60
seconds (meaning the delay between event time and arrival at the processor is uniform on
[0, 60] seconds).

a. What fraction of events will be dropped (i.e., arrive after their window has closed
plus the 30-second grace period)?
b. If you increase the late event tolerance to 45 seconds, what fraction would be
dropped?
c. What is the trade-off of increasing the tolerance?

Question 18

A log-based message broker has a disk capacity of 6 TB per node with sequential write
throughput of 150 MB/s. The average message size is 500 bytes and the system ingests
messages at 20% of maximum write bandwidth.

a. How many messages per second can the system sustain?

b. How many hours of data can the disk buffer before old messages are
overwritten?
c. If a consumer goes offline, how long can it remain offline before it starts missing
messages?
d. If you double the number of partitions (and thus nodes), how do these numbers
change?

Question 19

A stream-stream join correlates search events with click events using a 1-hour window. The
system receives 10,000 search events per second and 2,000 click events per second. Each
event is approximately 200 bytes including its session ID index entry.

a. How much state (in GB) must the stream processor maintain for the join window?
b. If you reduce the window to 15 minutes, how does the state requirement
change?
c. If 80% of clicks happen within 5 minutes of the search, what percentage of valid
joins would you miss by using a 5-minute window instead of 1 hour?

Section 5: True/False with Justification

Question 20

True or False: Change data capture makes the database the leader and derived systems
the followers, which is fundamentally the same mechanism as single-leader database
replication.

Justify your answer by explaining the similarities and differences.

Question 21

True or False: Stream processing systems are inherently approximate and lossy because
they use probabilistic algorithms like HyperLogLog and Bloom filters.

Justify your answer.

Question 22

True or False: In event sourcing, once an event has been written to the log, a downstream
consumer can reject it if business rules are violated.

Justify your answer and explain the distinction between commands and events.

Answer Key

Answer 1

Event time is when the event actually occurred; processing time is when the stream
processor handles it. They diverge due to queueing delays, network faults, consumer
restarts, or reprocessing.

Example: A stream processor measuring request rate (requests/second). If it is restarted

and must process a backlog of events, using processing time would show an artificial spike
during backlog processing, while the true request rate (based on event timestamps) was
steady. The chapter uses the Star Wars analogy: the episode number is like event time, and
the date you watched the movie is processing time – they can be wildly out of order.
Answer 2

Dual writes have two categories of problems: 1. Race conditions (concurrency): Two
clients updating the same item may interleave writes differently across systems (e.g.,
database ends with value B, search index ends with value A) with no mechanism to detect
this. 2. Partial failures (fault tolerance): One write may succeed while another fails,
leaving systems permanently inconsistent. Solving this requires atomic commit (2PC), which
is expensive.

CDC addresses both by designating a single system of record (the database) as the leader.
Changes are extracted from its replication log in a defined order and applied to derived
systems as followers. This eliminates race conditions (single ordering) and simplifies fault
tolerance (the log can be replayed).

Answer 3

The practical implications are: - State = integral of events: You can reconstruct any
application state by replaying the full event log from the beginning. This enables building
new derived views, recovering from bugs, and auditing. - Stream = derivative of state: Any
change to the database can be captured as an event and propagated to other systems. -
“Log is truth, database is cache”: The append-only event log is the authoritative system
of record. Mutable database state is a derived, materialized view that can be reconstructed
at any time. This inverts the traditional thinking and makes the system more resilient – if a
derived view is corrupted, rebuild it from the log rather than performing complex repairs.

Answer 4

CDC events typically represent the complete new state of a record keyed by primary key.
The most recent event for a given key fully determines the current value, so older events for
the same key can be discarded during compaction.

Event-sourced events represent user intentions at a higher level of abstraction (e.g.,

“student cancelled enrollment”) rather than state mutations. Later events typically do not
override prior events – they build upon them. The full history is needed to reconstruct current
state (e.g., you cannot skip the “reservation made” event just because a “reservation
cancelled” event exists later, since intermediate logic may depend on the sequence).
Therefore, log compaction in the CDC sense is not possible.

Answer 5

“Exactly-once” is misleading because in a distributed system with retries, messages may

physically be processed more than once. The actual guarantee is that the visible effect in the
output is as if each message was processed only once – hence “effectively-once.”

Two main mechanisms beyond distributed transactions: 1. Microbatching/Checkpointing:

Break the stream into small deterministic units. On failure, discard partial output and replay
from the last checkpoint. Within the framework this provides exactly-once, but external side
effects need additional handling. 2. Idempotence: Make operations safe to retry by
including metadata (e.g., the Kafka offset) with each output write, allowing deduplication.
Requirements: deterministic processing, same replay order, no concurrent writes to the
same value from other nodes.

Answer 6

In traditional brokers with load balancing, messages are distributed across consumers. If
consumer 2 crashes while processing message m3 (while consumer 1 is already processing
m4), m3 gets redelivered to consumer 1, which processes them in order m4, m3 – violating
the original send order.

The log-based approach solves this by assigning entire partitions to consumers. Within a
partition, messages are always read sequentially by a single consumer, preserving order.
The trade-off: the maximum parallelism equals the number of partitions (you cannot have
more consumers than partitions in a consumer group), and a slow message causes head-of-
line blocking for subsequent messages in that partition.

Answer 7

(a) A sliding window of 5 minutes is most appropriate for fraud detection because it captures
all purchases within 5 minutes of each other regardless of fixed time boundaries. A tumbling
window could miss correlations that span window boundaries.
(b) Given delays up to 45 seconds, set a watermark/late tolerance of at least 45 seconds.
Events arriving after this grace period can either be dropped (with monitoring) or trigger a
correction/re-evaluation of previously emitted fraud decisions.

(c) The stream processor must maintain: all purchase events for each user within the sliding
window, indexed by user ID, along with metadata about payment methods. This is effectively
a time-indexed, per-user buffer.

(d) The trade-off: longer grace periods increase correctness (fewer missed correlations) but
increase latency before a fraud determination is final, increase state/memory requirements,
and delay legitimate transactions. For fraud detection, some latency (30-60 seconds) is
usually acceptable in exchange for higher accuracy.

Answer 8

(a) Use CDC on PostgreSQL (e.g., Debezium parsing WAL) to publish changes to a log-
based message broker (Kafka). Elasticsearch, Redis, and Redshift become consumers of
this change stream, each applying changes independently. This makes PostgreSQL the
single leader and eliminates race conditions and partial-failure inconsistencies.

(b) For initial migration: take a consistent snapshot of PostgreSQL at a known log position
(LSN). Bulk-load that snapshot into each derived system. Then start consuming the CDC
stream from that LSN onward. Some CDC tools (Debezium) integrate this snapshot facility.

(c) For a new derived system: either replay the CDC log from the beginning (if using log
compaction, the compacted log contains the latest value for every key) or take a fresh
snapshot + start consuming from that point. No changes to the existing pipeline are needed
– just add a new consumer group.

(d) If Elasticsearch falls behind by hours: because Kafka retains messages on disk
(days/weeks of buffer), Elasticsearch simply catches up by reading faster. Other consumers
are unaffected. You should monitor consumer lag and alert if it grows too large, but there is
no data loss as long as the lag doesn’t exceed retention.

Answer 9
(a) Three timestamps per event: 1. t_event : Time event occurred (device clock) 2. t_sent :
Time event was sent to server (device clock) 3. t_received : Time event arrived at server
(server clock)

Estimated true event time = t_event + (t_received - t_sent) . The difference (t_received -

t_sent)estimates the device clock offset relative to the server clock (assuming negligible
network delay relative to required accuracy).

(b) Options: (1) Ignore straggler events and track the dropped-event rate as a metric –
acceptable if stragglers are a small fraction. (2) Publish corrections: when a late event
arrives, update the aggregate for the affected window and publish an amended result. This
is more complex but more correct.

(c) Downstream consumers must handle retractions/corrections: they may need to replace
previously emitted values, maintain versioned results, or accept eventual consistency in their
displays. Systems must be designed to tolerate and propagate corrections gracefully.

Answer 10

(a) This is a table-table join. Both inputs (tweets and follow relationships) are effectively
database changelogs, and any change on one side must be joined with the current state of
the other side.

(b) Two input streams: (1) Tweet events (create/delete), (2) Follow relationship events
(follow/unfollow).

(c) The stream processor must maintain: the set of followers for each user (from the follow
stream) and potentially recent tweets per user. When a tweet arrives, it looks up the
sender’s followers to know which timelines to update.

(d) The timeline cache is a materialized view of the join SELECT ... FROM tweets JOIN follows

ON follows.followee_id = tweets.sender_id GROUP BY follows.follower_id . The stream

processor incrementally maintains this view as the underlying “tables” change, applying the
product rule: changes to tweets are joined with current followers, and changes to followers
are joined with current tweets.

Answer 11
(a) At 50,000 clicks/second, querying a remote database adds network round-trip latency to
each event, creating a bottleneck. It also risks overloading the database with 50K
queries/second and introduces variable latency that makes the stream processor’s
throughput unpredictable.

(b) Subscribe to the CDC stream of the user profile database in addition to the click event
stream. Maintain a local copy (hash table in memory if it fits, or local disk-based index) that
is updated whenever a profile change event arrives. Click events are joined against this local
copy with no network round-trip.

(c) The join is not necessarily deterministic. If a profile update and click arrive at nearly the
same time, the ordering between the two streams is undefined – the click may be joined with
the old or new profile depending on which event is processed first. On reprocessing, a
different interleaving could produce different results.

(d) This is the “slowly changing dimension” problem from data warehousing. One solution is
to version the profile (give each version a unique ID) and include that version ID in the click
event at creation time, making the join deterministic. However, this prevents log compaction
of the profile changelog since all versions must be retained.

Answer 12

(a) Checkpointing ensures that within the Flink framework, state is consistent and can be
restored. However, the writes to PostgreSQL are external side effects. After a crash, Flink
restarts from the last checkpoint and replays events, but the partial results already written to
PostgreSQL cannot be “un-written.” This causes some results to be written twice.

(b) Include the Kafka offset (or checkpoint ID) with each write to PostgreSQL. Before writing,
check if that offset has already been applied. Alternatively, use upsert semantics where each
output row is keyed by (window_start, dimension) and writing the same aggregation result
again simply overwrites with an identical value.

(c) Assumptions for idempotent approach: (1) On restart, the same messages must be
replayed in the same order (guaranteed by Kafka’s log). (2) Processing must be
deterministic (same input produces same output). (3) No other node concurrently writes to
the same output rows. (4) Fencing may be needed to prevent a “zombie” node from writing
stale results after failover.
Answer 13

Dimension Log-based (Kafka) Traditional (RabbitMQ)

Attempts ordering but redelivery

Total order within a partition
after consumer failure breaks
(a) Ordering guaranteed. No ordering
message order when combined
across partitions.
with load balancing.

Messages retained on disk;

consumers can reset offset to Messages deleted after
re-read old messages. acknowledgment; no replay
(b) Replay
Multiple independent possible. Adding a new consumer
consumer groups read the only sees future messages.
same log without interference.

Buffered on disk
Unbounded in-memory queue
(days/weeks); slow consumer
grows; can degrade broker
(c) Slow only affects itself, not others.
performance. Slow consumers may
consumers If it falls too far behind, it
affect other consumers sharing the
misses messages (bounded
broker.
buffer).

One consumer per partition

Message-level load balancing
(coarse-grained). Max
across consumers (fine-grained).
(d) Parallelism parallelism = number of
Can add consumers freely to
partitions. Head-of-line
parallelize expensive processing.
blocking within a partition.

RabbitMQ: expensive per-message

Kafka: high throughput,
processing, message-level
(e) When to ordering matters, need
parallelism needed, ordering less
choose replay/reprocessing, building
important, task-queue patterns
derived data systems.
(RPC-style async).

Answer 14
 Dimension CDC Event Sourcing

Low-level: captures row-

High-level: captures user
level mutations
intentions and domain events
(a) Abstraction (INSERT/UPDATE/DELETE)
(e.g., “student cancelled
from the database
enrollment”).
replication log.

Application is unaware CDC Application is explicitly

(b) Application is occurring; it writes to the designed around immutable
awareness DB normally. CDC is events; the event store is the
extracted transparently. primary write model.

Not possible in the same way:

Possible: each event
events express intent, and later
contains the full new state
(c) Log compaction events don’t override earlier
for a key, so only the latest
ones – full history is needed to
event per key is needed.
reconstruct state.

Log is derived from mutable Log IS the primary store;

(d) Log vs. state database state (the DB is mutable state is derived by
primary, log is extracted). replaying the event log.

Need a consistent snapshot Applications store snapshots

at a known log position to as a performance optimization,
(e) Snapshots bootstrap new consumers but the event log from the
(unless using log beginning is the authoritative
compaction from offset 0). source.

Answer 15

Dimension Stream-Stream Join Stream-Table Join Table-Table Join

One activity event

Two streams of activity
stream + one Two database
(a) Inputs events (or same stream
database changelog changelog streams.
self-joined).
stream.
 Dimension Stream-Stream Join Stream-Table Join Table-Table Join

Activity stream: no
window (process
event-by-event). Both sides: infinite
Finite time window Table changelog: window; every
(b)
(e.g., 1 hour) – both infinite window change on one side
Window
sides bounded by time. (“beginning of time”) joins with latest state
with newer records of other side.
overwriting older
ones.

Full state of both

A full local copy of the
All events from both tables, plus the
table (maintained via
streams within the join materialized join
(c) State CDC), indexed by the
window, indexed by join result that must be
join key (e.g., user
key (e.g., session ID). incrementally
ID).
maintained.

Correlating search
Maintaining a Twitter
events with click events Enriching click events
(d) Use home timeline cache
within 1 hour to with user profile
case (joining tweets with
compute click-through information.
follow relationships).
rate.

Answer 16

Dimension Microbatching (Spark Streaming) Checkpointing (Flink)

Periodically snapshots
operator state to durable
Treats each small batch as a mini
storage. On failure, rolls
(a) Exactly- MapReduce job. If any task in the batch
back to last checkpoint and
once fails, the entire batch is discarded and
replays from that point,
mechanism retried. Output only becomes visible
discarding any output
when the batch completes successfully.
generated between
checkpoint and crash.
 Dimension Microbatching (Spark Streaming) Checkpointing (Flink)

Can achieve lower latency

since it processes events
Minimum latency equals the batch
continuously; checkpoints
interval (typically ~1 second). Smaller
(b) Latency happen asynchronously in
batches = lower latency but higher
the background without
overhead.
forcing processing
boundaries.

Checkpoints are orthogonal

Implicitly creates a tumbling window to windowing – barrier
equal to the batch size (by processing markers flow through the
(c)
time). Larger event-time windows require dataflow graph without
Windowing
explicit carry-over state across dictating window
microbatches. boundaries. Windows can
be of any type and size.

Neither approach prevents duplicate side

effects if output leaves the framework. A Same limitation as
failed batch may have already written to microbatching – external
(d) External
an external database; retrying the batch side effects cannot be
output
produces duplicates. Both require undone by the framework
idempotent writes or atomic commit for alone.
end-to-end exactly-once.

Answer 17

(a) A 1-minute tumbling window closes at the end of each minute. With 30 seconds of grace,
an event is dropped if it arrives more than 90 seconds after its event timestamp’s window
start (60s window + 30s grace). Since delay is uniform on [0, 60s], an event is dropped if its
delay exceeds the time remaining in its window plus 30 seconds.

More precisely: an event occurring at time t within its window (0 to 60 seconds into the
window) arrives at the processor at time t + d where d ~ Uniform(0, 60) . The window
closes at window_end , and the grace period extends to window_end + 30s . The event is

dropped if t + d > window_end + 30 .

For an event at position p seconds into the window (uniform on [0,60)), it needs d <= (60 -

p) + 30 = 90 - p. Since d ~ Uniform(0,60) : - P(dropped | p) = P(d > 90 - p) = max(0, (60 -

(90-p))/60) = max(0, (p-30)/60)

For p < 30: P(dropped) = 0 For p >= 30: P(dropped) = (p-30)/60

Average across p ~ Uniform(0,60): P(dropped) = (1/60) * integral from 30 to 60 of (p-30)/60

dp = (1/60) * [1/60 * (p-30)^2/2] from 30 to 60 = (1/60) * (900/120) = (1/60) * 7.5 = 1/8 =
12.5%

(b) With 45s tolerance: event dropped if d > (60-p) + 45 = 105 - p, so P(dropped|p) = max(0,
(p-45)/60). Only events with p > 45 can be dropped. P(dropped) = (1/60) * integral from 45 to
60 of (p-45)/60 dp = (1/60) * (225/120) = 225/7200 = 3.125%

(c) Trade-off: increasing tolerance means waiting longer before declaring a window final,
which increases end-to-end latency for results, requires more memory to buffer pending
window state, and delays downstream consumers from receiving finalized aggregations.

Answer 18

(a) Maximum write bandwidth = 150 MB/s. At 20%: effective write rate = 30 MB/s =
30,000,000 bytes/s. With 500-byte messages: 60,000 messages/second.

(b) Disk capacity = 6 TB = 6,000,000 MB. At 30 MB/s write rate: time to fill = 6,000,000 / 30
= 200,000 seconds = approximately 55.6 hours (about 2.3 days).

(c) The consumer can remain offline for up to 55.6 hours before the oldest messages it
hasn’t consumed are overwritten.

(d) Doubling partitions/nodes: each node still has 6 TB and 150 MB/s bandwidth. Per-node
numbers remain the same. Total system throughput doubles (120,000 messages/second
total), and total buffering capacity doubles (12 TB total), but per-partition retention time
remains ~55.6 hours since each partition’s write rate and disk capacity are unchanged.

Answer 19

(a) Total events per hour: (10,000 + 2,000) * 3,600 = 43,200,000 events. At 200 bytes each:
43,200,000 * 200 = 8,640,000,000 bytes = approximately 8.64 GB of state for the full 1-
hour window.

(b) With a 15-minute window: state = 8.64 GB / 4 = approximately 2.16 GB.

(c) If 80% of clicks happen within 5 minutes of the search, then a 5-minute window would
capture 80% of valid click-search correlations. You would miss approximately 20% of valid
joins (the clicks that occur between 5 minutes and 1 hour after the search). Whether this is
acceptable depends on the use case – for approximate click-through rates it may suffice; for
exact analytics it would not.

Answer 20

True, with nuances.

Similarities: CDC does make the database the single source of truth (leader) and derived
systems into followers that apply the stream of changes in order – exactly the pattern of
single-leader replication. The consumer offset is analogous to a replication log sequence
number. Followers can reconnect and resume from their last position.

Differences: In traditional database replication, leader and followers run the same database
software and maintain identical data representations. In CDC, the “followers” are
heterogeneous systems (search indexes, caches, warehouses) that apply changes to
fundamentally different data structures. Also, CDC is typically asynchronous (with replication
lag implications), and the derived systems cannot be promoted to become the new leader –
they are strictly one-way consumers with different schemas and query capabilities.

Answer 21

False. The chapter explicitly states: “there is nothing inherently approximate about stream
processing, and probabilistic algorithms are merely an optimization.” Stream processors can
compute exact results. Probabilistic algorithms (HyperLogLog, Bloom filters, percentile
estimators) are used as performance optimizations to reduce memory usage, but they are
optional. A stream processor can maintain exact counts, exact sets, and exact aggregations
– it simply requires more memory. The association of stream processing with approximation
is a misconception, not an inherent property.
Answer 22

False. In event sourcing, the distinction between commands and events is crucial. A
command is a request that may be rejected (e.g., validation fails). Once validation succeeds
and the command is accepted, it becomes an event – an immutable fact appended to the
log. By the time a downstream consumer sees an event, it is already a committed,
immutable part of the log that may have been seen by other consumers. Consumers are not
allowed to reject events. All validation must happen synchronously before the command
becomes an event (e.g., via a serializable transaction). Alternatively, the pattern of a
tentative reservation followed by a confirmation event allows asynchronous validation
without consumers rejecting committed events.