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`.

For a fresh start with from_latest, Turbine fixes the starting position during startup. Records arriving afterwards are still consumed, including records from a transaction that was open at that point and later commits. Resolving this position requires a broker request with a 10-second timeout. If it fails, the worker restarts and resolves a new starting position; the earlier attempt's boundary is not preserved. The same rules apply when a fresh stateless subscription starts at the topic head by default. The startup log line (Worker resume decision) reports the resolved position (next_offset), so the boundary each worker chose is visible in the logs.

If that position becomes unavailable before consumption because of retention, truncation or deletion and recreation of the topic, Turbine falls back to the earliest available offset. This can read the retained or recreated backlog even with from_latest; the option chooses a starting position, not a permanent policy of skipping to the head. Keep the source topic intact during startup when preserving that boundary matters.

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`.
`api_host`	`str \\| None`	`"0.0.0.0"`	Address the API listens on (keyword-only). In a cluster the other nodes call this port, so it must be reachable from them; `127.0.0.1` suits a single node you only query locally. Env: `TURBINE_API_HOST`.
`api_token`	`str \\| None`	`None`	Shared bearer token the API requires on every route except `/health` and `/ready` (keyword-only). Every node of a cluster must use the same value. Unset: the API is open and a warning is printed at startup. See Securing the API. Env: `TURBINE_API_TOKEN`.
`metrics_token`	`str \\| None`	`None`	Optional bearer token accepted only on `/metrics` (keyword-only). Requires a distinct `api_token`; the admin token continues to work for scrapes. Changing it requires restarting the node. See Scraping the metrics. Env: `TURBINE_METRICS_TOKEN`.
`api_cors_origins`	`list[str] \\| None`	any origin	Origins a browser may call the API from, as `scheme://host[:port]` (keyword-only). Only matters for a console or page served from another origin. Env: `TURBINE_API_CORS_ORIGINS` (comma-separated).
`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]] \\| str \\| None`	`None`	Every node of the cluster, this one included — the same list on every node. Where a node finds its cluster, and on the very first start, whom it forms it with. Env: `TURBINE_RAFT_PEERS` (format `"1=http://a:8400,2=http://b:8400"`). See How a node finds its cluster. Or an address template containing `{ordinal}`, for a Kubernetes StatefulSet (`"http://turbine-{ordinal}.turbine:8400"`, with `raft_initial_size`): see Kubernetes.
`raft_initial_size`	`int \\| None`	`None`	With a `raft_peers` template: the number of nodes of the first deployment. The node id becomes the ordinal at the end of the host name (`turbine-2` → `2`), the listen address the template with that ordinal, and the peers ordinals `0` up to the larger of `raft_initial_size` and the node's own. Keyword-only. Env: `TURBINE_RAFT_INITIAL_SIZE`.
`raft_rebalance_delay_ms`	`int`	`60000`	How long a stopped or crashed node keeps its partitions before they move to the other nodes. Nobody consumes them meanwhile; a node back within the delay resumes them. `0` moves them as soon as the node is seen down. Keyword-only. Env: `TURBINE_RAFT_REBALANCE_DELAY_MS`. See When a node goes down.

Standalone (single-process) deployment leaves all raft_* parameters None. See Deployment → Raft cluster for how nodes form and join a cluster.

