حل آموزشی تمرین‌های فصل 12Chapter 12 exercise solutions, explained

• Three partitions, key = user id, fixed key-to-partition routing (versioned if it ever changes). Teaching scale: about 10k clicks/s across the topic, checkpoint once per batch or per second — a teaching number, not a benchmark.
• The record database is transactional (teaching Postgres-like); the consumer group assigns one owner per partition; log replay is ordered inside each partition.
• No external effect is included. Retain active checkpoints while their corresponding state remains valid. Replay older than log retention requires rebuilding or explicit rejection.

An offset is a position within one partition; offset 5 in two partitions identifies two independent records. Commit checkpoints and totals together to avoid double counting or omission. Lock the checkpoint to serialize same-epoch batches; epoch fencing alone only rejects stale owners. Ten thousand clicks/s is a capacity input; choose batch size from lag and commit cost. Active checkpoints do not expire; falling behind log retention requires recovery.

1. Route by key, scope offsets Route every click by user id and treat offsets as partition-scoped: the same number on two partitions orders nothing, so the checkpoint key always carries its (topic, partition).
2. Guard every re-execution Lock and read the partition checkpoint. Offsets at or below it have already applied. Process records in order and advance the checkpoint only with the committed batch effects.
3. Commit effect with checkpoint Commit the effect and its checkpoint in one epoch-guarded transaction, so a crash between apply and commit leaves no half-state and a lost ack replays against an honest memory.
4. Fence the stale writer Install new ownership through a sink ownership-transfer protocol. An old transaction must finish before transfer or fail its epoch check after transfer; Kafka group callbacks alone do not guarantee sink fencing.
5. Name the boundaries Never expire active checkpoints by TTL. If the earliest available record is beyond the next required offset, input is missing: explicitly rebuild or recover rather than silently skip ahead.

Use click_counts(user_id,total) and consumer_checkpoints(topic,partition,last_offset,owner_epoch) with a composite key. A transaction first locks the checkpoint row and validates sink ownership, then applies ordered offsets newer than the checkpoint and commits effect plus checkpoint together. Same-epoch batches also serialize on this row. Broker ACK follows commit. Coordinate new-epoch installation at the database with group ownership; Kafka does not automatically fence database writes. The demo has one fixed topic, clicks.

The demo places u2 at P1 offsets 0..3 and u3 at 4..6; the shared P1 checkpoint becomes 6. After rebalance, u2’s next click is offset 7 and raises its total to 5. Offset 0 in P0 and P2 represents independent records; replay uses each partition checkpoint. A simulated pre-commit crash has no effect, while lost ACKs redeliver without changing totals.

• Event-ID dedup table instead of offset guards: store each applied event_id under a UNIQUE constraint and rely on the constraint to reject replays. Wins when replay is unordered or offsets are unavailable (compacted logs); costs one extra write plus a unique index per event, and the table itself needs a retention horizon.
• Fully transactional streaming (Kafka transactions, Flink checkpoints) with every effect inside the framework: wins when the sink participates in the same transaction; loses the moment the database is external, because the framework cannot roll back a foreign commit.
• At-least-once counting with an approximate sketch (HyperLogLog): wins when the product needs a rate, not an exact audit; loses every correctness argument this exercise makes.

• Crash between apply and commit: nothing was committed, so ordered replay rebuilds the same totals; the demo's garbage +99 vanishes with the uncommitted attempt.
• Lost ack after commit: the broker redelivers already-checkpointed offsets; the guard skips every one (14 skips in the demo) and totals do not move.
• Rebalance mid-batch: the open transaction rolls back, the new epoch-2 owner resumes from checkpoint 104, and only offset 4 (the genuinely new click) applies.
• Stale owner writes after revoke: presumed dead but alive, its epoch-1 commit fails the epoch condition and is fenced without touching totals.
• Offset out of range after retention: the fetched offset no longer exists, so the consumer alerts and re-seeds from a snapshot instead of guessing a total.
• Poison record that never applies: after bounded retries it moves to a dead-letter queue with a metric, and the partition advances past it explicitly.

Run: javac ExactlyOnceCounter.java && java ExactlyOnceCounter — stdlib Java only, instant. Partitions are in-memory lists: this proves the sink logic, not a broker integration.

Expected output: OK totals={u1=5, u2=6, u3=3, u4=2} checkpoints={0=4, 1=8, 2=1}; six scenario assertions pass. The extra u2 click at offset 8 is an explicit duplicate-within-batch test.

