Configuration¶

Reference for every parameter on Turbine(...) and @app.subscribe(...). For end-to-end usage examples, see Quick Start; for operational concerns (cluster mode, perf tuning, rolling upgrades) see Deployment.

`Turbine(...)` constructor¶

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

app = Turbine(
    kafka,
    state_dir="/tmp/turbine",
    state_format="json",
    checkpoint_url=None,
    app_id=None,
    # ...
)

Broker & data¶

Parameter	Type	Default	Description
`broker`	`KafkaBroker`	`KafkaBroker()` (localhost:9092)	The Kafka cluster handle. The same object builds every `input=`/`output=` topic reference, so the address is written once — see Partitioning. A topic built from a different handle is refused when the subscription is declared.
`from_earliest`	`bool`	`False`	Start consuming from the earliest offset when no committed offset exists. Mutually exclusive with `from_latest`.
`from_latest`	`bool`	`False`	Start consuming from the latest offset when no committed offset exists. Mutually exclusive with `from_earliest`.
`max_lag_seconds`	`int \\| None`	`None`	Hard cap on consumer lag (seconds). When set, the worker skips ahead if it falls further behind than this. Also configurable via `TURBINE_MAX_LAG_SECONDS`.

State & checkpointing¶

Parameter	Type	Default	Description
`state_dir`	`str`	`"/tmp/turbine"`	Local directory for the RocksDB state store. One subdirectory per partition.
`state_format`	`str`	`"json"`	Serialisation codec for values written through `state.get` / `state.put`. Accepts `"json"` or `"msgpack"`. `get_bytes` / `put_bytes` bypass this entirely.
`checkpoint_url`	`str \\| None`	`None`	Snapshot destination — where state is saved so another node (or this one, after a restart) can restore it. Accepts `file://...`, `s3://...`, `memory://...`. Defaults to `file://{state_dir}/snapshots` at run time, which is local to the node: a stateful app in cluster mode refuses to start on it, since every partition handoff would lose state. S3-compatible endpoints are configured with the standard `AWS_*` variables — see Deployment → Checkpoint storage.

Exactly-once¶

Parameter	Type	Default	Description
`app_id`	`str \\| None`	`None`	Stable identifier for this deployment, used to fence previous instances on restart. Required as soon as any subscribe uses `processing_guarantee="exactly_once"`. See Delivery Guarantees.

REST API & cluster¶

Parameter	Type	Default	Description
`api_port`	`int \\| None`	`8400`	Port for the REST/console API. Env var: `TURBINE_API_PORT`.
`raft_node_id`	`int \\| None`	`None`	Node id within the Raft cluster. Env: `TURBINE_RAFT_NODE_ID`.
`raft_listen_addr`	`str \\| None`	`None`	Address this node listens on for Raft RPC (`http://host:port`). Env: `TURBINE_RAFT_LISTEN_ADDR`.
`raft_peers`	`list[tuple[int, str]] \\| None`	`None`	Bootstrap mode peer list. Mutually exclusive with `raft_seed`. Env: `TURBINE_RAFT_PEERS` (format `"1=http://a:8400,2=http://b:8400"`).
`raft_seed`	`str \\| None`	`None`	Join mode — address of an existing node. Mutually exclusive with `raft_peers`. Env: `TURBINE_RAFT_SEED`.

Standalone (single-process) deployment leaves all raft_* parameters None. See Deployment → Raft cluster for the bootstrap / join workflow.

`@app.subscribe(...)` decorator¶

@app.subscribe(
    kafka.topic("input-topic"),
    output=kafka.topic("output-topic"),
    batch_size=1000,
    batch_timeout_ms=250,
    partition_key="user_id",
    processing_guarantee="at_least_once",
)
def handler(batch, state): ...

Topic & batching¶

