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¶

app = Turbine(
    brokers="localhost:9092",
    state_dir="/tmp/turbine",
    state_format="json",
    checkpoint_url=None,
    app_id=None,
    # ...
)

Broker & data¶

Parameter	Type	Default	Description
`brokers`	`str`	`"localhost:9092"`	Kafka bootstrap servers (comma-separated `host:port` list).
`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. Accepts `file://...`, `s3://...`, `memory://...`. Defaults to `file://{state_dir}/snapshots` at run time.

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.

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 handler/processing errors; this governs computation inside aggregation expressions over decoded rows. See Error handling.

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 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). `"fail"` (default): stop the app cleanly and exit non-zero, naming the offending topic/partition/offset (decode) or offset range (processing). `"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"`, rejected otherwise. 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_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).