Streaming: Exactly-Once Semantics

Batch pipelines get to re-run a whole window. Streaming ones cannot: the consumer is halfway through an unbounded log when it crashes, and what happens next decides whether your numbers are right.

Three delivery semantics

A broker and a consumer

# stream.py
import duckdb

con = duckdb.connect("stream.duckdb")

LOG = [  # (offset, order_id, amount, event_ts)
    (0, 1001, 25.50, "2026-01-04 09:14:02"),
    (1, 1002, 12.00, "2026-01-04 09:15:41"),
    (2, 1003, 40.00, "2026-01-04 09:17:02"),
    (3, 1004,  8.75, "2026-01-04 09:19:30"),
    (4, 1005, 63.20, "2026-01-04 09:21:11"),
]

def reset():
    con.execute("create or replace table sink (order_id bigint, amount decimal(10,2), event_ts timestamp)")
    con.execute("create or replace table offsets (consumer varchar primary key, last_offset bigint)")
    con.execute("insert into offsets values ('orders', -1)")

def poll(batch_size=2):
    last = con.execute("select last_offset from offsets where consumer = 'orders'").fetchone()[0]
    return [m for m in LOG if m[0] > last][:batch_size]

def report():
    rows, ids, total = con.execute(
        "select count(*), count(distinct order_id), coalesce(sum(amount), 0) from sink").fetchone()
    print(f"  sink: {rows} rows, {ids} distinct, total {total}")

At-least-once, and the duplicate

def consume_at_least_once(crash_after=None):
    while (batch := poll()):
        for off, oid, amt, ts in batch:
            con.execute("insert into sink values (?, ?, ?)", [oid, amt, ts])
            if crash_after is not None and off == crash_after:
                print(f"  ** crash after writing offset {off}, before commit **")
                return
        con.execute("update offsets set last_offset = ? where consumer = 'orders'", [batch[-1][0]])

reset()
consume_at_least_once(crash_after=1)
report()
print("restarting...")
consume_at_least_once()
report()

  ** crash after writing offset 1, before commit **
  sink: 2 rows, 2 distinct, total 37.50
restarting...
  sink: 7 rows, 5 distinct, total 75.00

Seven rows for five orders. The batch was written but the offset was not committed, so the restart replayed offsets 0 and 1 — and the total is £75.00 against a true £149.45… no, against a true £149.45 the reported figure is simply wrong in both directions once duplicates land. Every downstream sum is now inflated by the replayed batch.

Reverse the order and the failure changes shape:

def consume_at_most_once(crash_after=None):
    while (batch := poll()):
        con.execute("update offsets set last_offset = ? where consumer = 'orders'", [batch[-1][0]])
        for off, oid, amt, ts in batch:
            if crash_after is not None and off == crash_after:
                print(f"  ** crash after committing offset, before writing {off} **")
                return
            con.execute("insert into sink values (?, ?, ?)", [oid, amt, ts])

reset()
consume_at_most_once(crash_after=1)
report()
print("restarting...")
consume_at_most_once()
report()

  ** crash after committing offset, before writing 1 **
  sink: 1 rows, 1 distinct, total 25.50
restarting...
  sink: 4 rows, 4 distinct, total 124.45

Four rows for five orders — order 1002 is gone permanently, and nothing reported an error. Prefer at-least-once: a duplicate is a bug you can fix, a lost message is not.

Fix 1: commit the offset in the same transaction

def consume_transactional(crash_after=None):
    while (batch := poll()):
        con.execute("begin transaction")
        for off, oid, amt, ts in batch:
            con.execute("insert into sink values (?, ?, ?)", [oid, amt, ts])
            if crash_after is not None and off == crash_after:
                con.execute("rollback")
                print(f"  ** crash mid-batch at offset {off}, transaction rolled back **")
                return
        con.execute("update offsets set last_offset = ? where consumer = 'orders'", [batch[-1][0]])
        con.execute("commit")

reset()
consume_transactional(crash_after=1)
report()
print("restarting...")
consume_transactional()
report()

  ** crash mid-batch at offset 1, transaction rolled back **
  sink: 0 rows, 0 distinct, total 0.00
restarting...
  sink: 5 rows, 5 distinct, total 149.45

Exactly right. The data and the offset move together, so the sink can never disagree with the offset. This is what Kafka transactions plus a transactional sink give you, and what Spark Structured Streaming does by writing the batch id into the Delta commit.

It requires the offset store and the data store to be the same transactional system. When they are not — Kafka offsets and an object-store sink — you need one of the next two.

