Delivery Guarantees¶

Turbine offers two delivery modes per subscription:

At-least-once — the default. Simple, zero overhead, may produce duplicates on a crash.
Exactly-once — opt-in. Outputs are never duplicated, even across crashes. Wraps each batch in a Kafka transaction.

This page covers the contract of each mode and how to choose. After a hard crash, local state can briefly lag the output under exactly-once — see Crash recovery. For bad messages (poison pills), handler exceptions, and compute errors, see Error handling. For pure reference (every parameter and its default), see Configuration.

At a glance¶

	At-least-once	Exactly-once
Output duplicates on crash	Possible (the in-flight batch may re-emit)	Never
Overhead per batch	None	~10–20 % on commit time, often within noise on end-to-end throughput
Requires `Turbine(app_id=...)`	No	Yes
Requires `output`	No (sinks allowed)	Yes — including when the subscription writes only to side outputs; see Limitations
Default	✅	Opt-in

At-least-once (default)¶

You get this without doing anything special:

from turbine import KafkaBroker, RecordBatch, Turbine

kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(kafka)

@app.subscribe(kafka.topic("events"), output=kafka.topic("enriched"))
def enrich(batch: RecordBatch) -> RecordBatch:
    return enriched(batch)

app.run()

How it works: each batch is processed, the output messages are produced fire-and-forget, the state changes are persisted, and only then is the input offset committed. If the worker crashes mid-batch, the batch is replayed from the last committed offset on the next start — which means the same output messages may be emitted twice.

Use this when your downstream is idempotent. Upserts on a primary key, dedup keys carried on the message, append-only stores where a few duplicate rows are tolerable — all of these absorb at-least-once gracefully.

Exactly-once¶

Enable per subscription:

from turbine import KafkaBroker, RecordBatch, Turbine

kafka = KafkaBroker(bootstrap="kafka:9092")
app = Turbine(
    kafka,
    app_id="orders-enrichment-prod",   # ← required as soon as any subscribe is EOS
)

@app.subscribe(
    kafka.topic("input-events"),
    output=kafka.topic("enriched-events"),
    processing_guarantee="exactly_once",
)
def enrich(batch: RecordBatch) -> RecordBatch:
    return enriched(batch)

app.run()

When paired with a downstream consumer that uses isolation.level=read_committed:

Every output message is visible exactly once downstream.
Input offsets advance atomically with the output writes — no acknowledgement of a message whose output didn't commit.
A crash between batches doesn't produce duplicates: the next instance fences the previous transactional id at startup.

The guarantee is outputs-never-twice, not "state is always perfectly fresh". On a hard crash, your local state (windowed accumulators, key-keyed values) may briefly lag the output topic — see Crash recovery for the full shape and the knobs that tune it.

Why `app_id` is mandatory¶

Kafka EOS relies on each producer using a stable identifier — the transactional id — that the broker tracks across restarts. When a new instance starts and re-registers under the same id, the broker fences the previous one: aborts any transaction it had open and refuses further writes from it. That fencing is what makes "no duplicates across restarts" actually hold.

Turbine derives one transactional id per (topic, subscription name, partition) from app_id. The subscription name is part of it on purpose: two subscriptions reading the same topic are independent consumers, and sharing an id would make them fence each other on boot so that neither made progress. It is also why the name must be stable across restarts — see Several subscriptions on one topic.

If Turbine picked the app_id itself (e.g., a random UUID at process start), every restart would land on a fresh id and an old, still-running instance — split-brain network partition, kill -9 before cleanup, a previous pod Kubernetes hasn't terminated yet — would not be fenced. Both instances could commit to the same output topic in parallel, which is the exact duplicate-write scenario EOS exists to prevent.

app_id must be:

Stable across restarts of the same logical deployment — that's what fences the previous instance.
Unique per logical pipeline — two apps sharing an app_id will fence each other and neither will make progress.

"orders-enrichment-prod" or "alerting-staging" are good values — they name the deployment, not the host or pod.

Mixing modes in one app¶

