Skip to content

State and timers

Handlers receive a TurbineState when their handler method accepts a state parameter. On a subscription with partition_key=, a class handler can define process_key(key, rows, state) instead of process(batch, state): the batch is grouped per key and state arrives already scoped on it — the common case needs no explicit scoping at all. The primitive underneath is state.scope(key), for state associated with one value of the subscription's partition_key: for example state.scope("tenant-7") when partition_key="tenant_id".

Upgrading from 0.2.0a2 or earlier: persisted timers are not readable

This version stores timers in a new on-disk format and ships no migration path. It concerns any app with persistent windows (PersistentTumbling / PersistentSliding / PersistentSession) or with timers pending in state — not only users of custom timers. An app booting over old state halts the affected partition at startup with a fatal error naming this remedy:

  1. Stop the app.
  2. Delete its local state directory and purge its checkpoint bucket (a restored old checkpoint fails the same way).
  3. Restart and replay the input.

In a cluster there is no rolling upgrade: an old node restoring a new-format snapshot fails the same way, so every node must run the new version before new state is written.

Scoped state

Per-key handlers: process_key

On a subscription with partition_key=, a class handler can define process_key instead of process:

@app.subscribe(kafka.topic("tenant-activity"), partition_key="tenant_id")
class Inactivity:
    def process_key(self, key: str, rows: RecordBatch, state: TurbineStateView) -> None:
        state.put("status", "active")  # already scoped on `key`

The runtime groups each batch by the value of the partition_key column and calls process_key once per distinct key, in first-appearance order:

  • key is the str form of the partition-key value (a null groups under "", matching the on-disk encoding); state.scope_key == key.
  • rows is the sub-batch of that key's rows, in their original relative order. The interleaving across keys is not replayed: all of one key's rows are delivered before the next key's.
  • state is the TurbineStateView for that key — the same object state.scope(key) would return, without the per-row scope(...) calls. Timers registered here reach on_timer with this same view.
  • Non-None RecordBatch outputs are concatenated into a single batch.

process_key requires partition_key= on the subscription, and a class defines exactly one of process / process_key (declaring both is an error). It suits row-oriented per-key logic; whole-batch handlers — Arrow compute over mixed keys, cross-key reads, windowing primitives — keep process and use scope directly. The optional ctx parameter comes last: process_key(key, rows, state, ctx).

Three boundaries to know:

  • Key types: only string, int32 and int64 partition-key columns are accepted — the same types the windowing layer encodes, so both always agree on the on-disk key. Anything else is refused at @subscribe when the schema is declared, else on the first batch.
  • Windows follow the view: a process_key handler can drive persistent windows with its scoped state — the view is already the scope of the rows' key, so window.update(rows, state) writes exactly what the root store's scope(key) would (closes still fire with the root store). Handing a window a view whose scope_key does not match the rows' key raises a TypeError. View-driven ingestion takes the batched dict path: the Arrow-native merge stays on the root store.
  • Failure isolation is per batch, not per key: if one key raises, the whole batch — healthy keys included — follows the subscription's on_processing_error policy.

The scope primitive

scope returns a TurbineStateView, exported from turbine. It supports the same get, put, delete, byte and batched operations as the parent store, plus timer registration and cancellation. Both the parent and the view expose read-only metadata:

  • worker_id: the owning worker, usable with WorkerLocal.get(state) from process / process_key and from timer callbacks.
  • scope_key: the exact key passed to scope, including an empty string; None on the parent store. Callbacks can identify their tenant without duplicating it in user state or in the timer name.

A view has no nested scope; use the parent TurbineState to select a scope.

from turbine import TurbineState, TurbineStateView


def remember(state: TurbineState, tenant_id: str) -> None:
    tenant: TurbineStateView = state.scope(tenant_id)
    tenant.put("status", "active")
    # Another tenant's "status" is a separate entry.

Annotations on local views are optional: a type checker infers TurbineStateView from state.scope(...). This typing support describes the existing runtime API.

The parent store and each scope have separate keys: state.get("status") does not read state.scope("tenant-7").get("status"). Moving existing data from the parent store into a scope requires an application migration; adding type annotations alone does not change storage.

Per-key timers

Register timers through the scope matching the input's partition-key value. The callback reaches the handler that processes that key, including with parallelism > 1. SDK windowing primitives manage this routing automatically.

