Deployment¶

For the full list of parameters accepted by Turbine(...) and @app.subscribe(...), see Configuration. This page focuses on operational concerns: deployment topology, performance tuning, and rolling upgrades.

Packaging an app¶

A Turbine app is an ordinary Python process. The engine, the state store, the REST API and the console all live inside the installed package — there is no cluster manager to stand up and no job to submit. Deploying means shipping your own program plus its dependencies, and running it.

Two constraints shape the image. The interpreter must be free-threaded, and there is no official free-threaded CPython base image, so the shortest path is an image that has uv and lets it install the interpreter your project pins. And the version should be pinned: Turbine is alpha, and an unpinned pre-release resolves to whatever is newest the day the image happens to be rebuilt.

# pyproject.toml
[project]
name = "my-pipeline"
requires-python = ">=3.14"
dependencies = ["turbine-fast==0.2.0a2"]

[tool.uv]
prerelease = "allow"          # a property of the project, not of one command line

FROM ghcr.io/astral-sh/uv:debian-slim
WORKDIR /app

COPY pyproject.toml uv.lock .python-version ./
RUN uv sync --frozen --no-dev

COPY app.py ./
ENV TURBINE_API_PORT=8400
EXPOSE 8400
CMD ["uv", "run", "--frozen", "--no-dev", "python", "app.py"]

.python-version holds 3.14t; uv sync downloads that interpreter and resolves the cp314t wheel. --frozen on both commands is what stops the container from re-resolving against PyPI every time it starts — a startup that reaches the network is a startup that can fail for reasons unrelated to your app. Expect an image in the 400–500 MB range: pyarrow and the compiled engine dominate it.

Stopping it. Turbine shuts down on SIGTERM as well as SIGINT — it finishes the batch in flight and writes a final checkpoint. That is what docker stop and a Kubernetes pod termination send, and it reaches the app through uv run. Give it a termination grace period comfortably longer than one batch plus one checkpoint (how to size it: Sizing the shutdown); killing it earlier is a crash, and costs you the recovery path instead of the shutdown one. In a cluster, a stop is not a departure: the node stays a member and its partitions wait for it (see When a node goes down). To take a node out for good, see Removing a node.

What the container needs. A reachable broker; the API port published if you want the console or Prometheus; and — for anything that can be rescheduled onto another host — a checkpoint_url on shared storage, since state_dir is a working directory, not durable storage. Configuration can come from the environment rather than code: see the environment variable summary and, for object storage, the AWS_* variables below. Wire /health and /ready to the liveness and readiness probes — see Health probes, which explains why they must not be the same one.

Deployment Modes¶

Standalone (single node)¶

No extra configuration needed. One process handles all partitions.

kafka = KafkaBroker(bootstrap="kafka:9092")
app = Turbine(kafka)

The same kafka handle builds every input=/output= topic reference, so the cluster address appears once — see Configuration for the full constructor.

When this process can move to another host, configure checkpoint_url on shared storage and keep the same URL across deployments. Turbine stores subscription identities there alongside worker checkpoints, so changing declaration order or removing a subscription does not change surviving workers' recovery paths after the local working directory is lost. A checkpoint URL/prefix belongs to one application history; only one standalone process may run that history at a time.

Raft cluster (dynamic)¶

Nodes coordinate via embedded Raft consensus. Partition assignment is replicated and rebalanced automatically.

Every node gets the same list of peers — every node of the cluster, itself included — and the same command serves its first start and every restart:

kafka = KafkaBroker(bootstrap="kafka:9092")
app = Turbine(
    kafka,
    raft_node_id=1,                          # 2 and 3 on the other nodes
    raft_listen_addr="http://10.0.0.1:8400", # this node's own address
    raft_peers=[(1, "http://10.0.0.1:8400"),
                (2, "http://10.0.0.2:8400"),
                (3, "http://10.0.0.3:8400")],
)

All Raft parameters have TURBINE_RAFT_* environment variable equivalents:

export TURBINE_RAFT_NODE_ID=1
export TURBINE_RAFT_LISTEN_ADDR=http://10.0.0.1:8400
export TURBINE_RAFT_PEERS="1=http://10.0.0.1:8400,2=http://10.0.0.2:8400,3=http://10.0.0.3:8400"
# optional: how long a node that goes down keeps its partitions (default 60000)
export TURBINE_RAFT_REBALANCE_DELAY_MS=60000