Invariants: each user's total equals the sum of its applied deltas; checkpoints are monotonic per partition; a kill -9 plus ordered replay never moves a total. Measure consumer lag per partition, the redelivery-skip rate, and the fenced-write counter; alert when skips grow (broker trouble) or fences fire (ownership trouble). Local-run guide for a real broker (guide only, not executed here): run Kafka locally, create topic clicks with 3 partitions, produce keyed clicks, kill the consumer mid-batch and watch the totals stay put.

• Comparing offsets across partitions as if 104 on P0 meant anything next to 104 on P1. Offsets are per-partition sequences; cross-partition comparison is the original sin of this exercise.
• Committing the offset before the effect (at-most-once counting) or in a separate transaction from the effect (reopening the crash window the design closed).
• Assuming rebalances are instant: the old owner keeps a live batch and a live pen, so without epoch fencing its writes land after the new owner's.
• Claiming exactly-once while an external side effect (push, SMS, email) rides the same path without its own idempotency key — the broker's guarantee never crosses that boundary.

• Search and click streams are co-partitioned by session id (same key, same partition count), so each session's two sides meet on one machine. Teaching load: 1M searches/hour with 200k clicks; average lateness 40s, p99 6 minutes — teaching numbers, not benchmarks.
• Event time comes from the device interaction timestamp, corrected for device skew; duplicates arrive by redelivery and carry stable event ids.
• Corrections are acceptable downstream: the CTR consumer understands versioned values (v1 then v2 with retraction).

Choose windows from lateness distributions. A six-minute p99 suggests shorter retention excludes more late events, subject to what population that percentile describes. State scales roughly as arrival rate × retention duration × bytes/entry, so wider windows cost RAM and recovery time. Watermarks express event-time progress assumptions, not proof all events arrived. Retractions correct previously published results.

1. Buffer both arrival orders Keep a two-sided buffer per session: searches and clicks each wait for the other, so click-before-search joins exactly like search-before-click, and a seen-set on each side drops redeliveries.
2. Watermark the minimum, park idle Run on event time and watermark the minimum over partitions: a partition quiet past the idle timeout is parked and excluded with a meter, so one silent stream cannot hold every window hostage.
3. Budget lateness from data Set allowed lateness from the measured lateness distribution (teaching: 10 minutes over a p99 of 6): close at watermark past end-plus-lateness, because an unmeasured timeout is a guess wearing a number.
4. Correct or drop, explicitly Give laggards two explicit fates: inside lateness, publish a retracting correction (v1 out, v2 in); beyond the horizon, drop, count in late_dropped_total, and alert when the share grows.
5. Estimate 10x honestly Price tenfold growth before it arrives: buffer state scales with rate × window × bytes, so spill cold sessions to local disk and let bounded buffers push backpressure with lag alerts instead of silent drops.

Key both streams by session id with the same partitioner, so every session is processed on one machine with both sides local. State per session: the search list and click list inside the one-hour window, plus a seen-set of event ids on each side. On any arrival, buffer it and probe the other side; on a match emit clicked once (first click wins, later clicks in the same session are evidence, not new joins); when the watermark passes the window end plus allowed lateness, emit unclicked for the unmatched remainder.

CTR is defined as clicked searches over all searches in closed windows, versioned: a correction retracts v1 and publishes v2, and the denominator never changes under a correction. Watermark = minimum of per-partition watermarks; a partition silent past the idle timeout is parked (excluded with a meter) until data resumes.

Teaching numbers: 1M searches/hour ≈ 278/s; each buffered session holds ~200 bytes × 2 sides × 1 hour ≈ a few GB per hour of window — at 10x, tens of GB, so cold sessions spill to local disk while hot ones stay in memory. Lateness p99 of 6 minutes sets allowed lateness to 10; the 0.3% arriving later are dropped and counted, biasing CTR down by at most that share — stated on the dashboard next to the rate, never hidden. Corrections revise fewer than 1 in 300 windows, each retracting exactly one v1.

