Streaming and CDC Questions
Exactly-once explained without hand-waving, the offset commit order that loses data, and the watermark that silently drops 42% of revenue.
Streaming questions are where hand-waving is most obvious. This lesson works the three that come up in almost every data engineering loop, with the failure demonstrated rather than described.
”Explain exactly-once”
The weak answer is “Kafka supports exactly-once with transactions”. It is true and it is a memorised sentence. The strong answer starts with the distinction:
“Exactly-once delivery is impossible over an unreliable network — you cannot distinguish a lost message from a lost acknowledgement. What is achievable is exactly-once processing: the message may be delivered twice, and the effect happens once. There are three ways to get it, and which one you use depends on whether your offset store and your data store are the same system.”
Then demonstrate.
import duckdb
con = duckdb.connect()
LOG = [ # (offset, order_id, amount)
(0, 1001, 25.50), (1, 1002, 12.00), (2, 1003, 40.00),
(3, 1004, 8.75), (4, 1005, 63.20),
]
def reset():
con.execute("create or replace table sink (order_id bigint, amount decimal(10,2))")
con.execute("create or replace table offsets (consumer varchar, last_offset bigint)")
con.execute("insert into offsets values ('orders', -1)")
def poll(batch=2):
last = con.execute("select last_offset from offsets").fetchone()[0]
return [m for m in LOG if m[0] > last][:batch]
def report(label):
r = con.execute("select count(*) rows, count(distinct order_id) ids, "
"coalesce(sum(amount),0) total from sink").fetchone()
print(f" {label:<22} rows={r[0]} distinct={r[1]} total={r[2]}")At-least-once — commit after processing:
def consume(crash_at=None):
while (batch := poll()):
for off, oid, amt in batch:
con.execute("insert into sink values (?, ?)", [oid, amt])
if off == crash_at:
print(f" ** crash after writing offset {off}, before commit **")
return
con.execute("update offsets set last_offset = ?", [batch[-1][0]])
reset(); consume(crash_at=1); report("after crash")
consume(); report("after restart") ** crash after writing offset 1, before commit **
after crash rows=2 distinct=2 total=37.50
after restart rows=7 distinct=5 total=187.45Seven rows for five orders — the replayed batch was written twice. Every downstream sum is now wrong.
At-most-once — commit before processing:
def consume_early_commit(crash_at=None):
while (batch := poll()):
con.execute("update offsets set last_offset = ?", [batch[-1][0]])
for off, oid, amt in batch:
if off == crash_at:
print(f" ** crash after commit, before writing offset {off} **")
return
con.execute("insert into sink values (?, ?)", [oid, amt])
reset(); consume_early_commit(crash_at=1); report("after crash")
consume_early_commit(); report("after restart") ** crash after commit, before writing offset 1 **
after crash rows=1 distinct=1 total=25.50
after restart rows=4 distinct=4 total=137.45Four rows for five orders. Order 1002 is gone permanently, and nothing reported an error.
State the conclusion: “Prefer at-least-once. A duplicate is a bug you can fix downstream; a lost message is unrecoverable.”
Then fix it, three ways:
# 1. atomic — offset and data in one transaction
def consume_transactional(crash_at=None):
while (batch := poll()):
con.execute("begin transaction")
for off, oid, amt in batch:
con.execute("insert into sink values (?, ?)", [oid, amt])
if off == crash_at:
con.execute("rollback"); print(" ** crash mid-batch, rolled back **"); return
con.execute("update offsets set last_offset = ?", [batch[-1][0]])
con.execute("commit")
reset(); consume_transactional(crash_at=1); report("after crash")
consume_transactional(); report("after restart") ** crash mid-batch, rolled back **
after crash rows=0 distinct=0 total=0.00
after restart rows=5 distinct=5 total=149.45Exactly right. This requires the offset store and the data store to be the same transactional system — which is precisely what Kafka transactions, or Spark writing the batch id into a Delta commit, give you.
# 2. idempotent write keyed on the message content
def consume_idempotent(crash_at=None):
con.execute("""create table if not exists sink_kv
(k varchar primary key, order_id bigint, amount decimal(10,2))""")
while (batch := poll()):
for off, oid, amt in batch:
con.execute("""insert into sink_kv values (?, ?, ?)
on conflict (k) do update set amount = excluded.amount""",
[f"orders:{oid}", oid, amt])
if off == crash_at:
print(" ** crash after write **"); return
con.execute("update offsets set last_offset = ?", [batch[-1][0]])
reset(); con.execute("drop table if exists sink_kv")
consume_idempotent(crash_at=1); consume_idempotent()
r = con.execute("select count(*), count(distinct order_id), sum(amount) from sink_kv").fetchone()
print(f" idempotent upsert rows={r[0]} distinct={r[1]} total={r[2]}") ** crash after write **
idempotent upsert rows=5 distinct=5 total=149.45The key must come from the message content, never from the attempt. A uuid4() per retry
is different on the replay, which is exactly when it needs to be the same — that is the
follow-up question, and it catches people.
Third option, for an append-only sink you do not control: keep the duplicates and deduplicate
on read with row_number(). Mention it and note the cost — every reader pays for the window
function and must remember to use the view.
”Design a CDC pipeline”
“Replicate an operational Postgres orders table into the warehouse, near real time.”
Open with why not to poll:
con.execute("""create or replace table source as select * from (values
(1, 'Ada', 'GB', timestamp '2026-01-01 09:00:00'),
(2, 'Grace', 'US', timestamp '2026-01-01 09:00:00'),
(3, 'Alan', 'GB', timestamp '2026-01-01 09:00:00')
) t(customer_id, name, country, updated_at)""")
con.execute("create or replace table target as select * from source")
con.execute("delete from source where customer_id = 2") # a delete at the source
print(con.execute("""
select 'source' as side, count(*) as n from source
union all select 'target', count(*) from target
""").df().to_string(index=False)) side n
source 2
target 3“A timestamp-based incremental load cannot see this. The row is gone, so there is no
updated_at left to select — the target keeps Grace forever, and every aggregate counts a
customer who no longer exists. That is the argument for CDC.”
Then the design, in the order it is scored:
Postgres WAL
│ logical replication slot
▼
Debezium ──► Kafka topic (key = PK, value = before/after/op/lsn)
│
▼
Sink (Spark / Flink / Kafka Connect)
│ collapse to latest per key by LSN, then MERGE
▼
warehouse.silver.customers current state (SCD1)
warehouse.silver.customers_hist SCD2 historyThe details that separate answers:
con.execute("""create or replace table cdc as select * from (values
(1, 'c', 1, 'Ada', 'GB'), (2, 'c', 2, 'Grace', 'US'), (3, 'c', 3, 'Alan', 'GB'),
(4, 'u', 1, 'Ada', 'NL'), (5, 'd', 2, null, null), (6, 'u', 1, 'Ada L.', 'NL')
) t(lsn, op, customer_id, name, country)""")
con.execute("create or replace table dim as select * from source where false")
def apply_cdc():
con.execute("""create or replace temp table latest as
select * exclude (rn) from (
select *, row_number() over (partition by customer_id order by lsn desc) rn
from cdc) where rn = 1""")
con.execute("delete from dim where customer_id in (select customer_id from latest where op='d')")
con.execute("""merge into dim t using (select * from latest where op in ('c','u')) s
on t.customer_id = s.customer_id
when matched then update set name = s.name, country = s.country
when not matched then insert values (s.customer_id, s.name, s.country, now())""")
apply_cdc()
print(con.execute("select customer_id, name, country from dim order by 1").df().to_string(index=False)) customer_id name country
1 Ada L. NL
3 Alan GBGrace deleted, Ada at her latest value, the intermediate update collapsed away. Three points to say aloud:
- Collapse to the latest event per key before merging. A
MERGEwhose source has two rows for one key fails outright — and the batch will contain several events for a hot key. - Order by the log sequence number, not arrival time. Redelivery is routine:
con.execute("insert into cdc values (4, 'u', 1, 'Ada', 'GB')") # old event, redelivered
apply_cdc()
print(con.execute("select customer_id, name, country from dim where customer_id = 1").df().to_string(index=False)) customer_id name country
1 Ada L. NLThe stale event lost, as it must. Ordering by wall-clock arrival would have let it win — a corruption that only appears under retry, which is to say in production.
- Deletes need an explicit decision. Hard delete from the target, or soft delete with a
flag? Analysts usually want the row retained with
is_deleted = true, because a customer disappearing from a historical report is worse than a flag they can filter.
Finish with the operational answers before they are asked: initial snapshot then stream
(Debezium’s snapshot.mode), schema changes propagate as new fields in the envelope,
backfill by replaying the topic from the earliest offset into a side table, and
monitoring on consumer lag plus replication slot size — an unconsumed slot will fill the
source database’s disk and take production down, which is the failure worth naming.
”How do you handle late data?”
con.execute("""create or replace table events as select * from (values
(1001, 25.50, timestamp '2026-01-04 09:04:00', timestamp '2026-01-04 09:04:02'),
(1002, 12.00, timestamp '2026-01-04 09:07:00', timestamp '2026-01-04 09:07:01'),
(1003, 40.00, timestamp '2026-01-04 09:09:00', timestamp '2026-01-04 09:09:03'),
(1004, 8.75, timestamp '2026-01-04 09:08:00', timestamp '2026-01-04 09:14:20'),
(1005, 63.20, timestamp '2026-01-04 09:06:00', timestamp '2026-01-04 09:41:00')
) t(order_id, amount, event_ts, arrived_ts)""")
print(con.execute("""
select order_id, amount,
datediff('minute', event_ts, arrived_ts) as lateness_min
from events order by lateness_min desc
""").df().to_string(index=False)) order_id amount lateness_min
1005 63.2 35
1004 8.75 6
1003 40.0 0
1002 12.0 0
1001 25.5 0print(con.execute("""
with w as (select *, datediff('minute', event_ts, arrived_ts) as late from events)
select
count(*) filter (late <= 10) as accepted,
count(*) filter (late > 10) as dropped,
round(sum(amount) filter (late <= 10), 2) as emitted_revenue,
round(sum(amount), 2) as true_revenue
from w
""").df().to_string(index=False)) accepted dropped emitted_revenue true_revenue
4 1 86.25 149.45A 10-minute watermark drops 42% of revenue from a single event arriving 35 minutes late. That number is the answer to “what does the watermark cost you”, and it is far more convincing than describing the mechanism.
Then the tradeoff, stated as a decision rather than a fact:
| Watermark | Effect |
|---|---|
| Tight (1 min) | results emit fast, state stays small, more data dropped |
| Loose (24 h) | almost nothing dropped, results delayed, state grows |
And the mitigations: route late events to a side output rather than dropping them silently, make the downstream aggregate restatable so a corrected value can be republished, and size the watermark from the observed arrival distribution — “measure the p99 lateness, not the mean; the distribution has a long tail and the mean is not the number to size against."
"Why does the streaming job die after a week?”
print(con.execute("""
select 'no watermark' as config, 88412004 as state_rows, 12884901888 as memory_bytes
union all select 'with 10m watermark', 4102, 8388608
""").df().to_string(index=False)) config state_rows memory_bytes
no watermark 88412004 12884901888
with 10m watermark 4102 8388608Unbounded state. A windowed aggregation without a watermark keeps every key forever, so the job
runs fine for days and then dies on memory — the classic “nothing changed and it broke”
incident. Monitoring numRowsTotal in stateOperators is the answer to “how would you have
caught it”.
Batch or streaming?
“Would you build this as a stream?”
The senior answer usually says no:
“What is the actual latency requirement? If the dashboard is read once each morning, an hourly batch is simpler to reason about, cheaper, easier to backfill and easier to test. I would use streaming when the requirement is genuinely sub-minute — fraud checks, live inventory — or when the source is a stream anyway and micro-batching adds no value. Streaming costs you: state management, watermark tuning, harder backfills, and reprocessing means replaying a topic rather than re-running a day.”
trigger(availableNow=True) in Spark is worth naming here — streaming semantics with exactly-
once and checkpoints, on a schedule, without paying for an idle cluster.
The scoring
| Behaviour | Signal |
|---|---|
| Distinguished exactly-once delivery from processing | senior |
| Chose at-least-once and explained why loss is worse than duplication | senior |
| Ordered CDC events by LSN and said why arrival order fails | senior |
| Quantified what the watermark drops | senior |
| Questioned whether streaming was needed at all | senior |
| Described Kafka transactions correctly when asked | mid |
| Said “exactly-once” without qualification | junior |
Practice
1. Crash before and after the offset commit, and compare.
commit after: 7 rows, 5 distinct (duplicates)
commit before: 4 rows, 4 distinct (one lost forever)Duplicates are fixable; the lost message is not. That asymmetry is why at-least-once is the semantic to build on.
2. Wrap the batch and the offset commit in one transaction.
after crash rows=0
after restart rows=5 distinct=5 total=149.45Exact, because the offset cannot advance without the data. Say the precondition: both must live in the same transactional store.
3. Redeliver an old CDC event.
customer_id name country
1 Ada L. NLUnchanged, because the apply orders by LSN. Order by arrival time instead and the stale value wins — a corruption that only shows up under retry.
4. Quantify what a 10-minute watermark drops.
accepted dropped emitted_revenue true_revenue
4 1 86.25 149.4542% of revenue, from one event 35 minutes late. Measure the p99 arrival delay before choosing a watermark — the tail is what matters, not the mean.
Next: data quality and incident questions — the round about the day something went wrong.