How a node finds its cluster. At startup a node asks its peers about their cluster:

The very first start forms the cluster once every peer has answered that it has none: the node with the lowest id forms it, and the others join it. Until every peer answers, nothing is formed — start the nodes together (on Kubernetes, podManagementPolicy: Parallel, see Kubernetes). A cluster is never formed on silence: a node that cannot tell whether a cluster exists waits, rather than risk two clusters side by side. If a peer will never answer (a host that is gone for good), tell the lowest-id node to form the cluster without it:
```
curl --fail-with-body -X POST http://10.0.0.1:8400/cluster/force_bootstrap
```
It is refused once the node is part of a cluster, and a cluster found among the peers still wins: the node joins it instead. But only a cluster it can reach stops it: sent while a running cluster is merely unreachable from this node (a network split, a firewall rule), it forms a second cluster beside it, and both then process the same partitions. Use it only when you know no cluster is running.
Adding a node to a running cluster: give it a list with the existing nodes and itself. It finds the cluster through them and joins it. The other nodes' lists do not need to change for that; update them at their next restart so they stay accurate. The new node first copies the cluster's history without a vote. It becomes a voter only after its subscription names, topics and order match the cluster's declaration. A refused new node therefore leaves the existing voting quorum unchanged. A node already in the voter set keeps voting while its application configuration is checked on restart. If startup is refused, open the console's Nodes view on a surviving node to read the retained reason. Correct the startup error and restart that node; successful admission clears the refusal. If the cluster could not retain it, inspect the failing node's startup logs.
A node that left (/cluster/leave) and is started again joins back on its own: starting a node means wanting it in the cluster. To take a node out for good, leave and stop it (see Removing a node).

Use the same list on every node. A node that does not list some of its peers may form a cluster of its own with the ones it knows.