Parameter	Type	Default	Description
`input`	`kafka.topic(...)`	(required, positional)	Kafka topic to consume, as a broker topic reference (e.g. `kafka.topic("orders")`). Bare strings are rejected. How the input is read (schema, event time, decode-error policy) is configured on this reference — see Input topic — `kafka.topic(...)`.
`name`	`str \\| None`	(handler name)	Identifies the subscription. Must be unique per topic and stable across restarts — see Several subscriptions on one topic.
`output`	`kafka.topic(...) \\| None`	`None`	Output topic, as a broker topic reference (e.g. `kafka.topic("scores")`). Omit for a sink (the handler must return `None`).
`batch_size`	`int`	`1000`	Maximum records assembled before the handler is invoked.
`batch_timeout_ms`	`int`	`250`	Maximum wait (ms) before flushing a partial batch when traffic is light.

Several subscriptions on one topic¶

More than one handler may subscribe to the same topic. Each is an independent consumer: it sees every record of the topic (the partitions are not shared out between them), keeps its own state and offsets, and has its own output and processing_guarantee. One subscription lagging, failing or restarting does not hold up the others.

@app.subscribe(kafka.topic("events"), output=kafka.topic("scores"))
def score_users(batch): ...

@app.subscribe(kafka.topic("events"), output=kafka.topic("audit"))     # same input, own output
def audit_trail(batch): ...

Use this when two concerns happen to share an input but have nothing else in common. The cost is that the topic is read and decoded once per subscription; if you only want to split one stream by a predicate, a single handler that returns different rows is cheaper.

name identifies the subscription and defaults to the handler's name. It must be unique per topic — Turbine refuses to start otherwise — and stable across restarts, because it anchors where Turbine records that subscription's progress. Renaming the handler therefore makes the subscription resume from auto.offset.reset instead of where it left off; pass an explicit name="…" to keep the identity pinned while the function name changes.

Cluster mode

Multiple subscriptions on one topic are single-node only for now. An app configured for the Raft cluster refuses to start with more than one subscription per topic. Run those subscriptions as separate Turbine apps with distinct app_ids instead.

Decoding, event time & decode errors¶

How to read the input — the schema, event-time source, and decode-error policy — is configured on the input topic reference itself, not on subscribe. Pass these to kafka.topic("name", ...); see Input topic — kafka.topic(...) below for the full list.

Partitioning & parallelism¶

Parameter	Type	Default	Description
`partition_key`	`str \\| None`	(unset)	Name of the column that carries the partitioning key. Required for rescaling (stateful subscriptions without it cannot be rebalanced cleanly) and for `parallelism > 1`. Pass `None` explicitly to silence the rescale-readiness warning when running unkeyed on purpose.
`parallelism`	`int`	`1`	Number of in-process shards per Kafka partition. Records sharing the same `partition_key` value always land on the same shard, so each shard owns a disjoint slice of the keyspace and runs in parallel with the others. Requires a class-based handler (the runtime needs one independent instance per shard).

See Partitioning for the divisor-of-960 constraint on Kafka topic partition counts and the rationale behind sub-partition parallelism.

State lifetime¶

Parameter	Type	Default	Description
`ref_ttl_ms`	`int \\| None`	`None`	Lifetime, in milliseconds of event time, of the reference values a window publishes for downstream `Ref(...)` readers. Without it they are kept forever: each close overwrites the group's value, so a group that stops appearing leaves its last one behind indefinitely. Does not touch the window's own state.
`state_ttl`	`dict[str, int] \\| None`	`None`	Expire state keys by age. Each entry maps a state-key prefix to a lifetime in milliseconds of event time: every key starting with that prefix disappears once the stream's watermark has moved that far past the key's last write. Without it, nothing you write to state is ever removed.

A stateful handler that keys on something unbounded — a user id, a request id, a device id — grows its state store forever: every key ever seen is still there. state_ttl is how you bound it:

@app.subscribe(
    kafka.topic("events", event_time="ts"),
    state_ttl={"last_seen": 24 * 60 * 60 * 1000},   # one day
)
class Sessions:
    def process(self, batch, state):
        state.put(f"last_seen|{user}", ...)   # covered: the key starts with "last_seen"
        state.put(f"totals|{region}", ...)    # not covered: kept forever

What you get, and what you give up:

