Operating GatewayMeter in Production

The job you do not restart on a whim

At the end of module 10, GatewayMeter runs on 32 slots with set_max_parallelism(128), a uid on every stateful operator, and a stop-with-savepoint runbook. This module covers what comes after correctness: a customer outgrows the others, one JobManager holds the job's control state, and health lives only in the Web UI. The failure is ordinary. At 09:00 tenant acme becomes 40% of the gateway's traffic.

One key cannot split

Module 2's rule, restated as a constraint: key_by(tenant_id) sends a key to a key group, and a key group belongs to exactly one subtask. Every RequestRecord for acme goes to the same tenant-cost subtask whatever the parallelism is. Rescaling moves acme's key group; it never splits it, because the key's state lives in one place.

So one subtask caps the whole job. It is not just acme's minutes that fall behind. Every assemble subtask sends 40% of its output down the channel to that one subtask. When that channel runs out of credit, assemble runs out of output buffers and stops, and module 9's chain follows: the source stops fetching, and model-latency and alerts starve too, even though they never see a tenant key. Watermarks travel in the same buffers, so every tenant-cost subtask's event time falls behind. Idle subtasks cannot fire their windows early, because a watermark only arrives behind records that are stuck in the queue.

The failure, in numbers

Assume 1,000 RequestRecords per second at peak (about 5 events each), and about 250 records per second as the most one Python tenant-cost subtask can aggregate. The job runs at parallelism 32, checkpoints every 10 s with a 60 s timeout, and uses the aligned mode from module 9. Before 09:00, acme is 70 rec/s. After 09:00 it is 400. The hot subtask is number 17.

Throughput caps at 250 / 0.42 ≈ 600 rec/s, because subtask 17 receives acme's share plus its own 1/32 of everyone else. The watermark lag grows by 0.4 s per second of wall time, on top of the 20 s bound. Checkpoint duration jumps and then holds steady: the in-flight data in front of the barrier is capped by network buffers, and subtask 17 drains it at 250 rec/s. The row the diagnosis turns on is start delay, which upstream describes this way: "If only a specific task has a long start delay, the most likely reason is data skew."

In dollars: nothing is billed wrong and nothing is billed twice. By 09:30 every tenant's last twelve minutes are unbilled, and the per-model alerts that the capstone wants within 60 s arrive twelve minutes late. This is module 9's never-billed-temporarily failure, but the cause is a key, not a service.

On-call reading order. 1) pendingRecords rising means the job is behind. 2) The Back Pressure tab: assemble is HIGH, so it is a victim. 3) Find the first task downstream that is busy: tenant-cost. 4) Compare that vertex's subtasks: one is at 1,000 ms/s busy and 31 are idle, which is skew rather than an under-provisioned operator. More slots will not help. 5) Confirm with checkpointStartDelayNanos on that one subtask. 6) Find the key: count RequestRecords per tenant for the last minute on gateway-events, outside Flink.

Two-stage aggregation

Cost is a sum, and a sum can be split. In stage A, key by (tenant_id, salt), where the salt is a stable hash of request_id modulo 8, and compute a partial (usd, requests) per 1-minute window. In stage B, key by tenant_id and add up the partials for the same window. acme's 400 rec/s become eight streams of about 50 rec/s, and stage B receives at most eight records per tenant per minute.

Stage B lines up with stage A because of two documented facts. A window result's timestamp is "the maximum allowed timestamp of the processed window, which is end timestamp - 1". And an arriving watermark first "triggers computation of all windows where the maximum timestamp ... is smaller than the new watermark", and only then "is forwarded (as is) to downstream operations". So stage A's partials reach stage B before the watermark that closes the same minute there. Upstream calls this pattern "consecutive windowed operations", and it adds no event-time latency.

import zlib
from pyflink.common import Time, Types
from pyflink.datastream import OutputTag
from pyflink.datastream.window import TumblingEventTimeWindows

SALTS = 8
late_tag = OutputTag("late-request-records", Types.PICKLED_BYTE_ARRAY())
late_part_tag = OutputTag("late-cost-parts", Types.PICKLED_BYTE_ARRAY())

def salted_key(rec) -> str:
    # stable across Python workers and restarts; built-in hash() of a str is not
    return f"{rec.tenant_id}#{zlib.crc32(rec.request_id.encode()) % SALTS}"

class EmitPart(ProcessWindowFunction):
    def process(self, key, context, results):
        usd, requests = next(iter(results))
        yield CostPart(key.rsplit("#", 1)[0], usd, requests)