• Processing-time windows: close by the processor clock. Wins only when produce-to-process lag is guaranteed negligible; under any real lag it manufactures the artificial spike of Figure 12-8, so this exercise rejects it with evidence.
• Embedding search details inside the click event instead of joining: tells you everything about clicked searches and nothing about unclicked ones, so the rate's denominator is gone. Wins for attribution, loses for CTR.
• Stream–table join against a session table with an infinite window: wins when sessions carry rich slowly-changing profile state; costs unbounded table state and the nondeterminism of cross-stream order, where this exercise's bounded two-sided buffer stays honest.

• Click before search: the click waits in the session buffer and joins when the search lands — order-free by construction, shown live in the demo (c2 before q2).
• Idle partition holds the watermark: after the idle timeout it is parked and excluded from the minimum with a meter, so windows close; on data it is readmitted and its backlog replays by event time.
• Device clock skew: occurrence time is corrected with the send/receive pair, so a lying phone clock cannot file an event into the wrong window.
• Duplicate delivery on both sides: stable event ids make the second arrival a skip, so a redelivered click cannot join twice or inflate CTR.
• Rebalance moves a session mid-window: its buffer restores from the changelog on the new owner and replay converges, because window contents are a pure function of event time.
• Tenfold throughput: buffer state grows ~10x with the window (rate × window × bytes), so cold sessions spill to local disk and bounded buffers apply backpressure with lag alerts instead of dropping silently.

Run: javac WindowedJoin.java && java WindowedJoin — stdlib Java only, instant. Watermarks and partitions are simulated values: this proves the join/watermark logic, not a Flink/Kafka deployment.

Expected output: Both orders join (q1×c1, q2×c2 even though c2 arrived first); the watermark advances with partition 2 parked; q3 expires unclicked then is corrected to clicked; c9 is dropped and counted. Totals: emitted=3, unclicked=0, corrections=1, droppedLate=1, CTR 3/3. Ends with OK. (Actual output of this environment.)

Replay determinism: the same input log rebuilds byte-identical windows. Metrics: late_dropped_total, correction rate, watermark lag per partition, idle-park events. Method: plot the lateness CDF from a day of traffic and re-derive allowed lateness monthly; load-test at 10x with disk spill on and verify lag alerts fire before drops. A real Flink/Kafka deployment guide is out of scope here.

• Closing by processing time or by a fixed few minutes picked without measuring: the timeout in the lab brief is only a stand-in until the lateness distribution replaces it.
• A many-to-many join that counts every click-search pair: CTR explodes because one search with three clicks counts three times. First-click-wins (or an explicit stated rule) keeps the metric honest.
• Dropping laggards silently: the dropped share is a metric with an alert, because a growing late fraction means the lateness budget — not the data — was wrong.
• Treating the watermark as a guarantee that nothing late will come: it is a threshold for closing, and the whole drop-or-correct machinery exists because the threshold can be wrong.

• A transactional record database with a logical replication log (teaching Postgres-like) and a log-based broker with compaction and retention. Teaching scale: 50M rows, throttled snapshot at 20k rows/s, WAL retained 3 days — teaching numbers, not benchmarks.
• The indexer is idempotent and owns no state the record cannot rebuild; consumers downstream of the index tolerate versioned corrections but the index itself must converge to the record.
• Begin with the connector-supported snapshot protocol. Replica snapshots require version support and coordinated snapshot/LSN boundaries; replica load and WAL still cost resources.

50M rows / 20k rows/s = 2500 seconds, about 41.7 minutes, a teaching lower bound before overhead and concurrent changes. Three-day WAL retention suffices only if snapshot + outage + catch-up + margin fit and disk capacity supports it. Snapshots cost I/O and version retention even on replicas. Derive LSN boundaries through a consistent snapshot protocol. Event IDs dedup events; entity versions prevent state regression. Key-based compaction replaces neither complete record snapshots nor event history.

1. One transaction, stable contract Commit business and outbox together with versioned payload, event ID, entity key and entity version. Internal column changes preserve consumers only when the compatible payload mapping is maintained during rollout.
2. Snapshot with a boundary Coordinate snapshot and boundary B through the supported connector protocol; an ordinary SELECT with an independently sampled LSN is insufficient. Throttle snapshot work against I/O and lag budgets.
3. Catch up with dedup After installing the snapshot, catch up from the consistent boundary. Event IDs dedup repeated events; entity versions reject stale replay. A state snapshot does not list historical event IDs, so event-ID dedup alone cannot resolve snapshot overlap.
4. Three contracts, stated Keep three contracts apart: the one-off snapshot, the append-only event topic keyed by event id under retention, and the compacted state topic keyed by entity key where tombstones linger until observed. A purged diary never seeds.
5. Acceptance by kill test Prove it with a kill: stop the indexer mid-catch-up, resume, and checksum index versus record at a stated boundary; keep WAL retention wider than the slowest bootstrap and throttle the photo, or re-snapshot loudly.