A single Turbine app can host both at-least-once and exactly-once subscribes. Setting processing_guarantee="exactly_once" on one subscription doesn't affect the others — they keep the cheaper non-transactional path. The only constraint is one-way: as soon as any subscribe is EOS, Turbine(app_id=...) becomes mandatory.

kafka = KafkaBroker(bootstrap="kafka:9092")
app = Turbine(kafka, app_id="my-pipeline-prod")

@app.subscribe(kafka.topic("billing"), output=kafka.topic("ledger"), processing_guarantee="exactly_once")
def post_ledger(batch): ...

@app.subscribe(kafka.topic("metrics"), output=kafka.topic("enriched-metrics"))  # at-least-once, no overhead
def annotate(batch): ...

Idempotency contract¶

The exactly-once guarantee covers the Kafka output topic only. If your handler also does external side effects — HTTP calls, writes to a database other than Kafka, log shipping, metric pushes — those can fire more than once when the framework reprocesses or recovers, including during a replay.

Make those effects idempotent (a dedup key, an upsert, an If-None-Match header, etc.), or accept that they may run multiple times. This is the same contract every streaming framework offers: the framework can only make its own write paths transactional.

If your handler is pure (Arrow transforms + state ops + return a batch for Turbine to produce), the outputs Turbine produces from a record are covered and you have nothing to do.

One residual: window closes after a hard crash¶

There is one output path where exactly-once does not hold today, and it is worth knowing before you rely on it.

Outputs that a record produces are safe: their transaction commits the input offsets alongside them, so a restart can tell what already landed. Outputs produced by a timer — a window closing, an early fire, a session ending — have no such anchor. Closing a window emits its panes and, in the same commit, retires the window's accumulated state and its timer. If the process dies between the two, the panes are committed and visible but the retirement is lost: on restart the timer is still there, still due, and the same panes are emitted a second time.

The window is narrow (milliseconds, between a broker commit and a local disk write) and it takes a hard kill — SIGKILL, an OOM kill, a power loss. A clean shutdown, a rolling restart or a rebalance never hits it.

What to do about it:

If your sink is idempotent, nothing. Panes carry a natural identity — window name, window_start_ms, and the group key — so an upsert or a dedupe on those collapses the duplicate.
If it is not, dedupe on that identity downstream, or treat windowed aggregates as at-least-once and size your alerting accordingly.

To see whether you were exposed, watch turbine_recovered_due_timers_total. It counts timers that were already due the moment a worker recovered them — the closes a restart fires immediately. It does not prove a duplicate happened (a clean stop during an open window produces the same reading); it tells you which restarts could have produced one. The boot log carries the same warning in words.

This is tracked as a known defect, not an accepted design: see the entry in docs/TODO.md. Record-driven outputs are unaffected.

Cost¶

EOS adds one Kafka transaction round-trip per batch plus a durable state flush, so the per-batch commit phase gets noticeably more expensive — on the order of +10–20 % on commit time alone. On end-to-end throughput the impact is much smaller because the commit phase is only a fraction of the batch loop on realistic workloads; expect single-digit percent overhead at typical batch sizes (a few thousand records or more), and often within run-to-run noise.

Rule of thumb: at moderate-to-large batches, EOS is essentially free. Very small batches (a few hundred records) or near-zero handler latency may surface a higher overhead ratio because the fixed per-batch cost amortises over fewer messages — increase batch_size / batch_timeout_ms if you observe this.

The bench numbers behind these statements live in docs/internal/exactly_once.md; they are workload-specific and not a contract.

What gets committed atomically¶

Each batch is wrapped in a Kafka transaction that covers both the output messages and the input offsets. The transaction commits as one atomic unit, then the local state is durably persisted. On any error inside the batch, the transaction is aborted: no output is visible to read_committed, the input offset is not advanced, and the batch is retried from the last committed offset on the next run.

"Output messages" means every topic the batch writes to on the same cluster, not just output=:

dead-letter records (error handling),
late rows routed to late_data= (windowing),
everything written to a side_outputs= destination with ctx.emit, and every row a branch= arm claims (side outputs).