Keep each node's state directory. A node keeps its cluster membership — its Raft vote and log, and the replicated assignment table — in state_dir, next to its partitions' state. Put state_dir on storage that survives a restart (a persistent volume, not the container's filesystem):

A node restarted on its directory resumes as the member it was, whether it was stopped or crashed. The cluster restarts the same way after a full outage: membership and assignments are read back, nothing is rebuilt from scratch.
A restarted node starts no partition until it has caught up with the cluster. What its directory remembers is how the cluster looked when it stopped; a partition moved meanwhile is never resumed from that memory. A node that cannot reach a leader therefore waits, and runs nothing.
Each node needs its own directory. A node refuses to start on a directory another node id created — check the node id and the volume mounted on that node.
If a node's directory is lost while it is still a member, start it again as usual: it sees that the cluster still counts it while it remembers nothing, takes no part in elections, has itself removed and joins again as a new member. A node that voted with no memory of its earlier votes could break the cluster's consistency; this is what prevents it.

Protect the API with a token. Nodes talk to each other over the same API port, so it must be reachable between them — and without a token, anyone else who reaches it can read state and change membership. Give every node the same TURBINE_API_TOKEN, from your secret store (a Kubernetes Secret, for example):

export TURBINE_API_TOKEN=...   # identical on every node

A node started with a different token cannot join (the error names the token), and the cluster views list it as unauthorized. Changing the token means restarting every node. See Securing the API for what the token covers and what it does not (it is not encryption).

When a node goes down¶

A node can go down two ways: it is stopped (SIGTERM: a rolling update, a drain, a rescheduled pod) or it crashes (an OOM kill, a lost host, a SIGKILL). Either way it stays a member of the cluster, and its partitions wait for it for raft_rebalance_delay_ms (default 60 000, env TURBINE_RAFT_REBALANCE_DELAY_MS):

Back within the delay: it resumes its own partitions where it left off. Nothing moved, nothing was downloaded.
Still down after the delay: its partitions move to the other nodes. When it comes back, it gets a share again.
Meanwhile nobody consumes those partitions. Their lag grows, and event-time windows on them stop closing until someone resumes them. /cluster/nodes shows the node as waiting, with the time its partitions move (see the rebalance view), and the turbine_partitions_awaiting_rebalance gauge counts them.

A stop counts from the moment it happens; a crash counts from the last time the node answered. A node that is the leader hands its role to another node before it stops, so the cluster does not pause for an election.

Choosing the delay is a trade between two costs. A long delay keeps restarts cheap but leaves a dead node's partitions unconsumed for that long. A short one reacts quickly but moves partitions on every restart that takes longer, and moves them a second time when the node comes back. Size it to how long one of your restarts takes (pulling the image plus starting). 0 moves them as soon as the node is seen down, about two seconds after its last answer.

With processing_guarantee="exactly_once", a partition that changes owner waits for its previous owner to confirm the handover. A stopped node confirms on its way out. A crashed node cannot, so its partitions stay blocked on their new owner until the crashed node comes back (it confirms then, for the process that died) or until you release them by hand (Recovering a stalled partition handoff).

With processing_guarantee="at_least_once", the new owner starts without waiting for the previous owner to finish. Both may process the partition and publish outputs until the previous owner stops; downstream consumers must tolerate duplicates. Stateful checkpoints keep each owner's file contents separate when uploads overlap. A draining worker can still publish its final checkpoint while it holds its claim. A released claim or a newer published epoch fences late publications. The checkpoint store can retain bytes from a refused upload until a later collection. Revoked workers receive their stop requests together and finish their current work and final checkpoints concurrently. A node stop during that handover also signals all its remaining workers immediately. Draining still takes time: an idle worker may wait out its batch timeout, and processing and checkpoint uploads must finish. Size the pod's shutdown grace period for that work, including shared upload bandwidth.

If startup fails or is cancelled before application workers are launched, the node relinquishes its unstarted assignments and the other nodes take over without a manual release. They still wait for any earlier owner that is finishing a handover. A hard kill before startup cleanup, or a loss of quorum that prevents that cleanup from committing, retains the crash-recovery behavior above.

A new stateful worker in cluster mode needs a Raft quorum to authorize its assignment and obtain approved recovery checkpoints. Exactly-once stateless workers also need authorization. At-least-once workers still start without waiting for the predecessor to finish. When Raft is unavailable, an ordinary stateful worker that already completed recovery can restart at the same epoch from valid local state. Exactly-once still validates that state against the broker receipt before a local-only retry. Each worker releases its own assignment after draining. A successor can finish first while the predecessor still holds its claim. Cancellation retries transient coordination errors for up to five seconds. If the outcome remains unknown, the worker exits without accessing Kafka or releasing an unproven handoff.

Removing a node¶

A stopped node still counts as a member, so a node you take out for good — scaling down, decommissioning a host — has to leave first. Otherwise the cluster keeps waiting for it: its partitions move after the delay, but it still counts in the quorum, so the cluster tolerates one failure fewer. Ask the node itself to leave, then stop it:

curl --fail-with-body -X POST http://10.0.0.3:8400/cluster/leave   # the node that leaves, not another one

It hands its partitions over (each one through a final checkpoint) and leaves the voters. To remove a node that is already gone, ask any other node:

curl --fail-with-body -X POST http://any-node:8400/cluster/remove_voter \
  -H 'Content-Type: application/json' -d '{"remove_voter_ids": [3]}'

Both operations retain the removed voter as a learner: it no longer votes or receives partitions, but the leader continues attempting to replicate to it. A new node refused during startup can also remain a learner. There is no automatic cleanup after a timeout, so a temporarily unavailable node keeps its membership.

When the node is permanently retired, keep its process stopped and remove its learner membership through any reachable node:

curl --fail-with-body -X POST http://any-node:8400/cluster/remove_learner \
  -H 'Content-Type: application/json' -d '{"remove_learner_ids": [3]}'

Add -H "Authorization: Bearer $TURBINE_API_TOKEN" when a token is configured. A successful removal returns HTTP 200 with {}; a refusal uses an HTTP error status with the reason in its Problem Details detail field. The operation preserves the target voting set and refuses the whole request if any id remains a voter in that set. During a joint membership transition, an outgoing voter already absent from the target set may be removed; keep its process stopped as above. An already absent id is harmless, so retries are safe.

Removing a learner stops replication to it. It preserves partition ownership and handoff-release history, worker state and checkpoint pointers; it does not release a stalled handoff or drain a live worker. Use the separate handoff recovery procedure when needed. This is not a ban: starting the removed node again asks to rejoin the cluster, with its existing state and the usual startup admission checks.

Kubernetes¶

Run a cluster as a StatefulSet: each pod gets a stable name its peers can reach, a volume that follows it, and an ordinal that serves as its node id. The manifests, what each setting is for, the shutdown and memory sizing, opening the console, metrics and the day-to-day operations are on their own page: Kubernetes.

Checkpoint storage¶

Each partition keeps its working state in a local RocksDB and periodically writes a snapshot to the checkpoint store named by checkpoint_url. That store is what a different node reads when a partition moves to it — on failover, on a rebalance, during a rolling upgrade.

Hence the requirement: in a cluster, the checkpoint store has to be readable from every node. The default, file://{state_dir}/snapshots, is not — it is a directory on one machine. With a node-local store, every partition that moves loses its state twice over: the new owner finds no snapshot to restore, and when the old owner gets that partition back later it resumes from a RocksDB that has been going stale the whole time. This is not "state starts empty", it is "state is wrong" — a window that under-counts, a cooldown key that has silently expired, an alert that fires a second time and whose resolution never comes.

Nothing about that depends on the delivery guarantee. at_least_once promises no message is dropped; it promises nothing about reloading state. So a cluster + stateful app refuses to start on a node-local checkpoint store, whatever its processing_guarantee:

ValueError: Cluster mode + stateful processing requires a shared checkpoint_url
(e.g. 's3://bucket/path'). The current value None is local to this node, so
every partition that moves between nodes loses its RocksDB state: the new owner
has no snapshot to restore, and the old one comes back later with stale state —
wrong results, not empty ones.

Single-node deployments are untouched: nothing moves, local is the correct answer, and the boot banner marks it (local to this node).

Escape hatch: several processes, one filesystem¶

Running a cluster of processes on a single host — where file:// really is shared — is the one case the check gets wrong. Set TURBINE_ALLOW_LOCAL_CHECKPOINT_IN_CLUSTER=1 to proceed; the app starts with a warning, and the banner switches to [LOCAL - every partition handoff loses state]. Do not use it across hosts.

S3 and S3-compatible endpoints¶

app = Turbine(kafka, checkpoint_url="s3://turbine-state/my-app", ...)

The store must support and enable atomic conditional writes (If-None-Match). Turbine checks at startup that creating an existing object is refused without replacing its contents. An endpoint without this capability cannot safely store checkpoints and is refused before processing starts.

Manifest timestamps and collected SST generations have permanent retirement markers. Keep these markers: a delayed request can recreate deleted manifest bytes, but the timestamp marker prevents readers from selecting them. Pruned or corruption-purged timestamps cannot be reused.

Credentials and endpoint configuration come from the standard AWS_* environment variables — there is no Turbine-specific setting for them. The same variables are how you reach a non-AWS S3-compatible endpoint (OVH, MinIO, Scaleway, Ceph, Cloudflare R2):

Variable	Purpose
`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`	Static credentials. `~/.aws/credentials` and instance roles are also honoured.
`AWS_SESSION_TOKEN`	Temporary credentials.
`AWS_WEB_IDENTITY_TOKEN_FILE` + `AWS_ROLE_ARN`	Web-identity federation (IRSA on EKS, workload identity).
`AWS_REGION`	Region. `AWS_DEFAULT_REGION` is accepted too. Compatible endpoints usually want their own region name (`gra`, `us-east-1`, …).
`AWS_ENDPOINT_URL`	The endpoint of a non-AWS provider. Unset means real AWS.
`AWS_VIRTUAL_HOSTED_STYLE_REQUEST`	`true` if the provider requires the bucket in the hostname. Path-style is the default, which is what most compatible endpoints expect.
`AWS_ALLOW_HTTP`	`true` to allow a plaintext endpoint — a local MinIO, not production.

OVH Object Storage:

export AWS_ENDPOINT_URL=https://s3.gra.cloud.ovh.net
export AWS_REGION=gra
export AWS_ACCESS_KEY_ID=...
export AWS_SECRET_ACCESS_KEY=...

Local MinIO:

export AWS_ENDPOINT_URL=http://minio:9000
export AWS_ALLOW_HTTP=true
export AWS_REGION=us-east-1
export AWS_ACCESS_KEY_ID=minioadmin
export AWS_SECRET_ACCESS_KEY=minioadmin

The bucket must already exist, and the credentials must be allowed to read and write objects under the URL's prefix. Turbine does not create the bucket.

Checked at boot, watched afterwards¶

A stateful app writes a small object to the store at startup, reads it back and deletes it. If that fails, the app does not start, and the error names what to check. A store you cannot write is a configuration error, and the first real checkpoint is driven by batch count — so without this check you would find out much later, quietly, with state accumulating that nothing is saving.

After startup the same failure is handled the opposite way: the app keeps running. An unreachable bucket is usually an outage that repairs itself, or one someone has to go fix; stopping the pipeline brings it back no sooner and costs you the processing too. What you get instead is state you can alert on:

Signal	Meaning
`turbine_checkpoint_store_shared`	`0` when the store is local to this node. Alert on it together with cluster mode — that pair is the trap above.
`turbine_checkpoint_store_writable`	`0` when the store's last answer was a failure.
`turbine_checkpoint_consecutive_failures`	Failures since the last success.
`turbine_checkpoint_failures_total{operation}`	Cumulative failures, split by step (`upload`, `prune`, `probe`, …).
`/health` → `checkpoint`	The same facts as JSON: destination (credentials redacted), `shared`, `writable`, time of the last success, and the last error text.

After three consecutive failures /health reports status: "degraded" with reason: "checkpoint_store_failing". It stays 200 on /ready — the node is still processing, just not saving; taking it out of rotation would not help. It clears itself as soon as a write succeeds, with no restart.

The real snapshot uploads are the sensor. When they go quiet for a minute — an idle partition never checkpoints, because the cadence follows batches — a small probe keeps the answer current, so "no failures recently" never quietly means "nothing was tried".

Health probes¶

Two endpoints, and the difference between them matters for what you wire a probe to:

Endpoint	Answers	HTTP status
`/health`	Is this process alive?	always `200` while it serves HTTP
`/ready`	Can this node do its job?	`503` when it cannot, `200` otherwise

Both answer without a token even when the API requires one, so a probe needs no credentials. Everything else does — including /metrics: give your Prometheus scrape job a separate metrics-only token (see Scraping the metrics).

Both return the same JSON body, whose status field is the one thing to read:

`status`	Meaning	Typical action
`healthy`	Participating normally.	—
`degraded`	Working, but part of this node is down — a partition halted (see Error handling), a partition's handler is failing batch after batch, or the checkpoint store stopped accepting writes.	Alert a human; restarting won't help.
`unavailable`	Stopping (`reason: "stopping"`), or cannot participate: no Raft leader elected, this node is not (yet) a voter, or it has just restarted and not yet caught up with the cluster (it runs no partition until it has — usually a few seconds).	During shutdown, wait for drainage; otherwise investigate connectivity/quorum if it lasts.

The second degraded case is the one that hides best: under on_error="skip" or "dlq" a broken handler leaves the node looking busy — nothing is halted, offsets keep advancing — while every row it reads is discarded. reason is processing_errors and the processing_errors list names the partitions, so a page tells you which ones and since when; see Knowing when a partition has stopped producing results.

When status is not healthy, reason names the cause in a fixed vocabulary: stopping, no_raft_leader, not_a_voter, catching_up, halted_workers, processing_errors, checkpoint_store_failing, close_backlog. In cluster mode the body also carries leader_id (null when there is no leader) and voters, so one request tells you whether the cluster has a quorum — no second endpoint, no log reading:

# Node up but unable to work: peers unreachable, no leader elected.
curl -s http://10.0.0.3:8400/health
{"status":"unavailable","reason":"no_raft_leader","workers":0,
 "node_id":3,"role":"follower","leader_id":null,"voters":3}

# Readiness probe — no JSON parsing needed.
curl -fsS http://10.0.0.3:8400/ready >/dev/null || echo "not ready"

A node with zero workers is not a fault: with fewer partitions than nodes, some nodes legitimately hold nothing. That is why readiness keys on cluster participation rather than on the worker count.

For Kubernetes, wire /health to the liveness probe and /ready to the readiness probe (the StatefulSet does). Do not point a liveness probe at /ready: a node without quorum is not fixed by being killed, and restarting it makes the quorum problem worse.

Recovering a stalled partition handoff¶

Applies to cluster mode with processing_guarantee="exactly_once" only.

When a partition moves from one node to another, the new owner does not start processing immediately: it waits until the previous owner confirms it has finished and stepped away. Starting earlier would take over the partition's identity at the broker while the old owner still has a transaction in flight, and that transaction would be aborted — losing the outputs it had already produced. So the wait is deliberate, and it has no timeout: giving up after a delay would just re-introduce the problem it exists to prevent.

The consequence is the case to know about. If the previous owner dies hard — power loss, SIGKILL, kernel panic — it never sends its confirmation, and the new owner waits forever. That partition simply never starts. The rest of the node keeps working, which is exactly what makes it easy to miss.

How you see it:

Signal	What it shows
`turbine_handoff_barrier_waiting{sub_index,topic,partition}`	`1` while waiting. Seconds during a normal rebalance; minutes means stuck
Node logs	A warning every 10 s naming the topic and partition
`turbine_consumer_lag`	Climbing on that partition while others are flat

How to unblock it. Confirm that every earlier worker for this partition has stopped. The command releases all earlier assignments through the epoch immediately before the current owner, including any still-draining worker. It also removes their protection for checkpoints they may be reading.

Then, from any machine that can reach a cluster node — any node will do:

curl --fail-with-body -X POST http://10.0.0.1:8400/cluster/force_release \
  -H "Authorization: Bearer $TURBINE_API_TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"topic": "events", "partition": 3}'

When several subscriptions read that topic, include "sub_index": 1 in the request (using the index shown in /assignments). The CLI accepts --sub-index 1. Without a selector, an ambiguous request is refused.

It answers with what it resolved, so you can check it matched the partition you meant:

{"topic":"events","partition":3,"node_id":2,
 "worker_id":"work-00-p03","epoch":4,"released_epoch":3}

A node that has just restarted, or that cannot reach the cluster's leader, answers 503 instead: it will not act on its own possibly outdated view of who owns the partition. Send the request to another node, or retry once it has caught up.

If you run Turbine from a source checkout, turbine-cli wraps the same call and refuses to publish until you pass --yes — so a bare invocation is a dry run that shows you the partition's state first:

turbine-cli force-release --cluster http://10.0.0.1:8400 \
    --topic events --partition 3          # reports, changes nothing
turbine-cli force-release --cluster http://10.0.0.1:8400 \
    --topic events --partition 3 --yes    # publishes

It reads the token from TURBINE_API_TOKEN (or a .env file); --token also works, but a flag is visible to anyone listing processes.

The waiting node proceeds within a few seconds; watch the gauge drop back to 0. One run is enough even if several handoffs failed in a row on that partition.

Do not reach for it as a way to speed up a handoff that is merely in progress: a wait of a few seconds is the normal path, and cutting it short is what causes the loss described above. The command is for a wait that has no end because the other side is gone.

Performance Tuning¶

The batch loop is broken down into phases, each exposed as a Prometheus histogram:

Metric	Phase
`turbine_poll_seconds`	Waiting for messages from the broker
`turbine_decode_seconds`	Deserialising the batch into Arrow
`turbine_process_seconds`	Your handler (Python callback)
`turbine_produce_seconds`	Encoding + sending to the output topic
`turbine_commit_seconds`	Persisting state + advancing the input offset
`turbine_batch_duration_seconds`	Back-stage time (process → produce → commit)
`turbine_front_wait_seconds`	Time the worker waited for its next batch: poll + decode on serial loops, only the wait not hidden by the previous batch on pipelined loops. Empty polls are excluded

With a pipelined front stage (one batch ahead), poll and decode overlap the previous batch, and turbine_front_wait_seconds counts only the remaining wait. On serial loops, including replay, subscriptions with several inputs and repartition hops, it counts the whole poll + decode. Empty polls are excluded in every mode. Compare it against turbine_batch_duration_seconds to see how much time is spent waiting for input; the batch duration covers processing, producing and committing, with poll and decode excluded. For repartition hops, processing is key extraction. The pipeline is adaptive: a front-bound worker (front_wait ≈ the whole front) tries the serial loop on its own and resumes the pipeline if a full-batch trial increases processing time per message by more than 5% — watch turbine_pipeline_front_active (1 = pipelined, 0 = serial) and the adaptive pipeline gate switched log line. TURBINE_PIPELINE=0/1 pins the mode.

Compare them on your workload to find the dominant phase before changing anything else.

The main knobs you control from the SDK:

Knob	Where	What it changes
`batch_size` / `batch_timeout_ms`	`@app.subscribe(...)`	Larger batches amortise per-batch overhead (commit, produce, encode); smaller batches lower end-to-end latency.
`parallelism` + `partition_key`	`@app.subscribe(...)`	Scale CPU on a hot Kafka partition without changing the topic — see Partitioning.
`avro_schema` vs `schema` (Arrow or Pydantic)	`kafka.topic(...)` input	Avro can reduce decoding work on decode-bound workloads. An explicit JSON schema can skip undeclared columns; inferred schemas are cached too, so a declaration alone does not imply faster decoding. See input schemas.
`processing_guarantee`	`@app.subscribe(...)`	Stay on `at_least_once` if downstream is idempotent — EOS adds a per-batch transaction round-trip and a durable state flush. See Delivery Guarantees.
`state_format`	`Turbine(...)`	`"msgpack"` is a compact alternative to the default JSON state codec — typically lower CPU for large state values, identical durability.

Kafka client tuning (producer acks, linger.ms, compression, fetch sizing) is not exposed as a user-facing parameter today — Turbine picks defaults that are appropriate for the batch model.

Rolling Upgrade Procedure¶

First upgrade to HTTP membership errors

The membership API contract changes for node-to-node calls as well as external clients. Stop all nodes, upgrade them together, and restart with their existing state and checkpoints. Do not mix runtimes using the former 200 status envelope with runtimes using HTTP error statuses: an old refusal can be mistaken for success, including during recovery of a node that lost its state. A StatefulSet rolling update is not safe for this upgrade; use the full-stop commands.

First upgrade to startup-refusal diagnostics

These diagnostics add cluster messages that earlier runtimes cannot read. Stop all nodes before upgrading them together, retaining their existing state and checkpoints. Do not mix older runtimes with this runtime or downgrade their Raft stores after these messages have been committed. Existing ownership-format-5 and format-6 stores upgrade in place; upgrades from older formats still require the procedure below.

The rolling procedure below applies to compatible runtimes. A StatefulSet rolling update does not perform the coordinated stop required above; on Kubernetes, use the full-stop commands.

Restart the nodes one at a time, each with the same command and the same state_dir as before. Stop the node, upgrade it, start it again, then wait until it is back before you move on to the next one (with an API token set, add -H "Authorization: Bearer $TURBINE_API_TOKEN" to the call below):

curl http://10.0.0.3:8400/ready            # 200 once it is back in the cluster and caught up
curl http://10.0.0.3:8400/assignments      # its partitions are back

There is nothing else to do. The node resumes as the member it was, and a leader hands its role over before it stops. Its partitions wait for it, so keep each restart shorter than raft_rebalance_delay_ms; otherwise they move to the other nodes and back again. On Kubernetes, the StatefulSet's rolling update does all of this for you (see Kubernetes).

Upgrading the cluster ownership format¶

This release upgrades cluster ownership storage to format 8 for ordinary checkpoint recovery and retention. It accepts formats 5 through 7, preserves their Raft stores, worker state and checkpoints, and records format 8 before the node can append new messages. Older binaries refuse the upgraded directory.

Every cluster requires a coordinated full stop for this upgrade, even when no subscription enables resident storage. Stop all nodes, back up every state directory, upgrade all runtime versions, then restart with the original state/checkpoint locations and subscription declarations. Do not mix old and new runtimes in a rolling upgrade or reopen upgraded Raft stores with older binaries. Retain the pre-upgrade backups for any separately planned rollback. On Kubernetes, use the full-stop commands.

Once every node runs the compatible format-8 runtime, normal compatible application updates may use the rolling procedure above.

After the upgrade, the offline turbine-cli rescale command refuses checkpoint histories enrolled in Raft-managed recovery. See Rescaling a topic.

Upgrading from version 4 or earlier still requires fresh Raft stores on every node. Those stores are refused unchanged. Stop all nodes and move each node's {state_dir}/_raft directory aside, retaining it as a backup. Re-form the cluster with fresh Raft stores before restarting. Keep worker state directories and checkpoint storage in place, and use the original subscription declaration for that first startup. Older state or checkpoint formats may separately require the migration described in their release notes.