class SumParts(AggregateFunction):              # same accumulator as TenantCost
    def create_accumulator(self):
        return 0.0, 0
    def add(self, part, acc):
        return acc[0] + part.usd, acc[1] + part.requests
    def get_result(self, acc):
        return acc
    def merge(self, a, b):
        return a[0] + b[0], a[1] + b[1]

ACC = Types.TUPLE([Types.DOUBLE(), Types.LONG()])

parts = records \
    .key_by(salted_key, key_type=Types.STRING()) \
    .window(TumblingEventTimeWindows.of(Time.minutes(1))) \
    .side_output_late_data(late_tag) \
    .aggregate(TenantCost(), window_function=EmitPart(), accumulator_type=ACC,
               output_type=Types.PICKLED_BYTE_ARRAY()) \
    .uid("tenant-cost-partial")

late_parts = parts.get_side_output(late_tag) \
    .map(record_to_part, output_type=Types.PICKLED_BYTE_ARRAY()) \
    .uid("tenant-cost-late")                     # CostPart(tenant_id, usd, 1)

tenant_cost = parts.union(late_parts) \
    .key_by(lambda p: p.tenant_id, key_type=Types.STRING()) \
    .window(TumblingEventTimeWindows.of(Time.minutes(1))) \
    .allowed_lateness(120_000)  # milliseconds \
    .side_output_late_data(late_part_tag) \
    .aggregate(SumParts(), window_function=EmitTenantCostMinute(), accumulator_type=ACC,
               output_type=Types.PICKLED_BYTE_ARRAY()) \
    .uid("tenant-cost")

One detail matters for money: where the allowed lateness goes. Module 6 put it on tenant-cost, and module 6 also showed that a re-fire emits the whole updated total, not the difference. If stage A re-fired, stage B would add a partial it had already counted, and a tenant would be double-billed inside the job, which no sink can repair. So stage A has no allowed lateness. Its late RequestRecords go out the side output, are converted to one-request CostParts that keep their original timestamps, and are added once in stage B. Stage B keeps module 6's two minutes of allowed lateness and module 6's re-fire, whose duplicate module 8's end-to-end exactly-once sink or the upsert absorbs.

What this costs you:

• Scope. It works because usd and requests add up. ModelLatency's p99 does not add up across partials. Module 6's sketch merge is already approximate, so a hot model needs a different answer.
• State. Up to eight stage-A panes per tenant per open minute instead of one. At 400 tenants that is 3,200 small accumulators.
• Salting every tenant, or just the hot ones. Salting every tenant costs eight times as many partials (trivial here) and avoids a hot-tenant list you must maintain and redeploy. Choose the salt count from the rate: 400 / 8 = 50 rec/s per bucket, well under 250. Salted keys can still collide on one subtask. With 8 keys over 32 subtasks, expect about 7 distinct subtasks.
• Deployment. This is a graph change, so module 10's runbook applies. tenant-cost-partial is new and "will be initialized without any state". Stage B keeps the uid tenant-cost and the same (usd, requests) accumulator type, which is what gives its panes a chance to restore. Lab 11.2 checks this rather than assuming it. The stop-with-savepoint is aligned ("savepoints will always be aligned"), so it takes as long as the 31 s checkpoints do.

Table/SQL gets this from the optimizer: upstream's tuning page describes Local-Global aggregation as "proposed to solve data skew problem by dividing a group aggregation into two stages" (it depends on mini-batch). The DataStream API has no such switch.

Metrics that matter

Flink registers these metrics without being asked. Each name below is from the metrics reference, as of 2026-09.

Two gaps to know about. The reference has no per-operator "state size" gauge. lastCheckpointFullSize is the proxy, and key/value size tracking (state.size-track.keyed-state-enabled) is off by default and recommended "only ... for debugging purposes". Kafka's own consumer metrics are forwarded under KafkaSourceReader.KafkaConsumer (upstream's example is records-consumed-total). The spine's records-lag-max is a Kafka client name, which Flink's docs do not list. Look it up in Kafka's documentation and treat pendingRecords as the primary lag number.

Where to find them. In the Web UI, on a job's Metrics tab, "Task metrics are listed as <subtask_index>.<metric_name>" and operator metrics as <subtask_index>.<operator_name>.<metric_name>, refreshed every 10 seconds. The Checkpoints tab has the History and Summary views from module 7. Upstream warns that those statistics "don't survive a JobManager loss and are reset if your JobManager fails over".

