Kafka & Partitioning

This page covers the Kafka partitioning constraints Turbine imposes for stateful workloads, the role of partition_key for sub-partition parallelism, and repartition_by= for grouping on a column the producer didn't key by.

Topic Partitioning

Rescaling only works toward a partition count that divides 960

Rescaling means changing the number of partitions of a topic, typically adding partitions to spread a growing load over more workers. For a subscription that keeps state (its handler takes a state parameter, or it declares a window) and sets a partition_key, Turbine can only rescale to a partition count that divides 960, and only if the producer places records the way Turbine expects (Producer placement). Rescaling to any other count breaks the subscription's results:

• Keys lose their state. turbine-cli rescale, which carries each key's state to its new partition, refuses such a count: every key that changes partition starts again from nothing — counters, open windows, timers.
• Windows are published twice. The partition a key left still closes the windows that were open, with the data it had; the new partition publishes the same windows with the rest. The disk space left behind is never reclaimed.

That is why the app refuses to start when such a subscription reads a topic on any other count. (Without a partition_key, a stateful subscription cannot be rescaled at all.) Running on one costs nothing as long as it is never rescaled, and rescaling from it to a valid count works: to start anyway, set TURBINE_ALLOW_NON_DIVISOR_PARTITIONS=1 — see Starting on another count.

Supported partition counts

The 28 counts that divide 960:

Range	Valid partition counts
Small (dev, single-broker)	1, 2, 3, 4, 5, 6, 8, 10, 12
Medium (3-6 broker clusters)	15, 16, 20, 24, 30, 32
Large (high-resource deployments)	40, 48, 60, 64, 80, 96, 120
Very large (1 partition per Turbine worker thread)	160, 192, 240, 320, 480, 960

Common Kafka choices that are not valid: 7, 9, 11, 13, 17, 100, 128, 256, 512. The startup error names the closest valid counts.

Which topics this covers:

• The input topic of a keyed stateful subscription. A stateless subscription keeps no state, so any count works. A stateful one without partition_key cannot be rescaled at all (see Rescaling a topic), so its count does not matter either.
• The internal topics Turbine creates for repartition_by and combine_by. They take the source topic's count unless you set repartition_partitions= / combine_partitions=, so a source topic on an invalid count is reported there too, with the argument to set.

Producer placement

Rescaling and the startup cleanup assume one more thing about the producer: it keys each record by its partition_key value and places it with murmur2 — the default partitioner of Kafka's Java client, and of most clients. The record key must be that value as UTF-8 text, an integer column in decimal ("42"): a Java producer that serialises an integer key as binary (LongSerializer, IntegerSerializer) places it by other bytes, and reads as misplaced. Clients built on librdkafka (confluent-kafka, for instance) default to a different hash: set their partitioner to murmur2_random. Turbine's own outputs keyed with message_key are placed with murmur2, so one Turbine app can feed another.

Turbine cannot tell from one partition's records which hash the producer uses — a key it saw once may be one murmur2 would send here, or one another hash did. So the startup cleanup runs only on your word:

@app.subscribe(kafka.topic("events"), partition_key="tenant_id", partition_hash="murmur2")
def per_tenant(batch, state): ...

• Declared (partition_hash="murmur2"): after a partition-count change, each partition cleans up at startup the state of the keys it no longer receives — the keys that moved.
• Undeclared (the default): nothing is ever deleted on the strength of what Turbine observed. After a partition-count change without turbine-cli rescale, the state of the keys that moved stays where it was — a startup log line says so — and the windows still open on it may be published once more, with partial data, by the partition they left.

turbine-cli rescale is an explicit operator action and needs no declaration: running it is the same statement.

The declaration is checked, not trusted blindly. Each partition of a keyed stateful subscription samples its records for one that murmur2 would not have put there, and publishes what it saw as the turbine_partition_key_placement gauge:

• -1 — misplaced: a record sits where murmur2 would not have put it, under the topic's current partition count and every earlier one this state ran under. Durable: it is recorded with the state, and it stops both the cleanup and turbine-cli rescale for good, declaration or not. A later stretch of well-placed records does not lift it, because it says nothing about the state the misplaced ones already wrote — which is exactly what the cleanup would delete. Once the producer is fixed, its keys move partition anyway, so the way back is to rebuild the state: stop the app, delete its state directory and checkpoints, and replay the topic.
• 1 — looks right: enough distinct keys have been seen exactly where murmur2 puts them (40 distinct keys on a 2-partition topic, 20 on 4, 5 on 960). A diagnostic for you — it authorises nothing, and volume does not count: one busy key repeated is one key.
• 0 — nothing to say yet: a new app, a topic with very few distinct keys, or the window that follows a partition-count change (below).