The clock is your data's, not the machine's. The TTL is measured in event time, so replaying a day of history expires exactly what the live run expired. A wall-clock TTL would not: a backfill would expire nothing, and an app stopped overnight would expire a pile of keys on restart. This is why an event-time axis is required — event_time= or with_kafka_timestamp=True on the input topic. Declaring state_ttl without one is refused at startup rather than silently doing nothing.
The clock keeps running while a partition is quiet. Otherwise a topic that goes idle overnight would never age anything out. A key can therefore expire during a lull, with no new record involved.
Age is measured from the last write, not from creation. Re-writing a key refreshes it, so the TTL is a window of inactivity.
Expiry is by prefix, and prefixes are yours to choose. Only keys you named are touched; everything else is kept.
An expired key reads back as missing (state.get returns None) from the moment it ages out — you never see a stale value, whether or not the space has been reclaimed yet.
Reclaiming the disk is best-effort, and deliberately so. Expired keys are removed while the state store reorganises itself, which it does continuously as you write — so the busier the app, the faster the space comes back. What you get is a store that stays bounded: it settles around the size of the keys still alive instead of growing forever. What you do not get is a promise about when a particular key's bytes are freed. Since expired keys are already invisible to reads, this only ever affects disk usage.
Windows are not affected. Window state already disappears when its window closes, so state_ttl deliberately does not apply to it — its lifetime is derived, not declared. The one thing a window publishes that does outlive its window is a reference value (what a downstream window reads through Ref(...)): its key holds the latest value per group with no expiry of its own, so a group that stops appearing leaves its value behind for good. Bound it with ref_ttl_ms= on the subscription — how long a reference stays meaningful is a judgement call, not something Turbine can derive.

Two helpers are built for exactly this kind of state:

seen = agg.Dedup(state, prefix="seen")          # check(id) -> already seen?
last = agg.LastSeen(state, prefix="last_seen")  # the latest value per key

Both key on something unbounded by nature, so both warn if the subscription declared no state_ttl covering their prefix — without one, nothing would ever remove what they write. - Expiry is counted, never silent: turbine_state_keys_expired_total for how many keys have been removed, and turbine_state_expiry_frontier_ms for how far expiry has advanced. See Exposed metrics.

Delivery guarantees¶

Parameter	Type	Default	Description
`processing_guarantee`	`str`	`"at_least_once"`	Producer semantics for this subscription. `"exactly_once"` opts into Kafka transactions and requires `output` plus `Turbine(app_id=...)`. See Delivery Guarantees.
`on_crash_recovery`	`str`	`"accept"`	Crash-recovery policy. `"accept"` (default) boots with the observed state gap; `"replay"` silently re-consumes the gap to rebuild state before resuming with outputs enabled. EOS-only — raises `ValueError` on at-least-once subscribes. See Crash recovery.
`halt_if_gap_exceeds`	`int \\| None`	`None`	Circuit-breaker: if the boot-time state gap exceeds this many events, the worker refuses to start with a fatal error. EOS-only. Pair with `on_crash_recovery` if you want both an automatic recovery and a hard cap on catastrophic gaps.

Error handling¶

The decode / processing error policy (on_error, dlq) is set on the input topic — see Input topic — kafka.topic(...). The on_compute_error policy below governs expression failures over already-decoded rows and stays a subscribe parameter.

Parameter	Type	Default	Description
`on_compute_error`	`str`	`"null"`	What to do when an expression kernel fails on bad data (integer divide-by-zero, strict `cast` overflow, `parse_ts` parse failure). `"null"` (default): null just the offending rows, count them on `turbine_compute_errors_total{op}`, and process the rest of the batch (SQL-faithful — a nulled row drops out of skip-null aggregates). `"fail"`: re-raise, for correctness-critical pipelines. Distinct from `on_error` (on the input topic), which governs undecodable messages, and from `on_processing_error`, which governs an exception raised by the handler; this governs computation inside aggregation expressions over decoded rows. See Error handling.
`on_processing_error`	`str \\| None`	`None`	What to do when this handler raises on a batch that decoded fine, when it should differ from the input topic's `on_error`. `None` (default) inherits it. Same three values (`"fail"` / `"skip"` / `"dlq"`); `"dlq"` needs the input topic's `dlq=`, which both phases share. Worth setting because the two failures are not the same one: bad bytes belong in the dead-letter topic, whereas answering a handler bug the same way discards rows that were fine and advances past them. `on_error="dlq", on_processing_error="fail"` keeps the poison-pill behaviour while a bug stops that partition only, leaving the offset and the state for a fix-and-restart. Declared here rather than on the topic because a handler exception belongs to the handler. See Error handling.

