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 under exactly-once, the worker first replays the input its local state missed — 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 crash-recovery 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.

Timer output after a hard crash¶

Exactly-once subscriptions recover committed window closes, early fires and session closes without emitting them again. Turbine preserves the pending state changes durably before committing their output. After a restart, it confirms the commit with the sink and finishes the state changes before running timers. An uncommitted timer transaction can instead be retried normally.

This requires the matching local state/checkpoint and the subscription's broker commit history. If an older checkpoint lacks the changes for an already committed timer, or the broker's commit information has expired or been overwritten, the worker stops with a recovery error. on_crash_recovery="accept" and "replay" do not override this protection. Restore matching state and preserve the consumer group's offsets; deleting local state alone is not a repair.

The broker commit history lives in the consumer group's committed-offset metadata, which brokers expire after a retention — offsets.retention.minutes on Kafka, group_offset_retention_sec on Redpanda, both 7 days by default — counted from the last commit. While the app is running this cannot happen: every batch refreshes it, and partitions quiet enough to go a day without a commit re-commit the same receipt in an empty transaction (turbine_eos_receipt_keepalive_total counts these; the cadence defaults to 24 h and can be lowered with TURBINE_EOS_RECEIPT_KEEPALIVE_SECS for a shorter broker retention). What the keepalive cannot cover is downtime: configure the retention to exceed the longest stop you expect, with margin. If the history does expire while the app is down, the next start stops with the recovery error above rather than inventing a new one.

The other stop worth knowing about: a partition whose newest usable state — local or checkpoint — predates its last committed timer batches (losing a node and its disk is the canonical case) halts at recovery, because the broker receipt proves output that state does not contain. The safe repair is a matching checkpoint, and recovery always looks for one first: even with the override below enabled, a local or remote state that matches the receipt is selected strictly, and a clean handoff never re-enrolls.

When nothing matches — no checkpoint anywhere, or all of them stale — an operator can deliberately accept the pre-receipt behaviour (the timers covered by the gap fire again and may duplicate output) by setting TURBINE_EOS_RECEIPT_REENROLL=all, or a comma-separated list of topic:partition entries, before starting the app. If local state and checkpoints are all gone, the override also covers that: the partition adopts the receipt on empty state and resumes under on_crash_recovery ("replay", the default, rebuilds state from the retained input — the windows it closes again may be emitted a second time, the duplicate risk above; "accept" resumes at the broker position with an empty state). Each adoption is a loud WARN and increments turbine_eos_receipt_reenrolled_total{topic,partition}; unset the variable once the app runs again — leaving all in a deployment manifest would turn every later clean handoff into a duplicate risk, which is exactly what the strict-first selection protects against but should not have to. An ambiguous commit outcome (a pending intent the receipt can neither confirm nor deny) is never overridden this way.

Upgrading an existing exactly-once deployment does not require a new app_id: an empty local state enrolls the existing consumer group in place, at its already-committed offset — provided no pre-upgrade checkpoint remains in the checkpoint store (a legacy checkpoint would be selected and then rejected on format, halting the partition; use a fresh checkpoint location as well, as the upgrade notes prescribe). What cannot be enrolled is a nonempty state written before timer commit recovery was introduced — its history could contradict output the group already committed. A deliberate fresh run needs fresh state and a fresh app_id; that is a new processing history, not a migration of previous output.

One consequence of in-place enrollment: the empty state starts behind the group's committed offset, so the first boot meets a state gap covering the group's whole history, and on_crash_recovery decides. Under the default "replay", the worker rebuilds the state from the beginning of the topic before producing again, and refuses to start if retention no longer holds the beginning. To resume at the committed offset with an empty state instead, set on_crash_recovery="accept" for that first start.

turbine_recovered_due_timers_total counts timers due after recovery. Its meaning differs by mode: under at-least-once, those timers replay and may re-emit output, so a dashboard can legitimately alert on a spike; under exactly-once they have passed commit reconciliation and fire without re-emitting, so the same spike is normal recovery — alert on it only for at-least-once subscriptions.

Cost¶

Record batches under exactly-once pay for the Kafka transaction and durable state write: roughly +10–20 % on commit time, often within noise on end-to-end throughput, and essentially free at moderate batch sizes. Timer batches (window closes, early fires) keep that same single disk synchronization per batch, but write about 2.5× more bytes to local state, because the pending changes are journaled before the output commit. That is paid per timer batch, not per individual timer.

Measure ingestion and timer bursts separately on your storage and broker, including tail latency. At-least-once subscriptions and stateless subscriptions pay none of this.

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. Before commit, failures abort the transaction and permit retry. A timeout during commit can have an uncertain outcome; a timer batch is then reconciled on restart before any retry. The worker never assumes that a missing commit response proves an abort.

"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, the default on_crash_recovery="replay" gets 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, keeping its default crash recovery, which rebuilds the state.
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 carries the same aggregate result twice. Exactly-once timer commit recovery prevents this crash gap. At-least-once users can deduplicate pane deliveries downstream; this does not repair an aggregate already distorted by duplicated input events.