The check samples 64 rows per batch and asks the broker for the partition count only on a suspicion; it costs nothing measurable. It catches a wrong declaration on any key it samples, and cannot catch one on a key it never sees — a declaration is your assertion about the producer, not something Turbine can prove.

A partition-count change is not a misplacement — for a while. When a partition first sees a new count (at start-up, or from the broker while running), it opens a transition window. Producers get 10 minutes from that moment to pick the change up (Kafka clients refresh their metadata every 5 minutes by default); when they are up, the partition notes where its log ends, and every record before that point may still be placed under an earlier count. The window is bounded by positions in the log, never by record timestamps, which a producer may set to anything. Turbine records the partition counts each state ran under, a single partition included, and keeps them across restarts, so it recognises such records — but only inside the window:

• Inside the window records placed under an earlier count are passed over, records placed correctly do not count either (while an earlier count can excuse a misplaced record, a run of well-placed ones says nothing), and the cleanup does not run, even declared: the check is blind to excused records, so a wrong declaration gets its chance to be caught first.
• After the window no earlier count excuses a record. A record still placed under the old count is a misplacement: the worker asks the broker for the current count once more and, if it has not changed, the partition goes to -1. The window closes when the stream has moved past it — once the partition has read past the log end it noted — never by the clock at a start-up: a partition restarted with an hour of lag notes the end of its log as it is then, so its backlog is still read inside the window. A declared partition is cleaned up at the first start-up after that; the start-up that first sees the new count holds the cleanup too, before the window is even recorded.

So a producer must place records under the new count within 10 minutes of the change. One that takes longer is treated as misplacing — the safe direction: it loses the cleanup, never its state — and the way back is the same as for any -1.

On a record that looks misplaced while the app is running, the worker asks the broker for the current count before concluding; if it changed, a window opens and the same batch is judged again under the new count.

Null keys are never cleaned up. A record whose partition_key is null goes to a random partition, with every Kafka client, while its state always lands in the same key group. That group therefore holds live state on every partition at once, so the cleanup keeps it everywhere — and the same records are no evidence about the producer either way.

Rescaling a topic

Adding Kafka partitions changes which partition each key goes to. For a keyed stateful subscription, the state of every key that moves has to move with it, and that is what rescaling Turbine's state does. There are two ways to change the partition count, both toward a count from the table above:

• With turbine-cli rescale: stop the app, run it against the app's checkpoints with the old and new counts, change the topic's partition count, restart. Each key's state is waiting on the partition that now receives it — windows, counters and timers carry on as if nothing happened.
• Without it: change the partition count and restart. A key that changed partition starts again from empty state on its new partition, and — if the subscription declares partition_hash="murmur2" (Producer placement) — each worker drops the state of the keys it no longer receives at the first start-up after it has read everything appended during the 10 minutes that follow the change. Keys that stayed put keep their state.

turbine-cli is built from a source checkout — the published wheel does not ship it. turbine-cli rescale is an offline state migration: you run it against a stopped app's checkpoints, not against a live node.

A stateless subscription has nothing to move: any partition count, and any change of it, works.

For a cluster that publishes Raft-approved checkpoints, turbine-cli rescale refuses the history. Recovery choices live in the cluster and the offline tool only reads the checkpoint store.

A stateful subscription without partition_key cannot be rescaled. Turbine only knows which key a piece of state belongs to through that column, so without it the state of a partition is one indivisible block: after a partition-count change, every key that moved starts again from empty state and what it left behind is never cleaned up — the same failure as a count that does not divide 960, on every count. turbine-cli rescale refuses such a subscription. If a topic may ever be rescaled, declare its partition_key from the first run.

An exactly-once stateful subscription cannot be rescaled either: its committed timer output is chained to the sink receipt of its current partition layout, and that lineage cannot be transferred to a new one. turbine-cli rescale detects the enrolled sink cursor in the checkpoints and refuses before uploading anything. To change the partition count of an exactly-once subscription, start it under a new app_id at the target count (a new processing history — see Delivery Guarantees).

Starting on another count

When a subscription is refused, the app stops before it starts processing and names each topic, its count and the closest valid counts. Two ways forward:

• Change the count to one of the table above. This is the right answer for anything headed to production, since it keeps rescaling open.
• Set TURBINE_ALLOW_NON_DIVISOR_PARTITIONS=1 to start anyway. It is meant for evaluating Turbine on an existing topic you cannot re-partition yet, or for a topic that will never be rescaled (a few hundred partitions nobody plans to resize). Every startup then repeats the warning next to the topics concerned.

Leaving an invalid count later is a rescale like any other: toward a valid count, it works with or without turbine-cli rescale.

How it works

This part explains the constraint; you do not need it to use Turbine.

Turbine splits the state of a keyed subscription into 960 key groups. A key's group is a hash of its partition_key value modulo 960, and that hash is murmur2 — the one Kafka's default partitioner uses to pick a partition, murmur2(key) % N for a topic of N partitions.

Partition p owns the groups p, p + N, p + 2N, … — every N-th group. When N divides 960, the groups a partition owns hold exactly the keys Kafka delivers to it, because (h % 960) % N == h % N. When it does not, the two stop matching. With a key hashing to h = 1000:

	Kafka sends the key to partition 1000 % N	Its group 1000 % 960 = 40 belongs to partition 40 % N
N = 4 (divides 960)	0	0 — the same worker
N = 7 (does not)	6	5 — another worker

Both ways of rescaling rely on that match:

• turbine-cli rescale moves whole groups: under the new count, each group has exactly one owner, and it is the partition Kafka will send that group's keys to.
• Without it, a worker restarting on its local state drops the groups it no longer owns, which are precisely the keys that moved elsewhere.

On a count that does not divide 960, a partition receives keys from every group, so "the groups it owns" no longer says which keys it holds. turbine-cli rescale has no owner to give each group, and dropping the other groups would delete live state, so Turbine skips that cleanup — hence the windows published twice.

All of this also assumes Kafka placed each key with murmur2 in the first place. Under another hash, a partition's keys are spread over every group — the same situation as a count that does not divide 960 — which is why the cleanup runs only on a partition_hash declaration, and why both the cleanup and the rescale stop once a misplaced record is seen (Producer placement).

Repartitioning a Stream (repartition_by)

A stateful subscription can only group correctly on a column when all records sharing a value of that column arrive at the same worker — which is true when the producer keyed the topic by it, and false otherwise. repartition_by= closes that gap:

app = Turbine(app_id="fraud")  # explicit app_id required
kafka = KafkaBroker(bootstrap="localhost:9092")

@app.subscribe(kafka.topic("transactions"), repartition_by="user_id")
def per_user(batch, state):
    ...  # every record of a given user_id arrives here together,
         # even though "transactions" is keyed by tenant_id