REST. The Web UI is built on the monitoring API: "a REST-ful API that accepts HTTP requests and responds with JSON data", served by the JobManager on rest.port (default 8081). It is versioned (/v1/...), and an unversioned request goes to "the oldest version supporting the request". The paths this module uses come from upstream's metrics page and playground:

GET /jobs/<jobid>                                                  # vertices: id, name, parallelism
GET /jobs/<jobid>/metrics?get=lastCheckpointDuration,numberOfFailedCheckpoints
GET /jobs/<jobid>/vertices/<vertexid>/subtasks/metrics             # list metric ids
GET /jobs/<jobid>/vertices/<vertexid>/subtasks/metrics?get=<ids>&agg=min,max,avg

The aggregated form answers the skew question in one call: max divided by average of busy time across a vertex's subtasks. Near 1 is healthy. For tenant-cost at 09:05 it is about 1,000 / 42 ≈ 24.

# labs/11-production/pull_metrics.py — skew report for every vertex
import sys, requests
FLINK, JOB = "http://localhost:8081", sys.argv[1]
WANT = ("busyTimeMsPerSecond", "numRecordsInPerSecond")

def get(path, **params):
    return requests.get(FLINK + path, params=params, timeout=5).json()

for v in get(f"/jobs/{JOB}")["vertices"]:
    base = f"/jobs/{JOB}/vertices/{v['id']}/subtasks/metrics"
    ids = [m["id"] for m in get(base) if m["id"].endswith(WANT)]   # exact ids vary by scope
    for m in get(base, get=",".join(ids), agg="min,max,avg") if ids else []:
        skew = m["max"] / m["avg"] if m["avg"] else 0.0
        print(f"{v['name'][:40]:40} {m['id'][-24:]:24} max/avg={skew:5.1f}")