Operation Meaning
view.register_timer(id, delay_ms) Fire after a wall-clock delay in milliseconds.
view.register_event_time_timer(id, fire_time_ms) Fire when the partition watermark reaches an absolute event-time deadline in milliseconds. Requires an event-time source on the subscription.
view.cancel_timer(id) Cancel through the same scope used to register the timer.

Timer names are local to their scope. Both state.scope("a") and state.scope("b") can register "alarm"; cancelling one does not cancel the other. Registering the same name again in the same scope replaces its deadline, including when changing time domains. The runtime treats your name as opaque.

on_timer(timer_id, state) receives the original local name and the same scope that registered it. Annotate this argument as TurbineStateView: call get, put, cancel_timer or register_timer directly. SDK window callbacks such as on_close still receive the parent store.

For a single callback, implement on_timer as in the complete example below; no base class is required. For callbacks selected by a local-name prefix, inherit from TimerHandler to get a typed register_timer_handler method, including in __init__. It takes precedence over on_timer for matching custom timers; the first registered matching prefix wins. The callback receives the local name and registering store or view, and may return a RecordBatch or None. An unmatched custom timer falls back to on_timer when present. These registrations are separate from SDK window callbacks, even when their prefixes overlap; on_timer is optional when registered callbacks cover all your timers.

from turbine import RecordBatch, TimerHandler, TurbineState, TurbineStateView, WorkerLocal


class Inactivity(TimerHandler):
    def __init__(self) -> None:
        self.resources: WorkerLocal[object] = WorkerLocal(object)
        self.register_timer_handler("alarm:", self.on_alarm)

    def process(self, batch: RecordBatch, state: TurbineState) -> None:
        self.resources.get(state)
        for tenant_id in set(batch.column("tenant_id").to_pylist()):
            state.scope(tenant_id).register_timer("alarm:idle", delay_ms=60_000)

    def on_alarm(self, timer_id: str, state: TurbineStateView) -> None:
        self.resources.get(state)  # Same worker resource as in process.
        print(f"{state.scope_key}: {timer_id} on {state.worker_id}")

Register this class with app.subscribe(...) like any other handler. TimerHandler needs no super().__init__() call. Registering the same prefix again replaces its callback without changing its priority. Older handlers using the dynamically supplied method continue to work; inheriting from TimerHandler also makes the method visible to type checkers and editors.

The scope and local name survive restarts, as does the timer's routing. Registration, cancellation and user state changes are committed with the batch. For crash behavior and the known window-close limitation, see Delivery guarantees.

Example: per-tenant inactivity

Create a one-partition tenant-activity topic on your broker and save this app as scoped_timers.py. It expects JSON records such as {"tenant_id": "tenant-7", "active": true}. Run it with uv run python scoped_timers.py, then produce records. An active event resets a 60-second timeout; an inactive event cancels it. With more than one Kafka partition, the producer must partition records by tenant_id.

import pyarrow as pa

from turbine import KafkaBroker, RecordBatch, Turbine, TurbineStateView


Activity = pa.schema([
    pa.field("tenant_id", pa.string(), nullable=False),
    pa.field("active", pa.bool_(), nullable=False),
])


kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(kafka, app_id="scoped-timers")


@app.subscribe(
    kafka.topic("tenant-activity", schema=Activity),
    partition_key="tenant_id",
    parallelism=3,
)
class Inactivity:
    def process_key(self, tenant_id: str, rows: RecordBatch, tenant: TurbineStateView) -> None:
        # `tenant` is already scoped on this key — no state.scope(...) call.
        timer_id: str = "alarm"
        for row in rows.to_pylist():
            if row["active"]:
                tenant.put("last_event", row)
                tenant.register_timer(timer_id, delay_ms=60_000)
            else:
                tenant.cancel_timer(timer_id)
                tenant.delete("last_event")

    def on_timer(self, timer_id: str, tenant: TurbineStateView) -> None:
        # The runtime restores the registering scope, even after a restart.
        last_event = tenant.get("last_event")
        tenant.delete("last_event")
        print(f"Timer {timer_id}: tenant {tenant.scope_key} inactive (last event: {last_event})")


if __name__ == "__main__":
    app.run()

Each tenant registers the same name, "alarm". The callback receives that name and the matching view, so it can read "last_event" without parsing the tenant from the timer name. Cancellation uses the same local name and scope.

Advanced: partition timers

