Stateful Operators

Where the running total lives

Module 2 ended with RequestAssembler keyed by request_id, emitting one RequestRecord when a request's RequestCompleted arrives. It could only do that because it remembered the RequestStarted it had seen earlier, and "remembered" is what this module is about. GatewayMeter at the end of module 3 keeps two things: an OpenRequest per request in flight, and a running dollar total per tenant. No windows yet, and no checkpoints yet — everything here is volatile, and the last section says exactly how.

Keyed state is an embedded key/value store

Upstream's words, kept exact:

Keyed state is maintained in what can be thought of as an embedded key/value store. The state is partitioned and distributed strictly together with the streams that are read by the stateful operators. Hence, access to the key/value state is only possible on keyed streams, i.e. after a keyed/partitioned data exchange, and is restricted to the values associated with the current event's key. Aligning the keys of streams and state makes sure that all state updates are local operations, guaranteeing consistency without transaction overhead.

Against GatewayMeter: the subtask that receives request r-1093's events is the subtask holding r-1093's open entry, because the same key_by(request_id) routed both. A plain map after parse_event cannot hold keyed state; there is no key to scope it to. And while RequestAssembler processes r-1093, the entry for r-1094 does not exist as far as your code can tell. Module 2's key groups are how this store is redistributed; module 10 returns to them.

The etl page adds the sizing rule you will need by the end of this page: "Whenever the key space is unbounded, then so is the amount of state Flink will need." Tenants are a bounded key space. Request IDs are not.

The five primitives

The state page lists them — the table uses its Java names; the PyFlink spellings are in the API reference — and all are "scoped to the key of the current input element," and all have clear().

You never construct one of these. You construct a StateDescriptor — name plus type, "and possibly a user-specified function, such as a ReduceFunction" — and ask the RuntimeContext for a handle; names must be unique within an operator. The PyFlink pages show ValueStateDescriptor through runtime_context.get_state(...) and MapStateDescriptor through get_map_state(...); the others follow the same naming, with exact signatures in the PyFlink API reference at https://nightlies.apache.org/flink/flink-docs-release-2.3/api/python/.

One upstream sentence to keep in view: "the value you get from the state depends on the key of the input element. So the value you get in one invocation of your user function can differ from the value in another invocation if the keys involved are different." The handle is a pointer into a store whose key Flink sets before it calls you.

Declared in open(), used in process_element()

open() runs once per parallel instance, before any record, with no key in context; that is where a descriptor becomes a handle. process_element() runs per record with the record's key bound. Here is stage 3's RequestAssembler, in the shape of the upstream PyFlink KeyedProcessFunction example. The spine's state name is open; the Python attribute is open_state because open is the lifecycle method.

from pyflink.common.typeinfo import Types
from pyflink.datastream.functions import KeyedProcessFunction, RuntimeContext
from pyflink.datastream.state import ValueStateDescriptor


class RequestAssembler(KeyedProcessFunction):
    """Keyed by request_id. State: open -> the OpenRequest awaiting RequestCompleted."""

    def __init__(self):
        self.open_state = None

    def open(self, runtime_context: RuntimeContext):
        descriptor = ValueStateDescriptor("open", Types.PICKLED_BYTE_ARRAY())
        self.open_state = runtime_context.get_state(descriptor)

    def process_element(self, event, ctx: 'KeyedProcessFunction.Context'):
        if event.event_type == "RequestStarted":
            self.open_state.update(OpenRequest(
                event.request_id, event.tenant_id, event.model,
                started_ts=event.ts, prompt_tokens=event.payload["prompt_tokens"]))
        elif event.event_type == "RequestCompleted":
            started = self.open_state.value()
            if started is None:
                return  # completion with no start in state: dropped today; module 6 revisits
            self.open_state.clear()
            yield RequestRecord(
                started.request_id, started.tenant_id, started.model,
                started_ts=started.started_ts, completed_ts=event.ts,
                latency_ms=event.payload["latency_ms"],
                prompt_tokens=started.prompt_tokens,
                completion_tokens=event.payload["completion_tokens"],
                status=event.payload["status"])

Types.PICKLED_BYTE_ARRAY() is what the upstream Python examples use for an arbitrary Python object. The clear() on completion is not tidiness: it is the only thing keeping this operator's state proportional to requests in flight rather than requests ever seen. Hold that thought.