A custom counter. Abandoned requests (module 6's timer) are a gateway health signal that Flink cannot compute for you. PyFlink supports counters, gauges and meters. Upstream's Python tab reads "Still not supported in Python API" for histograms and custom Counter implementations. The accessor is get_metrics_group() on the runtime context, which is the PyFlink 2.3 source's spelling; one upstream Python tab writes get_metric_group(), which does not exist.

class RequestAssembler(KeyedProcessFunction):
    def open(self, runtime_context: RuntimeContext):
        self.open_state = runtime_context.get_state(
            ValueStateDescriptor("open", Types.PICKLED_BYTE_ARRAY()))
        self.abandoned = runtime_context.get_metrics_group() \
            .add_group("gatewaymeter").counter("abandoned_requests")

    def on_timer(self, timestamp, ctx):
        cur = self.open_state.value()
        if cur is not None and timestamp == cur.started_ts + ABANDON_MS:
            self.abandoned.inc()
            yield RequestRecord(...)                   # module 6's abandoned record
            self.open_state.clear()

High availability: the JobManager

Upstream, exact: "By default, there is a single JobManager instance per Flink cluster. This creates a single point of failure (SPOF): if the JobManager crashes, no new programs can be submitted and running programs fail." The TaskManagers hold the state. The JobManager holds the job itself: the JobGraph, the jars, and which checkpoint is the latest completed one.

High availability runs "a single leading JobManager at any time and multiple standby JobManagers". Its services do three things: leader election, service discovery ("Retrieving the address of the current leader"), and state persistence ("JobGraphs, user code jars, completed checkpoints"). Flink ships two implementations: ZooKeeper, which "can be used with every Flink cluster deployment" but needs a quorum, and Kubernetes, which works "only when running on Kubernetes". Both need high-availability.type and high-availability.storageDir. "JobManager metadata is persisted in the file system storageDir and only a pointer to this state is stored in ZooKeeper", or in Kubernetes ConfigMaps. ZooKeeper also needs high-availability.zookeeper.quorum, and upstream recommends setting high-availability.cluster-id per cluster. Kubernetes needs kubernetes.cluster-id.

The compose cluster is a standalone cluster, and "to enable HA for a standalone cluster, you have to use the ZooKeeper HA services" and start more than one JobManager. The current Lab 11 does not add that control-plane exercise; a future advanced HA lab would add a ZooKeeper container and a standby JobManager, with storageDir on shared durable storage.

high-availability.type: zookeeper
high-availability.zookeeper.quorum: zookeeper:2181
high-availability.cluster-id: /gatewaymeter
high-availability.storageDir: file:///checkpoints/ha

Three operational notes from the docs. After recovery, jobs "are matched by name". Upstream recommends a unique name via execute(jobName), so call env.execute("gatewaymeter"). TaskManagers in HA setups do not use jobmanager.rpc.address: they "discover the JobManager via the High-Availability-Service". By default a suspended ZooKeeper connection makes Flink "invalidate all leaderships of its components and thereby triggering a failover", so a flaky network to ZooKeeper shows up as JobManager failovers (high-availability.zookeeper.client.tolerate-suspended-connections relaxes this). The docs do not describe what running tasks do during a leader change. They promise that jobs "resume execution (from the latest checkpoint)", so treat a JobManager failover as a restart from checkpoint k.

Disaggregated state and ForSt

Module 7 listed ForSt; here is what it is for. Upstream names three limits of local state: size bounded by the TaskManager's memory or disk, CPU and I/O spikes during checkpoints and compaction, and "Heavy Recovery", where recovery time grows with state size. Flink 2.0 introduced disaggregated state: the forst backend keeps its files on S3 or HDFS and uses local disk as a cache, State V2 provides asynchronous state access, and SQL operators are rewritten to use both. As of 2026-09: "The disaggregated state management is still in experimental state ... The APIs and configurations may change in future release", and the backend "is not fully available for production".

Three facts keep GatewayMeter off it. First, "By default, ForSt will ONLY disaggregate state when asynchronous APIs (State V2) are used", and the State V2 page documents only the Java enableAsyncState(). PyFlink state is synchronous, so ForSt would act as a local store. Second, its limits include no canonical savepoint, which module 10's runbook depends on for backend and schema changes. Third, upstream itself says: "when the state size is small, the local state management with synchronous state access is a better choice". GatewayMeter's state is a few thousand accumulators plus the open requests. It stays on RocksDB. The same page records a Rust rewrite of ForSt, projected "for later 2026 (August, optimistically)", after which the current version "will be phased out". Check the upstream page before choosing ForSt for anything.

The checklist, in our words

From upstream's production readiness checklist, mapped to GatewayMeter:

1. Explicit max parallelism. There is "no way to change" it later without discarding state. Set to 128 in module 10.
2. A uid on every operator. Auto-generated ids are "very fragile". Module 10, plus the two new ones above.
3. The right state backend. RocksDB with incremental checkpoints (module 7); ForSt when state outgrows disks and the job can use async state.
4. The checkpoint interval as an SLA. 10 s means at most about 10 s of input is reprocessed. And because "Exactly once sinks ... only make results visible on checkpoint completion", it is also the minimum delay before tenant-cost is visible to the billing consumer.
5. JobManager HA. "highly recommended for production setups".
6. Secure access. "Flink is intentionally designed to support remote code execution." Keep port 8081 off the internet. Note that web.submit.enable: false only hides the UI upload: "session clusters still accept jobs through REST requests".

If a TaskManager dies right now

Survives: checkpoint k: stage-A partial panes, stage-B panes, open requests, timers, and offsets S_k. With the hot tenant still pinned, the 10 s interval has become about 33 s (a 31 s checkpoint plus a 2 s minimum pause), so k is older than you expect. Recomputed: everything after S_k, into restored panes, once per key. Restore time shows in initializationTime, where "most of the initialization time is usually spent in restoring from the checkpoint". Emitted twice: nothing on tenant-cost under EXACTLY_ONCE; alerts may repeat.

If the JobManager dies instead. With HA, a standby takes leadership, reads the JobGraph and the latest completed checkpoint from storageDir, and the job resumes from k: the same answer as above. The Web UI's checkpoint counts start from zero. Without HA, running programs fail and a restarted JobManager has no record of GatewayMeter. The retained chk-<k>/ files are still on the volume, and a person resubmits with flink run -s (module 7, exercise 7.4). Until then every minute is unbilled. If the outage outlasts the sink's transaction.timeout.ms, you are in module 8's table 3.

Try it

Lab 11 is in labs/11-production/:

• 11.1 Make acme 40% of traffic and capture records, busy time, backpressure and input watermark by subtask to locate the hot key.
• 11.2 Explain why adding parallelism alone cannot split that single tenant key across subtasks; the lab asks for the partitioning argument before the salting test.
• 11.3 Run the lab's replacement-merge test, including a late re-fire of one salt. Compare final totals with the unsalted oracle rather than adding both versions of a partial result.
• 11.4 Build an operator report from the REST API with per-vertex/subtask records, busy/backpressured/idle time, watermark, checkpoint data, restarts and Kafka lag where available. Save the response with your notes.

JobManager HA failover is discussed above but is not an exercise in the current lab.

Lab page: lab-11-production.html — the exercises, every source file, and the run-it-locally steps.