For a subscription without a partition_key, a timer registered directly on state can flush a buffer or publish a count for its Kafka partition. Its callback receives the parent TurbineState, with scope_key=None. Each partition has its own timer names and state: a timer named "flush" on a three-partition input can fire independently in all three partitions. This is per-partition scheduling, rather than an application-wide scheduler.

With parallelism > 1, such a timer fires on shard 0 only. On a subscription with a partition_key that changes what the callback can see: each shard caches the keys it owns, so a callback running on shard 0 reads and writes through shard 0's cache while the shard actually owning a key keeps serving its own cached value — an update made in the callback can be lost without any error. The runtime logs a one-time warning when a root timer is registered in that configuration. For a deadline associated with a tenant or another key value, always use state.scope(key).register_timer(...), which routes the callback to the shard that processes that key.

For example, create partition-timer-input and partition-timer-counts topics, save this app as partition_timers.py, and run uv run python partition_timers.py. Produce JSON records to the input topic. For each active partition, the app publishes the accumulated count ten seconds after its first batch; later batches add to the count without postponing the timer.

from turbine import KafkaBroker, RecordBatch, Turbine, TurbineState

kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(kafka, app_id="partition-counts", from_earliest=True)


@app.subscribe(
    kafka.topic("partition-timer-input"),
    output=kafka.topic("partition-timer-counts"),
    partition_key=None,
)
class PartitionCounts:
    def process(self, batch: RecordBatch, state: TurbineState) -> None:
        pending: int = state.get("pending") or 0
        state.put("pending", pending + batch.num_rows)
        if pending == 0:
            state.register_timer("flush", delay_ms=10_000)

    def on_timer(self, timer_id: str, state: TurbineState) -> RecordBatch:
        count: int = state.get("pending") or 0
        state.delete("pending")
        return RecordBatch.from_pylist([{"worker_id": state.worker_id, "count": count}])


if __name__ == "__main__":
    app.run()

The timer is armed by the first batch, fires once, and is armed again only when a new batch arrives after that flush. Use per-key timers for deadlines associated with a tenant or another partition-key value.

Updating existing custom timers

Who is affected, in one read:

  • Window users (PersistentTumbling / PersistentSliding / PersistentSession): nothing to change — the SDK scopes window state and routes its timers itself. The upgrade warning about persisted state still applies.
  • State without timers (state.get / state.put on the root store): unchanged; scope stays optional.
  • Root timers with parallelism=1 (state.register_timer("flush", ...)): unchanged — same name, root store in the callback.
  • Per-key custom timers (one per tenant, per user, …): use scope, as shown in Per-key timers. It matters as soon as parallelism > 1.

If custom timers previously embedded the tenant in their ID, that duplication is no longer needed:

# Before: manual identification in the timer ID.
state.scope(tenant_id).register_timer(f"idle:{tenant_id}", delay_ms=60_000)

# Now: the scope identifies the tenant.
state.scope(tenant_id).register_timer("alarm", delay_ms=60_000)

Adapt on_timer to use its TurbineStateView directly: remove ID parsing and calls to state.scope(...) inside that callback. Keep parent-store annotations on process and SDK window callbacks. Read scope_key to identify the tenant; worker_id and WorkerLocal.get(view) remain available in scoped callbacks. For typed prefix registration, inherit from TimerHandler.

Existing patterns to review, because they change without an error or now fail:

Existing pattern Behaviour with this version
state.register_timer(f"idle:{tenant}", ...) + timer_id.removeprefix("idle:") in on_timer Still runs, but fires on shard 0: with parallelism > 1 the key's shard affinity is lost, silently (a one-time warning is logged). Migrate to state.scope(tenant).register_timer("idle", ...).
state.scope(t).register_timer(...) + state.scope(...) inside on_timer Breaks: the callback now receives the registering view, which has no nested scope. Use the view directly; its scope_key identifies the tenant.
on_timer parsing a scope out of the timer ID The callback now receives the local name only.
Cancelling from another scope, or from the root, a timer registered in a scope No-op: names are local to their scope. Cancel through the same scope.
register_timer_handler on a window prefix to intercept its closes Intercepts nothing: user and window callback registries are separate.
Hand-written on_timer_batch User timers now arrive grouped by scope, one call per scope with its view.
Existing state holding any pending timer (windows included) Rejected at boot (new timer format): see the upgrade warning at the top of this page.