The second operator consumes RequestRecord, keyed by tenant_id, and keeps one float per tenant. The price table is a constant this module — the spine's rule is that it is never fetched inside a keyed operator's process_element; broadcast state below is the alternative, and module 9 handles the per-tenant negotiated price.

PRICE_PER_1K = {  # model -> (input_usd, output_usd) per 1,000 tokens; a constant this module
    "model-a": (0.0025, 0.0100),
    "model-b": (0.0005, 0.0015),
}


def usd_for(record) -> float:
    input_usd, output_usd = PRICE_PER_1K[record.model]
    return (record.prompt_tokens * input_usd + record.completion_tokens * output_usd) / 1000.0


class TenantCost(KeyedProcessFunction):
    """Keyed by tenant_id. State: usd_total -> running USD total since the job started."""

    def __init__(self):
        self.usd_total = None

    def open(self, runtime_context: RuntimeContext):
        self.usd_total = runtime_context.get_state(
            ValueStateDescriptor("usd_total", Types.DOUBLE()))

    def process_element(self, record, ctx: 'KeyedProcessFunction.Context'):
        total = self.usd_total.value() or 0.0
        total += usd_for(record)
        self.usd_total.update(total)
        yield record.tenant_id, total

The wiring in build_job(env), with operator uids from here on — upstream's savepoints page calls setting them "highly recommended," and module 10 shows the cost of one missing:

records = events.key_by(lambda e: e.request_id).process(RequestAssembler()).uid("assemble")
totals = records.key_by(lambda r: r.tenant_id).process(TenantCost()).uid("tenant-cost")
totals.print()

Module 4 replaces the running total with a 1-minute window; the spine's TenantCost(AggregateFunction) is that version. Where the number lives does not change.

Per operator, per key — never shared

TenantCost cannot read RequestAssembler's open state, for two stacked reasons. It is a different operator: a descriptor named open in TenantCost would create a second, empty, unrelated store. And it is a different key: TenantCost runs with a tenant_id bound, and open is scoped by request_id. The broadcast-state page states the rule underneath: "in Flink there is no cross-task communication." The only channel between operators is a record — which is why RequestRecord carries tenant_id, model and both token counts. Everything the total needs has to travel in the record.

The same wall stands between keys of one operator: processing tenant acme's record, globex's total is not readable, writable or enumerable. Upstream's KeyedBroadcastProcessFunction has an applyToKeyedState escape hatch on the broadcast side; the same page notes it "is not supported in PyFlink yet."

Operator state, and what the Kafka source keeps

Keyed state is the state you write. Operator state is the other kind: "Operator State (or non-keyed state) is state that is bound to one parallel operator instance." Upstream's motivating example is already in your job: "Each parallel instance of the Kafka consumer maintains a map of topic partitions and offsets as its Operator State." No key could scope a partition→offset map, so it is bound to the instance.

Because it is not keyed, rescaling needs a redistribution scheme, and the state page defines two for list-style operator state: even-split, where "the list is evenly divided into as many sublists as there are parallel operators," and union, where "each operator gets the complete list of state elements," obtained through getUnionListState(descriptor). Module 10 is where that matters.

Two facts for stage 3. The state page says plainly: "Operator state is still not supported in Python DataStream API." GatewayMeter never needs it — its only operator state lives inside KafkaSource, which is Java. And that offset map is Flink's, not Kafka's. The connector page: "Kafka source does NOT rely on committed offsets for fault tolerance. Committing offset is only for exposing the progress of consumer and consuming group for monitoring." With checkpointing off, as now, the source relies on nothing at all for recovery.

Broadcast state: distributing the price table

Suppose PRICE_PER_1K changes weekly and you refuse to redeploy for it. Broadcast state is upstream's pattern: "Broadcast State is a special type of Operator State. It was introduced to support use cases where records of one stream need to be broadcasted to all downstream tasks, where they are used to maintain the same state among all subtasks." It differs from other operator state in that "it has a map format," it "is only available to specific operators that have as inputs a broadcasted stream and a non-broadcasted one," and an operator "can have multiple broadcast states with different names."

The PyFlink shape, from the broadcast-state page: declare a MapStateDescriptor("prices", Types.STRING(), Types.PICKLED_BYTE_ARRAY()), call .broadcast(descriptor) on the low-volume price stream, then records.connect(price_broadcast).process(...) — connect on the non-broadcast side, and since that side is keyed the function is a KeyedBroadcastProcessFunction. process_broadcast_element sees price updates with read-write access through ctx.get_broadcast_state(descriptor); process_element sees RequestRecords with read-only access to the same map. Read-only on the data side is the consistency mechanism: every instance applies the same broadcast elements with the same deterministic logic and ends with the same map without talking to another.