`@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(...)` or named topic mapping	(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.

For stateful subscriptions, batch_size also influences how many due timers run in one committed batch, within a range of 1,024–16,384 timers. This applies to persistent window closes, early fires and user state.register_timer callbacks. Larger values amortise publishing and state writes when many timers are due together. Each timer batch finishes before input processing or shutdown resumes: expensive callbacks can cause longer ingestion stalls and slower shutdown, even when the timer service's time budget has expired. Smaller values can improve responsiveness. This is a throughput/latency trade-off, not a close-delay guarantee: a window becomes eligible at its deadline, and publication still takes time when many groups are due together.

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. Pass an explicit name="…" to keep the identity pinned while the function name changes. For a handler with persisted state, Turbine records this name and refuses to restore state or a checkpoint owned by another subscription. A stateless handler with a new name instead resumes from auto.offset.reset under its new consumer group.

Subscriptions have stable identities based on their input topic and name, in standalone and cluster mode. Reordering declarations preserves workers, progress and checkpoints. Adding a subscription starts its own processing history; removing one stops its workers once no node declares it, keeping its state for a later reintroduction. Renaming creates a new subscription and consumer group, with fresh state; it does not migrate the old subscription's history.

Reintroduction can only reuse history that is still retained. A stateless subscription loses its resume position if the broker expires its consumer-group offsets while it is removed, and then uses the fresh-group starting policy (the topic head by default). Exactly-once timer recovery also requires retained broker commit history; see Retention during an absence. Input-topic retention must cover the unconsumed records too.

During a rolling update, nodes may declare different sets of subscriptions. Turbine only assigns a subscription to nodes that declare it, including when the leader still runs the previous application version. Removing a subscription from one node can move its work to another node that still declares it. A standalone app stores its subscription identities locally and in its checkpoint storage before starting workers. If a rescheduled process loses its local working directory, it recovers these identities from the same checkpoint URL, preserving indices after removal or reordering. Keep that storage for the application's lifetime and use a separate checkpoint URL/prefix for each application history. If startup was interrupted after saving the checkpoint registry, the next boot repairs an older local registry from that compatible copy automatically. Conflicting subscription identities or established input geometry still refuse startup; preserve both copies and restore the matching history before restarting. Run only one standalone process for that history at a time. Without shared checkpoints, preserve state_dir across restarts. When first upgrading an existing standalone application to the registry, start it once with its original declaration before reordering or removing subscriptions. If the only remaining pre-registry state is an offline rescale export, first finish recovery with the previous runtime and original declaration, allowing it to write full checkpoints. The export alone does not contain enough topic ownership evidence to initialize the registry, and startup preserves it unchanged. See cluster format upgrades for the initial cluster runtime upgrade.

Existing persisted state and checkpoints from versions that did not record subscription ownership cannot establish that identity and are refused unchanged. There is no automatic migration or override through the starting-offset flags or receipt re-enrollment. To start a separate history, use a new app_id and fresh state and checkpoints; preserve the old copies if you need to recover their history with the previous version.

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.
`partition_hash`	`str \\| None`	`None`	The hash the producer places records with, on the `partition_key` column. Only `"murmur2"` (Kafka's default partitioner; librdkafka clients: `partitioner=murmur2_random`). Declared, a partition cleans up at start-up the state of the keys it no longer receives after a partition-count change; undeclared, Turbine never deletes state on the strength of what it observed. Requires `partition_key`. See Producer placement.
`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.

Per-key custom timers must use state.scope(partition_key_value) to reach the same handler as their input. See State and timers; SDK windows manage this automatically.

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 counts removals by compaction; turbine_state_keys_expired_by_sweep_total counts removals by the optional active sweep, broken down by scope. The dashboard combines their rates as cleanup activity; an older stored version can be counted after its logical key was deleted by the sweep. turbine_state_expiry_frontier_ms shows how far expiry has advanced. See Metrics reference.

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 \\| None`	`None` (= `"replay"`)	Crash-recovery policy for the state gap a hard crash leaves. `"replay"` (default) silently re-consumes the gap to rebuild state before resuming with outputs enabled, and refuses to start if retention has deleted part of it; `"accept"` skips the gap and loses its state updates (counted in `turbine_eos_state_gap_skipped_events`). EOS-only — any explicit value 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.

import pyarrow as pa

Event = pa.schema([
    pa.field("user_id", pa.string(), nullable=False),
    pa.field("event_ts", pa.int64(), nullable=False),
    pa.field("score", pa.float64(), nullable=True),
])
events = kafka.topic(
    "events",
    schema=Event,            # explicit Arrow decode schema
    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`	`pyarrow.Schema \\| type \\| None`	`None`	Explicit payload columns, Arrow types and nullability for JSON decoding. Accepts an Arrow schema directly or a plain Pydantic `BaseModel` subclass (see below).
`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. When no payload `event_time=` is declared, use `time_column="_kafka_ts_ms"` on a window for event-time semantics. A window must use the clock driving its subscription's watermark; other columns are refused.
`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.

An explicit schema also determines which payload columns reach the handler: undeclared fields are not materialized in its batch. If the application uses only a few fields, declaring those fields can reduce decoding work. Include every payload field needed by the handler, partitioning, event-time extraction and inference. Without an explicit schema, Turbine infers one from the first batch of each partition after startup and reuses it; fields absent from that batch remain absent until restart. An explicit declaration keeps columns and types stable across partitions and restarts; the benefit of a narrower declaration is avoiding unused columns, rather than avoiding repeated inference.

Arrow schemas and Pydantic models¶

Use pa.field("title", pa.string(), nullable=True) for a column that accepts null or missing values, and nullable=False for a required value. Arrow fields default to nullable=True; declare this explicitly when the distinction matters, especially for join schema evolution. Handlers always receive Arrow batches. The declared JSON schema is checked at startup against the decoder. Unsupported types, such as dictionary, interval or union fields, refuse startup and name the input and field, even under on_error="skip" or "dlq"; they are configuration errors, not rejected records.

Pydantic is optional: install turbine-fast[pydantic] if you prefer model declarations. Schema models must describe JSON objects with named fields; Pydantic RootModel declarations are refused. Plain models support str, int, float, bool, bytes, Optional[X] / X | None, and homogeneous arrays list[X] / tuple[X, ...] (including nesting). Only | None / Optional makes a model field nullable:

from pydantic import BaseModel

class Event(BaseModel):
    user_id: str
    event_ts: int
    score: float | None = None

Turbine reads field annotations without constructing or validating model instances. title: str = None is refused at subscription declaration: write title: str | None = None. Models with custom constructors, model_post_init or core-schema hooks, validators, constraints (including nested constrained annotations), aliases, non-null defaults, default factories, serializers, computed fields or explicit model configuration are also refused, because these behaviors are not applied by the decoder. A None default on a nullable field is accepted; a missing value decodes as null. Descriptive field metadata such as Field(description="...") is accepted. Apply business validation and defaults explicitly in your handler, or use an Arrow schema for the decode contract.

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`
`api_host`	`TURBINE_API_HOST`
`api_token`	`TURBINE_API_TOKEN`
`metrics_token`	`TURBINE_METRICS_TOKEN`
`api_cors_origins`	`TURBINE_API_CORS_ORIGINS`
`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_rebalance_delay_ms`	`TURBINE_RAFT_REBALANCE_DELAY_MS`
`raft_initial_size`	`TURBINE_RAFT_INITIAL_SIZE`

Other operational settings are available as environment variables:

Environment variable	Default	Description
`TURBINE_ALLOW_GIL`	unset	Set to `1`, `true` or `yes` (case-insensitive) to let Turbine start with the GIL enabled. Turbine emits a `RuntimeWarning` because the GIL limits Python parallelism across partitions. With the GIL enabled, startup fails by default, including on a free-threaded build if an extension re-enabled the GIL.
`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_CLOSE_BACKLOG_THRESHOLD_MS`	`60000`	Health reporting only: `/health` reports `degraded` with `reason="close_backlog"` when continuously observed real-time debt reaches this threshold. Event-time age is watermark lateness; a fresh backfill jump alone does not degrade health. Lower the threshold for earlier alerts, or raise it for expected slow boundary bursts. Closing automatically gains priority after one second of observed due work, with an independent safety drain for timers aged 60 s. Changing this value does not change scheduling or guarantee a publication delay.
`TURBINE_ALLOW_NON_DIVISOR_PARTITIONS`	unset	Set to `1` to start a stateful subscription with a `partition_key` on a topic whose partition count does not divide 960, which Turbine otherwise refuses. The app runs normally, but a topic cannot be rescaled to such a count without losing state. See Topic Partitioning.
`TURBINE_POINTER_WAIT_MS`	`2000`	Maximum time (ms) the resident cold-restart path waits for a local copy of the cluster checkpoint pointer. Set to `0` to skip that wait. Ordinary cluster workers obtain their checkpoint choices directly from Raft.
`TURBINE_PIPELINE`	adaptive	The worker loop's front stage (poll + decode) runs one batch ahead on a dedicated thread, overlapping the previous batch's process/produce/commit. The overlap only pays when the back stage is a real share of the batch cycle, so the default is adaptive: every 128 non-empty batches the worker measures the back's share of its cycle and falls back to the serial loop below 30 % (front-dominated — the pipeline's handoff costs more than the overlap it buys), re-arming above 40 %. With full batches, switching to serial starts a 32-batch trial: if the measured time per record is more than 5 % higher than in the preceding pipelined period, the worker restores the pipeline and waits eight qualifying 128-batch periods (1 024 non-empty batches with a back share below 30 %) before trying serial again. Partial batches keep the share-based rule without this throughput comparison. The cycle is read mode-locally (`front_wait + back` pipelined, `front + back` serial) so arrival-bound streams cannot flap the gate. `1` pins the pipeline on, `0` pins the serial loop (bench protocols needing deterministic arms). Always off during replay.
`TURBINE_PARALLEL_DECODE`	adaptive	Batches above `TURBINE_PARALLEL_DECODE_MIN_ROWS` (default 25 000) are split into ~25k-row chunks decoded in parallel and concatenated in order. In the pipelined loop the gate is adaptive: parallel decode stays on only while the front stage is the bottleneck (the back waited for it); a back-bound worker falls back to serial decode to avoid contending with process+commit. The serial loop (`TURBINE_PIPELINE=0`) always decodes in parallel past the threshold. `always` pins it on; `0` disables chunking entirely.
`TURBINE_BLOCK_CACHE_MB`	a quarter of the memory limit	Shared memory budget in MiB for cached on-disk state blocks and state write buffers. One budget serves every partition of the process. By default it is a quarter of the process's memory limit: the container's limit, or the host's memory when there is none. A positive integer with no unit (`256`, not `256MB`); invalid values fail at startup. Raise it to retain more read blocks and flush writes less often; lower it to leave more memory to the rest of the process. This is a soft budget, not a process memory limit. See Memory.
`TURBINE_STATE_CACHE_SIZE`	`500000`	Positive integer entry limit, with no size suffix (use `8000000`, not `8M`); invalid values fail at startup. Applies to each store's read cache and the hints that avoid re-registering open-window close timers. Sliding windows also size their close-timer cache with this setting. Size it for the working set (`distinct groups × simultaneously open windows per group`): a smaller cache can cause repeated timer registrations. Each worker/shard store and sliding window has its own cache. Larger limits can reserve memory at startup as well as allow more growth with active entries; this is not an application-wide memory budget. `turbine_state_cache_bytes` shows what the caches hold; see Memory.

Memory¶

Besides the batches in flight and the Kafka client's read-ahead, a stateful application keeps part of its state in memory. That part grows with the state, not with the traffic, and stays at its bound once it has reached it, so count it in full:

The block cache keeps recently read blocks of the on-disk state. One cache serves every partition of the process, and once full it evicts the least recently used blocks. Its budget also accounts for the write buffers below. It takes a quarter of the process's memory limit by default — the container's limit, or the host's memory without one — and TURBINE_BLOCK_CACHE_MB sets it.
The value caches keep the state values a partition recently read or wrote, so a key that comes back in a later batch is not read from disk again. Each stateful partition has one per parallelism thread, bounded at TURBINE_STATE_CACHE_SIZE entries. What an entry weighs depends on your values: read the total from turbine_state_cache_bytes once the state has grown.
The write buffers hold recent state writes until they reach disk. All partitions share a flush target of half the cache budget, and their allocations are charged against that same budget. The other half leaves headroom for read blocks; concurrent writes and retained reservations can reduce this headroom. Checkpoints also flush these buffers. A smaller budget can cause more frequent flushes and disk work. A write that reaches the shared target waits for its own partition to flush before returning. Concurrent batches can temporarily exceed the target. Existing per-partition write limits still apply.

The shared budget is not an RSS cap: value caches, Kafka queues, batches, Python and other allocations still need headroom, and more partitions increase those costs. turbine_block_cache_usage_bytes includes write-buffer reservations; do not add turbine_memtable_usage_bytes to it. Reservations shrink only after buffer usage falls far enough. They can remain above current usage while the process is idle. Use turbine_memtable_budget_bytes to inspect the flush target and turbine_state_cache_bytes for the separate value caches.

Window input and state contracts¶

Compact window calculations support at most 255 aggregate state columns. Count, Sum, Min and Max each use one column; Mean uses two. Multi shares identical state columns between its members. Wider definitions raise an error when the shared operator is constructed. Custom opaque calculation payloads do not use this compact column layout.

A sliding window spans at most 2^20 panes (size_ms / slide_ms); a wider ratio raises an error when the shared Rust operator is built, at app boot or on the first delegated batch. All windows use Rust mechanics, including custom Python calculations and late-event routing. There is no Python window engine or engine-selection switch.

Native window keys accept string, int32 and int64 columns; sessions also accept an inferred all-null key column. Event-time columns must contain non-null integral epoch milliseconds within the int64 range. Convert other key/time types explicitly before updating a window. Unsupported types raise an error.

Native Sum, Mean, Min and Max accept boolean, integer, floating-point (including float16), decimal and null columns. Convert text or binary values to a numeric column before aggregation; they are not implicitly parsed. Count(column) accepts any column type and counts non-null values.

Persistent-window state requires a fresh start on upgrade

This version changes the persisted format of fixed windows (tumbling and sliding). Upgrading from an earlier untagged format requires a fresh state directory, a fresh checkpoint location and a new app_id, with a consumer start position chosen to replay the input needed to rebuild state. Incompatible stored state is refused; there is no automatic migration.

Event-time persistent sessions also require a fresh start when upgrading from a version that armed their closes on wall-clock timers. Existing session timers are not automatically converted to watermark deadlines. This applies to both Python and native execution; their early-fire timers remain wall-clock timers.

Integer and decimal partials reduce before conversion to the accumulator's storage type. Float64 storage can round large integers, and floating reductions can differ from the Python implementation, especially under cancellation. Changing a calculation, storage schema or application language does not guarantee that existing state can be resumed. Start with fresh application state and a consumer identity/start position appropriate for the intended replay.

Named inputs¶

input={"left": topic_a, "right": topic_b} reads co-partitioned topics through one stateful handler. ctx.input identifies the source; input_idle_timeout_ms=None holds progress for a never-seen input, and a positive duration explicitly permits excluding it after inactivity. input_idle_policy="coast" keeps advancing expiry clocks during silence. Set input_idle_policy="hold" with a positive timeout to pause coasting independently in each partition after all its inputs become idle, without counting the pause on resume. A partial resume can delay expiry while a sparse input's clock catches up. Under "hold", the timeout governs both never-seen input exclusion and outage detection; a longer value accommodates more startup lag but lets clocks advance further before freezing. See Reading co-partitioned inputs for placement requirements, watermark behavior and recovery guarantees.