Extra destinations¶

A subscription may write to more than its output. These parameters declare where — declaration is what lets Turbine validate the destinations at startup, provision them, and draw them on the topology page. Full treatment in Side outputs.

Parameter	Type	Default	Description
`side_outputs`	`dict[str, kafka.topic(...)] \\| None`	`None`	Named destinations the handler writes with `ctx.emit(name, batch)`. A handler opts into the emission context by naming a parameter `ctx` (last). Emission is non-exclusive: the same rows may go to any number of destinations. An undeclared name raises immediately, naming the declared set; a declared destination nobody writes to is normal. A destination equal to the input (infinite loop) or to `output` (branches merging into the result stream) is refused at startup.
`branch`	`list[Branch] \\| None`	`None`	Declarative first-match-exclusive split of the batch the handler returns. Each `Branch(name, predicate, to=kafka.topic(...))` carries a `turbine.functions` expression; every row goes to the first arm whose predicate holds and to that arm only. A list, not a dict, because the order decides where a row lands. Rows matching no arm take `output` as the default arm — so adding `branch=` narrows what the output topic carries. Because the predicates are expressions rather than callables, they are shown on the topology page and served by `/topology`. Mutually exclusive with `combine_by`.
`on_unmatched`	`str \\| None`	`None`	What happens to rows matching no `branch` arm. Omit when `output` is set — declaring a default destination already answers it. Required as `"drop"` when there is no `output`, because otherwise those rows vanish silently; dropped rows are counted on `turbine_branch_rows_dropped_total{worker}`. `"drop"` together with `output` is refused (nothing would ever reach the output topic).
`on_late`	`str`	`"drop"`	What to do with a row whose event-time window has already closed (past `window_end + allowed_lateness_ms`). `"drop"` discards it and counts it on `turbine_late_rows_dropped_total{worker, window}`; `"route"` publishes it to `late_data` instead, so a straggler is never lost and never re-opens a window that was already emitted. See Windowing.
`late_data`	`kafka.topic(...) \\| None`	`None`	Destination for `on_late="route"`. Required when `on_late="route"`, rejected otherwise. The routed record is the row as the engine decoded it, plus `_late_window` / `_late_window_end_ms` / `_late_watermark_ms` describing which window rejected it and why. Refused if it is the input or the `output` topic.

Repartitioning & cross-key aggregation¶

Parameter	Type	Default	Description
`repartition_by`	`str \\| None`	`None`	Re-key the stream on a column the producer did not partition by, via an internal Kafka topic, before this subscription sees it. Both hops name their internal topic after the app, so they require an explicit `Turbine(app_id=...)` — checked at `run()`. See Repartitioning a stream.
`repartition_partitions`	`int \\| None`	`None`	Partition count of that internal topic. Defaults to the source topic's count. Only meaningful with `repartition_by`.
`combine_by`	`str \\| None`	`None`	Two-phase aggregation on a key that is not the source `partition_key`: the window pre-aggregates locally and ships accumulator partials to a synthesized second stage that merges them. Costs `O(unique keys)` on the wire instead of `O(events)`. Implies `partition_key`, and requires an explicit `Turbine(app_id=...)`. See Cross-key aggregation.
`combine_partitions`	`int \\| None`	`None`	Partition count of the combine internal topic — the second stage's parallelism axis. Defaults to the source topic's count. Only meaningful with `combine_by`.

Input topic — `kafka.topic(...)`¶

The first positional argument of @app.subscribe(...) is a broker topic reference. Beyond the topic name, it carries how to read the input: the decode schema, the event-time source, and the decode-error policy. These seven kwargs live here (not on subscribe), because they describe the input stream itself. The matching kafka.topic(...) used for output= accepts only the topic name and message_key= — these input-only properties are rejected there.

events = kafka.topic(
    "events",
    schema=Event,            # Pydantic model → schema-aware JSON decode
    event_time="event_ts",   # event-time column
    event_time_unit="ms",
    on_error="dlq",
    dlq="events-dlq",
)