Three upstream considerations before you reach for it: broadcast elements "may arrive in a different order to each task," so updates must not depend on order; "all tasks checkpoint their broadcast state," so a checkpoint carries it p times; and "Broadcast state is kept in-memory at runtime" whatever the backend — "This holds for all operator states." A table of a few hundred models is nothing. The per-tenant negotiated price is a lookup, not a table — module 9.

State TTL

"A time-to-live (TTL) can be assigned to the keyed state of any type. If a TTL is configured and a state value has expired, the stored value will be cleaned up on a best effort basis." Collections expire per entry. The PyFlink configuration, following the state page's Python tab:

from pyflink.common import Duration
from pyflink.datastream.state import StateTtlConfig, ValueStateDescriptor

ttl_config = StateTtlConfig \
    .new_builder(Duration.of_minutes(10)) \
    .set_update_type(StateTtlConfig.UpdateType.OnCreateAndWrite) \
    .set_state_visibility(StateTtlConfig.StateVisibility.NeverReturnExpired) \
    .build()

descriptor = ValueStateDescriptor("open", Types.PICKLED_BYTE_ARRAY())
descriptor.enable_time_to_live(ttl_config)

The duration is mandatory; the two optional settings are shown at their documented defaults. The update type says when the clock restarts: OnCreateAndWrite (default) "only on creation and write access"; OnReadAndWrite "also on read access." The visibility says what a read returns for an entry past its TTL that has not been removed yet: NeverReturnExpired (default), where "the expired state behaves as if it does not exist anymore, even if it still has to be removed," or ReturnExpiredIfNotCleanedUp. So "visible" and "deleted" run on different clocks: with the default, open_state.value() returns None the moment the entry is older than ten minutes, while the bytes may sit in the backend a while longer. The same page warns that NeverReturnExpired disables the state read/write cache, "which will cause some performance loss in PyFlink."

When do the bytes go? "By default, expired values are explicitly removed on read, such as ValueState#value, and periodically garbage collected in the background if supported by the configured state backend." The background part is per backend (as of 2026-09; see the upstream state page). Heap: an incremental sweep that advances on each access and by default "checks 5 entries without cleanup per record processing." RocksDB: a compaction filter that drops expired entries during compaction, querying the current time "each time 1000 entries have been processed," with periodic compaction defaulting to 30 days so rarely touched files still meet the filter. The sentence that matters below: "If no access happens to the state or no records are processed, expired state will persist."

Costs: the backend stores a last-modification timestamp per value — on RocksDB "8 bytes per stored value, list entry or map entry," on heap an extra object holding a long. Only processing-time TTL exists. And "The TTL configuration is not part of check- or savepoints" — it is a property of the running job.

The failure: requests that never complete

A client disconnects mid-stream. The gateway crashes between RequestStarted and RequestCompleted. A load balancer kills a hung tool call and never tells the gateway. Each leaves an OpenRequest in open with no completion coming, and since clear() runs only on completion, nothing removes it. Nothing reads it either: value() is called only when a later event for that request_id arrives, and none will. The on-read removal path is dead for exactly the entries that need it; only background cleanup can reach them.

The arithmetic, with the lab's figures. Bytes per entry is an estimate — pickled OpenRequest, serialized key, backend overhead — and exercise 3.3 has you measure it.

Nine gigabytes a year sounds survivable until you recall that the default backend keeps it as objects on the JVM heap of four TaskManagers, and that every checkpoint from module 7 on carries it. The tenant total is bounded by the number of tenants and does not grow with traffic at all — the bounded-versus-unbounded key space, in dollars of memory.

Two bounds fix it. The first is the TTL above on open, and it is blunt: choose it longer than any legitimate request, or a slow agent run's completion finds value() returning None, is dropped as a completion without a start, and is never billed. Ten minutes is a guess; your gateway's p99.9 request duration is the number. Does cleanup keep up? On heap the sweep is driven by record processing — 5 entries per record, several records for each of 5,000,000 requests a day — so a 100,000-entry backlog is traversed many times over; on RocksDB it waits for compaction. The second bound is precise: a per-key timer registered at RequestStarted that fires five minutes of event time later, emits a RequestRecord with status="abandoned", and clears the state. That is module 6's timer; it turns an abandoned request from a leak into a record the invoice can see. This module only names it. Once the timer exists, module 6 lengthens this TTL to a backstop measured in hours: a processing-time TTL shorter than a watermark stall would delete an open request before its event-time timer could fire.

