Error Handling¶

A poison pill is a message Turbine can't decode — malformed JSON, a truncated Avro frame, binary garbage where a record was expected (usually a non-conforming producer). Left unhandled, one bad message at a fixed offset would block its partition forever: the worker can't get past it, so nothing after it is ever processed.

The on_error knob on the input topic (kafka.topic(..., on_error=...)) decides what happens. By default it governs both an undecodable message and a batch whose handler raises; a subscription that wants a different answer for the second sets on_processing_error:

`on_error`	Behaviour on a bad message / a failing batch
`"fail"` (default)	Stop that partition and keep the rest of the app running.
`"skip"`	Drop it, count it, and keep going.
`"dlq"`	Route the raw message(s) to a dead-letter topic, then keep going.

# Default: halt on the first message we can't decode.
@app.subscribe(kafka.topic("events"), output=kafka.topic("enriched"))
def enrich(batch): ...

# Tolerant: drop undecodable messages and keep the partition moving.
@app.subscribe(kafka.topic("events", on_error="skip"), output=kafka.topic("enriched"))
def enrich(batch): ...

# Dead-letter: keep the raw bad message for inspection / replay.
@app.subscribe(kafka.topic("events", on_error="dlq", dlq="events-dlq"), output=kafka.topic("enriched"))
def enrich(batch): ...

There are three distinct error surfaces, one knob each:

Decode (bad bytes — a message that won't parse): on_error on the input topic, below.
Processing (the handler raises an exception on an otherwise-decoded batch): on_processing_error on @subscribe, which defaults to on_error — see Processing errors.
Compute (a kernel in your aggregation expressions fails on bad data, e.g. a divide-by-zero): on_compute_error, also on @subscribe.

The first is a property of the stream, which is why it lives on the topic; the other two are properties of this subscription's code.

For the duplicate/ordering guarantees of the messages Turbine can process, see Delivery Guarantees.

Turbine can also discard rows that are perfectly valid, because you asked it to — those are not errors and on_error has no say in them. If you are here to work out where your data went, read Rows dropped on purpose at the bottom too.

`on_error="fail"` (default) — halt that partition¶

A message that can't be decoded stops the partition it arrived on, and nothing else. Concretely:

that partition's worker stops for good — it is not restarted, so it never crash-loops on the same record;
every other partition and every other subscription keeps consuming;
the process keeps running, and in a cluster the node stays a full member (it does not leave and rejoin);
the offset of the bad message is not committed, so nothing is skipped and nothing after it on that partition is processed.

"Never silently drop data; stop and let a human decide" — but the blast radius is one partition, not the whole deployment.

The halt is loud. Turbine logs the offending record's topic, partition, and offset at ERROR, and it is visible in three machine-readable places:

GET /health (and /ready) reports "status": "degraded" with "reason": "halted_workers", and lists each halted partition with the error text — including the offset. One request tells you what is down and why;
the /events feed carries a worker_fatal entry;
metrics: turbine_halted_workers (gauge — alert on > 0) and turbine_worker_fatal_total.

curl -s localhost:8400/health
{"status":"degraded","reason":"halted_workers","workers":11,
 "halted_workers":[{"worker_id":"work-00-p03","topic":"events","partition":3,
   "error":"deserialize failed (on_error=\"fail\") at topic=events partition=3 offset=48306635: ...",
   "since_ms":1753900000000}]}

Restarting the app does not fix it: the same record is still there, at the same offset, so the partition halts again — the process comes back up and keeps serving its other partitions, but that one stays down until you act. To get unstuck: switch the subscription to on_error="skip" or on_error="dlq" and restart, or advance the committed offset past the bad record by other means. Fixing the upstream producer alone is not enough — the bad message is already in the topic.

Use fail for pipelines where a message you can't process signals a real problem you want to know about immediately (financial, audit, correctness- critical), and where processing less is better than processing wrong.

Why the failure is contained rather than fatal to the process

Escalating one bad record to a process exit made it far more expensive than the data it protected: a supervisor (restart: unless-stopped, systemd, Kubernetes) restarts the process, it re-reads the same offset and exits again, forever — and in a cluster each cycle takes the node out of the membership and back in, disturbing partitions that had nothing to do with the bad record. Containment keeps the failure the size of the problem and makes it visible where operators already look.

`on_error="skip"` — drop and continue¶

The bad message is dropped, the offset advances past it, and the surviving messages in the same batch are processed normally. Dropping is never silent — every drop produces two signals:

a count on the turbine_dlq_messages_total metric (labelled by topic, partition, and phase) — always on, never throttled, so it's the signal to alert on;
a WARN log with the offending offset, the decode error, and a truncated sample of the payload — throttled to at most one every few seconds per worker, so a misbehaving producer flooding bad data can't flood your logs.

(So yes: skip still logs. It's a WARN, not an ERROR — dropping is a tolerated, opted-into outcome, not a failure; on_error="fail" is the one that logs an ERROR and stops.)

Use skip for analytics-tolerant pipelines where a few unparsable messages out of millions shouldn't take the whole job down.

`on_error="dlq"` — route to a dead-letter topic¶

Like skip (the partition keeps moving, every record is counted on turbine_dlq_messages_total and a throttled WARN is logged), but instead of dropping the bad message it is routed to the DLQ topic (dlq=) you configure, so you can inspect it and replay it later:

@app.subscribe(kafka.topic("events", on_error="dlq", dlq="events-dlq"), output=kafka.topic("enriched"))
def enrich(batch): ...

Each dead-lettered record keeps:

the original Kafka key, and the raw payload as the message body — untouched, so you can replay it by producing it straight back to the source topic;
headers carrying the context: turbine_dlq_source_topic, turbine_dlq_source_partition, turbine_dlq_source_offset, turbine_dlq_error (the decode error), turbine_dlq_phase, and turbine_dlq_worker. They show up directly in rpk topic consume and the Redpanda console.

No loss: the dead-letter write is flushed before the input offset commits, so a crash never advances past a record that wasn't dead-lettered. Under processing_guarantee="exactly_once" the write rides the same transaction as the offset commit (no duplicates either); under at-least-once a crash may re-deliver a DLQ record, deduplicable via the turbine_dlq_source_offset header.

The dlq topic is created on first write if your broker allows topic auto-creation; otherwise pre-create it. dlq is required when on_error="dlq" and rejected otherwise. Both live on the input topic (kafka.topic(..., on_error="dlq", dlq="events-dlq")).

Use dlq when you need the bad records back — debugging a flaky producer, audit/compliance, or replay after a fix — rather than just a count.

Processing errors¶

The sections above are about a message Turbine can't decode. This section is about the next stage: your handler raising an exception on a batch that decoded fine (a bug, a transient dependency, a KeyError on an unexpected shape). Left unhandled, that exception used to restart the worker at the same offset — a crash-loop on a deterministic error.

By default this phase follows on_error, and on_processing_error= on @subscribe overrides it when the two failures deserve different answers.

A handler runs over a whole batch at once (one vectorised call, never per-message), so a processing error isn't tied to a single row. The policy therefore acts at batch granularity:

policy	Behaviour when the handler raises
`"fail"` (default)	Stop that partition (same containment as above), naming the batch's offset range and the handler error.
`"skip"`	Drop the whole batch, advance past it, count it on `turbine_dlq_messages_total{phase="processing"}`, log a throttled `WARN`.
`"dlq"`	Route every raw input payload in the batch to the `dlq` topic (headers carry `turbine_dlq_phase=processing` + the exception), then advance.

Under skip / dlq the batch's durable (RocksDB) state writes are rolled back before the offset advances, so a half-applied handler can't leave inconsistent state behind. (In-memory windows keep their state in Python and are best-effort here, matching their no-durability contract.) The no-loss and exactly-once couplings are the same as the decode DLQ: the dead-letter write is flushed before the offset commits, and rides the transaction under processing_guarantee="exactly_once".

fail is the default because a processing error often signals a real bug you want to halt on rather than silently skip; switch to skip/dlq for pipelines where one bad batch shouldn't stop a partition. Note that transient errors are not retried (a retry/backoff policy is separate and not yet available) — fail halts, skip/dlq drop.

Giving the two phases different answers¶

An undecodable payload is bad data: routing it to the dead-letter topic is exactly right, and the partition should keep going. A handler that raises is usually a bug — and answering it with the same dlq verdict throws away rows that were fine and advances the offset past them. Those rows are no longer in your aggregate, and nothing puts them back short of replaying the dead-letter topic yourself.

on_processing_error= on @subscribe decouples the two:

@app.subscribe(
    kafka.topic("events", schema=Event, on_error="dlq", dlq="events-dlq"),
    output=kafka.topic("scores"),
    on_processing_error="fail",
)
def score(batch, state): ...

A payload that won't decode still goes to events-dlq; a bug in score stops that partition only — the offset does not advance and the state is untouched, so a fix and a restart reprocess the batch instead of leaving a hole in the results. The other partitions and the other subscriptions keep running.

It is declared on the subscription rather than on the topic because a handler exception belongs to the handler: two subscriptions can read the same bytes and only one of them have a bug. Left unset it inherits on_error, so nothing changes for a pipeline that doesn't use it. on_processing_error="dlq" needs the input topic's dlq= — the dead-letter topic is shared by both phases.

Knowing when a partition has stopped producing results¶

skip and dlq keep the partition running, which is the point — and it means a handler that fails on every batch is easy to miss. No partition is halted, the offsets keep advancing, and the node looks busy while everything it reads is dropped or dead-lettered.

Two signals, answering two different questions:

"did batches fail?" — turbine_dlq_messages_total{phase="processing"}, the counter. Good for a rate alert; it stays high forever after an incident, so it can't tell you whether things are broken now.
"is this partition still failing, and since when?" — after a run of consecutive failing batches, /health reports "status": "degraded" with "reason": "processing_errors" and a processing_errors list naming the worker, the topic, the partition, how long it has been failing, where the rows went (skip or dlq) and the exception. The turbine_processing_errors_consecutive gauge carries the same run length, including before it crosses the threshold.

The state clears itself on the first batch that goes through — a bad minute reads as a bad minute, not as a flag stuck on until the next restart. /ready stays 200: the node is still serving its other partitions, and taking it out of rotation would not fix a handler that raises.

Compute errors¶

The previous section is about bad bytes — a message Turbine can't decode at all. This section is about bad data: the message decoded fine, but a computation in your aggregation expressions fails on some rows. The classic cases are an integer divide-by-zero, a strict cast overflow, and a parse_ts that can't parse a string. In PyArrow these raise on the whole batch, so one bad row would otherwise sink the thousand good rows next to it.

The on_compute_error knob on @app.subscribe(...) decides what happens:

`on_compute_error`	Behaviour on a failing expression kernel
`"null"` (default)	Null just the offending rows, count them, and process the rest of the batch.
`"fail"`	Re-raise — stop the batch (and the worker), for correctness-critical pipelines.

from turbine import functions as e

# Default: a zero divisor nulls that row's score; the batch keeps going.
@app.subscribe(kafka.topic("events"), output=kafka.topic("scores"))
class Score:
    def __init__(self):
        self.win = PersistentTumbling(
            size="1m",
            value=agg.Mean(e.col("errors") / e.col("requests")),  # /0 → null
        )

`"null"` (default) — null the bad rows¶

This is the SQL-faithful default: the same model as streaming SQL engines (RisingWave, Flink), where a row that fails a computation becomes null and a null drops out of skip-null aggregates (sum, mean, p95, …). One divide-by-zero in a window therefore lowers that window's sample count by one instead of crashing the job. The rest of the batch — and every other row in the window — is unaffected.

Nulling is never silent. Every nulled row is counted on the always-on turbine_compute_errors_total{op} metric, where op is the kernel that failed (divide, cast, parse_ts, …) — a bounded label, never the row content, so it's safe to alert on. A spike there is your signal that upstream data quality slipped, even though the pipeline kept running.

Two variants give you the same permissive outcome explicitly and faster (a vectorised masked kernel, no row-by-row fallback), independent of the policy: divide_or_null / divide_or_zero and cast_or_null / cast_or_default. Use those when you expect zeros/overflows as normal data; let the default policy catch the ones you didn't.

The policy only catches data errors (PyArrow ArrowInvalid). A programming mistake — referencing a column that doesn't exist — still raises loudly under either policy; it's a bug, not bad data.

`"fail"` — stop instead¶

Set on_compute_error="fail" when a computation that can't be evaluated signals a problem you'd rather halt on than paper over with nulls (financial, audit, correctness-critical). The kernel re-raises, stopping the batch. This is the strict, fail-fast counterpart to on_error="fail" on the decode side.

Rows dropped on purpose¶

Everything above is about data Turbine couldn't handle. Two settings make it discard data it handled fine, because you told it to. They are configured elsewhere and are not errors — but they are the other half of the answer to "where did my rows go", so they belong on the same page as the counters you would alert on.

What	Set by	Counter
A row arriving after its event-time window already closed	`on_late="drop"` (the default)	`turbine_late_rows_dropped_total{worker, window}`
A row matching no arm of a declared split	`on_unmatched="drop"`	`turbine_branch_rows_dropped_total{worker}`

Both have a keep-it variant, and in both cases the counter is the point: a drop you configured is still a drop, and it is never silent.

Late rows. on_late="route" publishes them to a late_data= topic instead of discarding them — nothing is lost and no already-published window result is contradicted. See Windowing. Note that turbine_late_rows_dropped_total is the default behaviour's counter, so a pipeline nobody configured is already reporting on this series.
Unmatched rows. Only a subscription that declares branch= without an output= can drop rows this way, and Turbine refuses to start unless you write on_unmatched="drop" explicitly. With an output=, unmatched rows go there instead and the counter stays at zero. See Side outputs.

The distinction that matters for alerting: turbine_dlq_messages_total means Turbine could not process this, while these two mean Turbine did what you configured. A spike in the first is a data-quality or code problem; a non-zero second is a policy you chose, and worth watching only against the rate you expected.

Error Handling¶

on_error="fail" (default) — halt that partition¶

on_error="skip" — drop and continue¶

on_error="dlq" — route to a dead-letter topic¶