Declaring repartition_by="user_id" inserts a re-key hop in front of your handler: Turbine re-emits each record onto an internal topic, keyed and placed by the user_id value (using the same hash as Kafka's standard Java partitioner), and your handler consumes that internal topic instead. Records keep their original bytes and their original broker timestamps — event-time windows and with_kafka_timestamp behave exactly as they would on the source. The subscription's partition_key automatically follows repartition_by.

What you control:

• repartition_by="col" — the grouping column. Must exist in the decoded payload and be a string column. Rows where it is null are grouped together (they all land on one partition, like a SQL null group).
• repartition_partitions=N (optional) — partition count of the internal topic. Defaults to the source topic's count. Prefer a divisor of 960 (same rule as any stateful input, see above).

What Turbine guarantees:

• Several subscriptions declaring the same (topic, repartition_by) share one hop and one internal topic — the re-emit happens once, and each subscription remains an independent consumer of it.
• The hop inherits your delivery guarantee: if any sharing subscription is exactly_once, the hop runs exactly-once too, and end-to-end exactly-once holds across the chain (each stage commits its own transaction; readers only see committed records).
• Undecodable records follow the input topic's on_error / dlq policy at the hop, same semantics as everywhere else.

What it costs / does not do:

• One extra broker round-trip per record (produce + re-consume), plus the internal topic's storage. This is the price of a correct cross-key shuffle; there is no hidden worker-to-worker networking.
• Latency across the hop is bounded by the hop's batch cadence (and, under exactly-once, by its transaction commits) — sub-second in practice, but not zero.
• It does not retract or reorder anything: it is an identity re-emit, only the placement changes.

Operational notes. The internal topic is created automatically on the input topic's cluster, named turbine-{app_id}-rep-{topic}-by-{column} — this is why an explicit app_id is required (two apps with a default id would collide). It is a normal topic: it shows up in your broker tooling, can be inspected for debugging, and follows the broker's default retention. If it already exists with a different partition count than requested, Turbine refuses to start (changing the count would re-route every key); delete the topic deliberately or match its count. Renaming your handler does not change the internal topic or lose its position — the hop's identity comes from the topic and the column, not the handler name.

Monitoring the hop. The web console's Topology page draws the hop as a single "repartition by column" node: its input rate and lag are the upstream side (is the hop keeping up with the source topic?), while the lag shown on each downstream subscription is the downstream side (are your handlers keeping up with the re-keyed stream?). Expanding the node reveals the internal topic's real name and the per-partition output distribution — a single dominant bar means one hot key concentrates the downstream load, which no amount of extra partitions will fix. A "null keys" counter appears when rows are missing the grouping column. The same signals are exported on /metrics (turbine_repartition_*) for alerting.

One thing Turbine does not watch for you: the internal topic's disk footprint. It follows the broker's default retention, so on a high-volume stream, set a retention policy on it (or monitor it) with your broker tooling like any other large topic.

Cross-Key Aggregation (combine_by)

combine_by= solves the same problem as repartition_by= — "group by a column the producer didn't key the topic by" — with a different strategy built for windowed aggregation. Instead of moving every record across the broker, each worker pre-aggregates its own slice locally and ships only the partial aggregation state; a merge stage combines the partials per (window, key) and runs your close callback:

app = Turbine(app_id="metrics")  # explicit app_id required
kafka = KafkaBroker(bootstrap="localhost:9092")

@app.subscribe(kafka.topic("events"), combine_by="user_id", output=kafka.topic("scores"))
class per_user:
    def __init__(self):
        self._w = win.PersistentTumbling(
            self, name="risk", size_ms=60_000,
            key=["user_id"],
            value=agg.Mean("cpu_usage"),
            on_close=self.emit,
        )

    def process(self, batch, state):
        self._w.update(batch, state)   # pre-aggregates locally
        # must return None: results come out of on_close, below

    def emit(self, state, panes):
        return panes.to_batches()[0]   # published to "scores"

The handler looks exactly like the repartition_by version of the same program. What changes is the wire cost: with a million events per minute spread over ten thousand users, repartition_by re-emits a million records through the broker; combine_by ships ten thousand small accumulator states. The aggregates are identical either way.

Choosing between the two:

	repartition_by	combine_by
What crosses the broker	every record	one partial state per key per window
Best when	you need the raw rows grouped (arbitrary handler logic, joins, row-level output)	you aggregate with the built-in aggregate functions
Handler constraints	none	one tumbling window, mergeable aggregates, process() returns None
Wire cost	grows with event volume	grows with distinct-key count

They are mutually exclusive on a subscription — declaring both raises at subscribe() time. There is no case where stacking them wins: once records are re-keyed by a column, every key lives on one worker and there is nothing left for a combine stage to merge; and combining works regardless of how the input is placed, so a re-key in front of it buys nothing. Pick one per subscription.

What you control:

• combine_by="col" — the aggregation axis. Must be a string column and a member of the window's key= (a multi-column key like ["tenant_id", "user_id"] with combine_by="user_id" is fine). Null values form one group, like a SQL null group. The subscription's partition_key automatically follows combine_by; declaring a different one is refused.
• combine_partitions=N (optional) — partition count of the internal topic, which is the merge stage's scaling axis. Defaults to the source topic's count.
• Everything else is your normal subscription: decode config, error policy and batching apply to the ingesting side; output and the delivery guarantee describe your output.

What Turbine guarantees:

• Merged aggregates equal a single-stage GROUP BY over all events — the built-in aggregate functions (Sum, Count, Mean, Min/Max, Variance, First/Last, the sketches, Multi bundles of them) merge across workers without approximation beyond what the aggregate itself documents.
• A window only closes once every source partition has confirmed it has shipped its data for that window — a slow partition delays results rather than corrupting them. The delay is observable (turbine_combine_min_watermark_lag_ms names the lag).
• Under processing_guarantee="exactly_once", partials ride the same transactions as everything else: a crash and restart never loses a window's contribution and never emits a pane twice.
• Under at-least-once, a redelivered partial after a window already closed is dropped and counted (turbine_combine_late_partials_total), never merged twice into a fresh duplicate pane. As with any at-least-once pipeline, a redelivery before the close can double-count — use exactly-once when the aggregates must be exact.

What it costs / current limits:

• Results appear roughly one window-boundary after the window ends (the merge stage waits for every source partition's confirmation, which flows at window-boundary cadence) — bounded and deterministic, not immediate.
• One tumbling persistent window per handler; sliding and session windows, early-fire, and cross-window references are not combinable today.
• Custom aggregate implementations must declare themselves mergeable (IS_MERGEABLE + a merge() that is order-insensitive); Turbine refuses the subscription otherwise and names the offending aggregate.
• A source partition whose records are all dropped by your error policy never confirms progress, and the merge stage waits on it indefinitely — visible as a frozen turbine_combine_min_watermark_lag_ms.
• Not available in cluster (Raft) mode yet.

Operational notes. Same model as the repartition hop: an internal topic named turbine-{app_id}-cmb-{topic}-{window_name} is created automatically (explicit app_id required; count mismatches with an existing topic are refused; broker-default retention — watch its footprint yourself). The identity comes from the window name, so renaming the handler changes nothing, while renaming the window starts a fresh internal topic — the same contract as window state. The console's Topology page draws the pair as one "combine by column" node feeding your subscription; the turbine_combine_* metric family covers partial/marker volume, late drops and the merge stage's watermark lag.

partition_key & Sub-Partition Parallelism

TODO

Dedicated explainer needed:

• What partition_key is and why you'd use it (sharding inside a single Kafka partition for higher per-partition throughput on free-threaded Python).
• The parallelism=N knob and how records are dispatched across shards.
• The windowing constraint (every window must be keyed by the same column as the subscription) — already enforced at construction time.
• Trade-offs vs. simply increasing the Kafka topic's partition count (sub-partition parallelism is intra-process; multi-partition is cluster-wide).
• Migration path between the two when scaling up.

Keying the Output (message_key)

By default the messages a handler publishes carry no Kafka key. Set an output key when you need downstream consumers to see each entity's messages on the same partition (per-entity ordering, co-location) or when the output topic is log-compacted (compaction keeps the last message per key).

The output key is a column of the RecordBatch your handler returns. Name it on the output topic:

from turbine import KafkaBroker, Turbine

kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(app_id="scoring")

@app.subscribe(
    input=kafka.topic("transactions"),
    output=kafka.topic("scores", message_key="tenant_id"),
)
def score(batch):
    ...
    return result  # each row's `tenant_id` becomes that message's Kafka key

Behaviour and rules:

• Per-row keying. Every produced message is keyed by that row's value of the named column. Two rows with the same value are guaranteed to land on the same partition of the output topic.
• Placement is murmur2, the default of Kafka's Java client, so a downstream Turbine subscription whose partition_key is that column reads a correctly placed topic. Turbine up to 0.2.0a2 used librdkafka's default (a CRC32 hash) instead: after upgrading, each key moves once to its murmur2 partition. Its order holds on each side of the upgrade, not across it, and on a compacted topic its last record from before the upgrade stays in the former partition until you delete it.
• Type. The column must be a string (Utf8). A missing or non-string column fails fast at startup with a clear error.
• Null rows produce an unkeyed message (no key), which Kafka spreads across partitions — the documented behaviour for a row whose key value is null.
• The column stays in the message value. Keying does not strip it; tenant_id remains a field of the JSON payload as well.
• Requires an output. message_key= is meaningful only on an output= topic; it has no meaning on an input= topic and is rejected there.
• Distinct from partition_key. message_key sets the raw Kafka record key on messages you produce; partition_key is the keying axis of the stream you consume (state routing, sharding). They often name the same column but answer different questions.

This is not repartition_by

message_key keys a topic you publish to for a downstream consumer's benefit. It does not move data between Turbine workers or change how this app is partitioned — for that (grouping on a column your producer didn't key by), see repartition_by above.

Brokers as objects (input= / output=)

A subscription's endpoints are broker-scoped topic references, not bare strings. KafkaBroker is a handle to a Kafka cluster; broker.topic(name) (and broker.topic(name, message_key=...) for an output) names a topic on it:

from turbine import KafkaBroker, Turbine

kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(kafka)          # the app takes the same handle

# producer
@app.subscribe(input=kafka.topic("transactions"), output=kafka.topic("scores"))
def score(batch): ...

# sink (no output)
@app.subscribe(kafka.topic("metrics"))
def audit(batch): ...

• input is required and is the first positional argument — pass a topic reference (kafka.topic(...)), never a bare string.
• Omit output for a sink; set output=kafka.topic(name, message_key=...) to publish (and optionally key — see above).
• An app talks to one Kafka cluster: the same KafkaBroker you pass to Turbine(...) builds every input/output topic, so the address is written once. A topic built from a different handle is refused when the subscription is declared.