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.

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.

Raft cluster (dynamic)¶

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

Bootstrap mode — start the initial cluster with a known peer list:

kafka = KafkaBroker(bootstrap="kafka:9092")
app = Turbine(
    kafka,
    raft_node_id=1,
    raft_listen_addr="http://10.0.0.1:8400",
    raft_peers=[(1, "http://10.0.0.1:8400"),
                (2, "http://10.0.0.2:8400"),
                (3, "http://10.0.0.3:8400")],
)

Join mode — add a node to a running cluster:

kafka = KafkaBroker(bootstrap="kafka:9092")
app = Turbine(
    kafka,
    raft_node_id=4,
    raft_listen_addr="http://10.0.0.4:8400",
    raft_seed="http://10.0.0.1: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"
# or for join mode:
export TURBINE_RAFT_SEED=http://10.0.0.1:8400

Automatic failover¶

In Raft mode, the cluster monitors node health. If a node stops responding for longer than the liveness grace period (default: 60 seconds), its partitions are automatically redistributed to the surviving nodes. When the node comes back, it is re-included and partitions are rebalanced.

No manual intervention required.

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", ...)

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 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`	Cannot participate at all: no Raft leader elected, or this node is not (yet) a voter.	Investigate connectivity/quorum.

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: no_raft_leader, not_a_voter, halted_workers, processing_errors, checkpoint_store_failing. 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. 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{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. First confirm the previous owner is really gone — its process is dead, or it has left the cluster's voter list. This matters: if it is alive and mid-transaction, releasing the partition by hand causes the very data loss the wait prevents.

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

curl -X POST http://10.0.0.1:8400/cluster/force_release \
  -H 'Content-Type: application/json' \
  -d '{"topic": "events", "partition": 3}'

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}

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

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`	Total batch time (all phases)

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` (Pydantic)	`kafka.topic(...)` input	Avro is faster to decode than JSON at scale. Pydantic-schema-aware JSON is faster than schema-inferred JSON.
`InMemoryTumbling` / `InMemorySliding`	`turbine.windowing`	Removes per-event state-store cost when the workload tolerates losing in-flight window state on crash — see In-memory windows.
`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¶

To upgrade a node without downtime:

If upgrading the current leader, transfer leadership first:
```
curl -X POST http://any-node:8400/cluster/transfer_leader
```
The leader picks another voter automatically and steps down. Verify the transfer completed before proceeding:
```
curl http://any-node:8400/health  # `leader_id` should name the new leader
```

Drain the node — remove it from the voter set so its partitions migrate to other nodes:

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

Upgrade and restart the node in join mode:

export TURBINE_RAFT_NODE_ID=3
export TURBINE_RAFT_LISTEN_ADDR=http://10.0.0.3:8400
export TURBINE_RAFT_SEED=http://10.0.0.1:8400
python app.py

Verify the node is healthy and receiving partitions:

curl http://10.0.0.3:8400/ready        # 200 + "status":"healthy"
curl http://10.0.0.3:8400/assignments

Repeat for the next node.