Commit outbox(event_id UUID PK,entity_key,entity_version,type,payload) with business rows. This design partitions events by entity_key to preserve entity order and carries event_id in payload/headers for dedup. Event-ID keys are also possible but require separate entity-order handling. An optional compacted entity-keyed state topic uses null tombstones. Compaction over unique event keys usually saves no space.

Obtain the complete record snapshot and boundary B through one consistent protocol; a snapshot of a purged outbox is incomplete state. Bulk-load, then apply ordered changes under version guards. Retention must cover snapshot + outage + catch-up with margin; missing history requires explicit rebuilding. Compare checksums only at a valid shared boundary.

At the teaching 50M rows and 20k rows/s, snapshotting takes at least 2500 seconds or 41.7 minutes. Overhead, I/O and catch-up add time; measure service impact. Three-day retention is about 103.7 times that lower bound, not protection against every outage. The demo models an atomic in-memory snapshot and logical crash only; its throughput is not a database benchmark.

• Direct table CDC without an outbox (Debezium on the domain tables): wins on day one with zero code changes; loses the moment an internal column drops, because the internal schema is now public API and a production consumer breaks customer-faced.
• Dual writes from application code (database, then index): loses on both faults of §4.1 — the ordering race with no error at all, and the half-committed pair. Kept here only as the anti-pattern the outbox replaces.
• Periodic batch ETL into the index: wins on simplicity and on tiny catalogs; loses freshness (stale until the next run) and replays history needlessly on every run.

• Snapshot/stream gap: the snapshot records LSN B and streaming starts there, so nothing is missed; the overlap replays and event-ID dedup absorbs it (demo: e3 applies twice, harmlessly).
• Snapshot/stream duplicate: same overlap, same dedup — the guard is the event id, so replaying the boundary twice converges instead of doubling.
• Kill mid-bootstrap: the indexer resumes from its checkpoint, replays, dedups, and the boundary checksum proves convergence (demo part 2: one skip, match true).
• Internal column dropped: the relay reads the outbox contract, never the domain tables, so the pipeline stays green and no consumer pages.
• Tombstone lost before every follower observed it: the delete resurrects on the laggard, so tombstone retention must exceed the slowest follower's lag plus margin.
• Outbox purged before a newcomer arrives: the diary cannot seed, so the newcomer reads the compacted entity-state topic or triggers a fresh throttled snapshot — decided loudly, never by guessing.
• Unbounded snapshot pressure: reads come from a replica at a capped rate with a stated budget (teaching 20k rows/s), and a blown budget slows the snapshot — never the leader.

Run: javac CdcBootstrap.java && java CdcBootstrap — stdlib Java only, instant. The log, topics and snapshot are in-memory models: this proves the boundary/dedup/contract logic, not a Debezium/Kafka deployment.

Expected output: PASS snapshot B=3, pause/resume, duplicate, stale overlap, version regression, tombstone; PASS purged events cannot seed full state; index={p1=v7, p2=navy coat}. This is an in-memory model, not a broker/database crash test.

Acceptance: kill the indexer mid-catch-up and require lag-to-zero plus a boundary checksum match; drop an internal column and require the pipeline to stay green; purge the outbox and require the newcomer runbook to refuse it as seed. Metrics: end-to-end lag, snapshot read rate vs budget, tombstone age vs slowest follower, dedup-skip rate. Real Debezium/Kafka steps are guide-only here.

• Seeding a newcomer from the purged outbox: retention dropped those pages, so replay rebuilds partial history and the missing keys fail silently. The outbox is a diary, not a phone book.
• Compacting the outbox event topic by entity key to 'save space': it destroys the history other consumers need while keeping one copy per entity — the worst of both contracts.
• Snapshotting without recording the LSN boundary: the stream then starts from a guess, and the gap-or-duplicate question has no honest answer.
• Reading internal tables through CDC as if they were a stable contract: the first dropped column pages a production consumer at night.
• Assuming snapshot pressure is zero: an unthrottled bulk export contends with the leader, and the outage it causes is self-inflicted.