@app.subscribe(events, output=kafka.topic("scores"), partition_key="user_id")
def score(batch): ...

Inline them when there are only one or two; bind the topic to a variable first (as above) when several add up.

Parameter	Type	Default	Description
`schema`	`type \\| None`	`None`	Pydantic `BaseModel` subclass — its fields are converted to a `pyarrow.Schema` for schema-aware JSON decoding (faster, no per-batch inference). Supported field types: `str`, `int`, `float`, `bool`, `bytes`, `Optional[X]`, and arrays `list[X]` / `tuple[X, ...]` (→ Arrow `list<X>`, nesting allowed).
`avro_schema`	`str \\| None`	`None`	Avro writer schema (JSON string). When set, payloads are decoded as Avro Single-Object Encoding frames. Wins over `schema` if both are set.
`event_time`	`str \\| None`	`None`	Dotted path to the record field carrying the event timestamp. Every window under the subscription then uses it as its time column — see Windowing → time model.
`event_time_unit`	`str`	`"ms"`	Unit of an integer/float `event_time` field (`"s"`, `"ms"`, `"us"`, `"ns"`). Ignored for `Timestamp` and RFC 3339 string fields.
`with_kafka_timestamp`	`bool`	`False`	Append an `Int64` column named `_kafka_ts_ms` to each batch, carrying the broker-side message timestamp. Use it as the `time_column=` of a window to switch from processing-time to event-time semantics.
`on_error`	`str`	`"fail"`	What to do with a message that can't be decoded (poison pill) or a batch whose handler raises (processing error — override that phase alone with `on_processing_error=` on `@subscribe`). `"fail"` (default): stop that partition — the rest of the app keeps running — naming the offending topic/partition/offset (decode) or offset range (processing); the halt shows up as `"status": "degraded"` on `/health`. `"skip"`: drop the bad message (decode) or the whole batch (processing), count it on `turbine_dlq_messages_total`, log a throttled sample, and keep going. `"dlq"`: route the raw payload(s) to the `dlq` topic (source metadata + `turbine_dlq_phase` in headers), then keep going. Decode failures are isolated per-row; a processing error is batch-level (one vectorised handler call). See Error handling.
`dlq`	`str \\| None`	`None`	Dead-letter topic for `on_error="dlq"`. Required when `on_error="dlq"` or `on_processing_error="dlq"`, rejected when neither asks for it. The original key + raw payload are preserved (replayable); source topic/partition/offset/error ride Kafka headers.

Environment variable summary¶

For deployments where parameters are injected by an orchestrator rather than hard-coded, the following constructor arguments have environment variable equivalents (constructor argument wins when both are set):

Constructor	Environment variable
`api_port`	`TURBINE_API_PORT`
`max_lag_seconds`	`TURBINE_MAX_LAG_SECONDS`
`raft_node_id`	`TURBINE_RAFT_NODE_ID`
`raft_listen_addr`	`TURBINE_RAFT_LISTEN_ADDR`
`raft_peers`	`TURBINE_RAFT_PEERS`
`raft_seed`	`TURBINE_RAFT_SEED`

A few EOS-specific tunables don't have a constructor equivalent — they live only as env vars because they're operational knobs, not API contracts:

Environment variable	Default	Description
`TURBINE_TTL_SWEEP_MS`	`0`	Milliseconds per polling cycle spent actively hunting for expired `state_ttl` keys, on top of the normal best-effort reclamation. `0` (default) leaves reclamation entirely to the storage engine, which is free — it drops expired keys while reorganising files it was rewriting anyway. Raise it only if you need disk back sooner than that and can afford the throughput: the active hunt has to read every key to find the expired ones, so it costs real time and buys nothing in correctness (an expired key is already invisible to reads).
`TURBINE_POINTER_WAIT_MS`	`2000`	Maximum time (ms) the cold-restart boot path waits for a Raft-replicated snapshot pointer to appear before falling back to listing the object store. Covers the cross-node replication latency on graceful handoffs in cluster mode. Set to `0` to disable the wait entirely (lower deployment latency, slightly larger state gap on cluster handoffs).