All of them ride the same transaction as the batch that produced them. There is no window in which a record is diverted but its offset is not committed, or the reverse — and no window in which one branch of a fan-out is visible and another is not.

This costs nothing extra, and the reason is worth knowing before you design around it: a Kafka transaction already spans every produce in the batch, so adding destinations adds produces, not commit points. It holds precisely because they are all topics on one cluster. A fan-out that mixed Kafka with a second system would have two independent commit points and therefore no joint guarantee, which is why Turbine does not offer that combination under exactly_once rather than offering it with a footnote.

Limitations¶

Single Kafka source. EOS across multiple brokers (e.g. a join over two clusters) is out of scope. Same for non-Kafka sinks — the transaction is Kafka-only.
EOS requires output=, even when other destinations are declared. A subscribe with no output is rejected at decoration time. For a true sink that is the right answer — there is nothing to make transactional. But the check is on output= specifically, so a subscription that writes only to side_outputs= or to branch= arms is refused too, even though those writes would ride a transaction perfectly well. If you want EOS on such a pipeline today, give it an output= (with branch=, that is also the default arm and costs you nothing but a topic). This is a limitation of where the check sits, not a property of the transaction.
State durability is RocksDB + object-store snapshots, not Kafka changelog. This is what makes the state gap exist (the changelog approach folds state into the same TX and removes the gap entirely). A Kafka-changelog state backend is on the roadmap as an optional alternative for workloads that need zero gap by construction; until it lands, on_crash_recovery="replay" is the user-facing way to get the same outcome at the cost of replay time.

Choosing between the two¶

Workload	Recommended mode
Idempotent downstream (upserts, dedup keys, append-with-PK)	At-least-once — simpler, free.
Additive aggregates — counts, sums, rates, anything billed	Exactly-once. The duplicate rate goes straight into the answer.
Distribution aggregates over large populations — percentiles, min/max, extremes	Judge by the aggregate, not by the word "analytics" — see below. Often at-least-once.
Stateful enrichment writing to a non-idempotent sink	Exactly-once.
Mission-critical state (financial, security) where staleness is unacceptable	Exactly-once with `on_crash_recovery="replay"`.
Sink (no `output`)	At-least-once (only option).

Not every aggregate is equally sensitive to a duplicate¶

"Analytics" is not a useful criterion. The question that decides the mode is narrower: if this event were observed twice, how much would my answer move? Three regimes, and the aggregate you picked tells you which one you are in.

Unaffected. Max, Min, First, Last, ArgMax, ArgMin, AnyTrue, AllTrue. Observing a value you already have changes nothing at all — these are idempotent under repetition, so at-least-once costs you exactly zero accuracy.
Attenuated. TDigestQuantile / TDigestMedian / TDigestQuantiles, Mean, Variance, StdDev, Correlation. A repeated observation re-weights a value that is already in the distribution; the error scales with the duplicate rate over the population size. Monitoring latency percentiles over millions of events per window is the canonical case: a handful of re-processed records after a rare crash moves a p95 by far less than the sampling noise you already accept. Paying for exactly-once there buys precision you cannot measure.
Directly biased. Count, Sum, and everything derived from them — rates, totals, invoiced volume. A duplicate is added in full, so the error is the duplicate rate. This is where exactly-once earns its cost.

The population size matters as much as the function. The same TDigestQuantile over a window holding a handful of rows is not in the attenuated regime — with few observations, one duplicate is a large fraction of the sample.

Duplicated events and duplicated panes are different problems¶

Worth separating, because the remedies are not the same:

A duplicated event (what at-least-once allows) is absorbed into the aggregate. The pane's value is already wrong when it is emitted, and no amount of downstream deduplication recovers it. Only the mode choice above protects you.
A duplicated pane (the window-close residual) carries a correct value, emitted twice. Deduplicating downstream on the pane identity fixes it completely — which is why an idempotent sink makes that residual a non-issue regardless of which aggregate you use.

The internal design document — producer construction details, recovery flow, the changelog-backend plan — lives at docs/internal/exactly_once.md.