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 — it governs both an undecodable message and a batch whose handler raises (see Processing errors below):

`on_error`	Behaviour on a bad message / a failing batch
`"fail"` (default)	Stop the whole app cleanly and exit non-zero.
`"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, and two knobs:

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

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

`on_error="fail"` (default) — fail fast¶

A message that can't be decoded stops the app. Turbine logs the offending record's topic, partition, and offset (so you can find it), and app.run() raises — the process exits with a non-zero status, distinct from a clean shutdown (Ctrl-C / SIGTERM / stop_event, which exit 0). An orchestrator (Kubernetes, systemd, a job runner) sees the failure.

The offset of the bad message is not committed, so restarting without changing anything re-hits the same message and halts again — by design. This is fail-fast: "never silently drop data; stop and let a human decide." To get unstuck, switch the subscription to on_error="skip" and restart (the error message tells you so), 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).

`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. The same on_error knob also governs 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.

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:

`on_error`	Behaviour when the handler raises
`"fail"` (default)	Stop the app cleanly, exit non-zero, 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 take the job down. Note that transient errors are not retried (a retry/backoff policy is separate and not yet available) — fail halts, skip/dlq drop.

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.

Error Handling¶

on_error="fail" (default) — fail fast¶

on_error="skip" — drop and continue¶

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