State backends: same API, different bytes

Nothing in RequestAssembler or TenantCost changes when the backend does: "State backends can be configured without changing your application logic." What changes is where a handle's bytes live.

"If nothing else is configured, the system will use the HashMapStateBackend," which "holds data internally as objects on the Java heap"; the ceiling is TaskManager memory. EmbeddedRocksDBStateBackend "holds in-flight data in a RocksDB database that is (per default) stored in the TaskManager local data directories," as "serialized byte arrays," so "the amount of state that you can keep is only limited by the amount of disk space available" — at the price of (de)serialization on every access, which upstream puts at "an order of magnitude slower than the memory state backends." ForStStateBackend is the third, "still in the experimental stage" as of 2026-09: an LSM store built on RocksDB that keeps its files on a remote filesystem, and the only backend supporting asynchronous state access. Broadcast and operator state stay in memory under all three.

Selection is one key, state.backend.type, with values hashmap, rocksdb or forst:

config = Configuration()
config.set_string('state.backend.type', 'rocksdb')
env = StreamExecutionEnvironment.get_execution_environment(config)

Module 7 puts RocksDB under GatewayMeter when checkpoints arrive, because it is the backend with incremental checkpoints; module 11 takes up ForSt and disaggregated state. For this lab, hashmap makes the leak visible in a heap gauge.

State V2, sidebar. Flink 2.x also ships "Working with State V2," an API for asynchronous access: enableAsyncState() on the KeyedStream, descriptors from the v2 package, and async methods on every primitive — asyncValue() returns a StateFuture chained with thenAccept/thenApply. It exists for ForSt, where state may be a network hop away. The local docs carry Java tabs only, so this course does not use it; the shape, once:

// Java — State V2; no PyFlink tab as of 2026-09
sum.asyncValue().thenAccept(current -> { /* ... */ sum.asyncUpdate(next); });

If a TaskManager dies right now

If a TaskManager dies right now, which numbers survive, which are recomputed, and which could be emitted twice?

Everything above is a store that lives in a process. Upstream's persistence story begins "Flink implements fault tolerance using a combination of stream replay and checkpointing," with the warning module 1 made you read: "By default, checkpointing is disabled." Stage 3 has not enabled it. A TaskManager holding one of TenantCost's four subtasks dies; the job fails or, with a restart strategy, restarts every task with empty state — including the three subtasks on TaskManagers that did not die, because there is no snapshot to give them their state back.

Survives: nothing inside Flink. Every open entry and every usd_total is gone. What survives is what already left the job — lines on stdout, since stage 3 has no sink.

Recomputed: whatever the source re-reads. Its partition→offset map is operator state and is gone too, so KafkaSource starts from its configured starting offsets, not from where it was. open entries are rebuilt only for requests whose RequestStarted is read again.

Emitted twice or never: usd_total restarts at 0.0 for every tenant. If the source restarts from the beginning of the topic, every total is recomputed and every (tenant_id, total) line is printed a second time. If it restarts from the latest offsets, every dollar accumulated before the crash is gone and the invoice starts over — never-billed revenue. No setting of the source makes this stage right; only a snapshot does.

Try it

Go to labs/03-stateful-operators/. The stage-3 gatewaymeter/ is the starting point; the compose cluster and the module-1 event generator are unchanged.

• 3.1 Locate assemble's subtasks in the Web UI and record their TaskManagers and heap use before and after a fixed traffic run. Treat heap as a proxy, not a direct keyed-state size metric.
• 3.2 Extend the deterministic oracle test to leave 10,000 request IDs open; assert that it retains 10,000 entries. This isolates the leak without inferring bytes per entry from JVM heap.
• 3.3 Compare where hashmap and rocksdb backends keep state and why switching backends does not change the RequestAssembler API. Check the lab's note about checkpointing before attempting the RocksDB comparison.
• 3.4 Choose a cleanup bound from the longest legitimate request, a plausible watermark stall and diagnostic retention. Explain why a ten-minute processing-time TTL is only a backstop for the five-minute event-time abandonment rule introduced in module 6.

Lab page: lab-03-stateful-operators.html — the exercises, every source file, and the run-it-locally steps.