Fix 2: an idempotent key

def consume_idempotent(crash_after=None):
    con.execute("""create table if not exists sink_kv (
        message_key varchar primary key, order_id bigint, amount decimal(10,2), event_ts timestamp)""")
    while (batch := poll()):
        for off, oid, amt, ts in batch:
            key = f"orders:{oid}"                       # derived from the data, stable across replays
            con.execute("""
                insert into sink_kv values (?, ?, ?, ?)
                on conflict (message_key) do update set amount = excluded.amount, event_ts = excluded.event_ts
            """, [key, oid, amt, ts])
            if crash_after is not None and off == crash_after:
                print(f"  ** crash after writing offset {off} **")
                return
        con.execute("update offsets set last_offset = ? where consumer = 'orders'", [batch[-1][0]])

reset()
con.execute("drop table if exists sink_kv")
consume_idempotent(crash_after=1)
print("restarting...")
consume_idempotent()
rows, ids, total = con.execute("select count(*), count(distinct order_id), sum(amount) from sink_kv").fetchone()
print(f"  sink_kv: {rows} rows, {ids} distinct, total {total}")

  ** crash after writing offset 1 **
restarting...
  sink_kv: 5 rows, 5 distinct, total 149.45

The replay rewrote the same two rows instead of appending. The key must come from the message content, never from arrival — a uuid4() per attempt is different on the replay, which is exactly when it needs to be the same.

Fix 3: deduplicate on read

When the sink is append-only by design — an event log, a Parquet directory — accept the duplicates and remove them at query time:

reset()
consume_at_least_once(crash_after=1)
consume_at_least_once()

con.execute("""
    create or replace view sink_deduped as
    select order_id, amount, event_ts
    from (select *, row_number() over (partition by order_id order by event_ts desc) rn from sink)
    where rn = 1
""")

print(con.execute("select count(*) as raw from sink").df().to_string(index=False))
print(con.execute("select count(*) as deduped, sum(amount) as total from sink_deduped").df().to_string(index=False))

 raw
   7

 deduped   total
       5  149.45

Seven physical rows, five logical ones. The trade is that every reader pays for the window function and must remember to use the view — a raw select sum(amount) from sink is still wrong.

Late events and watermarks

Correctness of delivery is only half of it. Windowed aggregates need a bound on lateness:

con.execute("""
    create or replace table events (order_id bigint, amount decimal(10,2),
                                    event_ts timestamp, arrived_ts timestamp)
""")
con.execute("""
    insert into events 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')
""")

print(con.execute("""
    select time_bucket(interval '5 minutes', event_ts) as window_start,
           count(*) as events, sum(amount) as revenue,
           max(datediff('second', event_ts, arrived_ts)) as max_lateness_s
    from events group by 1 order by 1
""").df().to_string(index=False))

        window_start  events  revenue  max_lateness_s
 2026-01-04 09:05:00       3    83.95            2100
 2026-01-04 09:00:00       1    25.50               2
 2026-01-04 09:09:00       1    40.00               3

Order 1005 belongs to the 09:05 window but arrived 35 minutes late. With a 10-minute watermark that window was closed and emitted long before it turned up:

print(con.execute("""
    with watermarked as (
        select *, datediff('minute', event_ts, arrived_ts) as lateness_min from events
    )
    select
        count(*) filter (lateness_min <= 10) as accepted,
        count(*) filter (lateness_min >  10) as dropped_late,
        sum(amount) filter (lateness_min <= 10) as revenue_emitted,
        sum(amount) as revenue_true
    from watermarked
""").df().to_string(index=False))

 accepted  dropped_late  revenue_emitted  revenue_true
        4             1            86.25        149.45

£63.20 dropped — 42% of revenue, from one late event. Widening the watermark keeps more late data at the cost of holding state longer and emitting results later; that trade is the central tuning decision in stream processing, and there is no setting that avoids it.

Two mitigations worth knowing: route late events to a side output rather than dropping them, and make downstream windows restatable so a corrected value can be republished.

What the frameworks give you

All four are one of the three fixes above. Knowing which one your stack uses tells you what it guarantees — and where you still have to supply a key.

Practice

  sink: 7 rows, 5 distinct

  sink: 5 rows, 5 distinct, total 149.45

  sink_kv: 5 rows, 5 distinct, total 149.45

 accepted  dropped_late  revenue_emitted  revenue_true
        4             1            86.25        149.45

Next: schema evolution and data contracts — surviving the change you did not agree to.