Capstone: GatewayMeter Under Fire

The exam

This page is mostly original: derive the job from its requirements, run a reproducible incident, and defend it. The Temporal capstone's rule applies: write the failure contract before you draw a box.

GatewayMeter must:

• meter 5,000,000 requests a day for 400 tenants across 30 models;
• bill each tenant per minute, right to the cent: never doubled, never missing, corrections within 2 minutes;
• detect abandoned requests;
• tolerate events 20 s out of order and an occasional burst 30 s late;
• survive one tenant sending 40% of traffic, a billing service that takes up to 2 s, TaskManager deaths, and two deploys a day;
• be replayable for an audit.

The runnable capstone proves the invoice path: assembly, pricing, tenant-cost windows, recovery, transactional output, salting, and replacement merge. Per-model p99, error-rate alerts, broadcast pricing, and HA remain production-design extensions discussed in the reading; the current lab does not pretend to execute them.

The failure contract

One row per operator, one line per cell; mp is max parallelism (128 for every keyed operator). Rows for model-latency, alerts, and a broadcast price table describe the target production review, not operators present in the current stage-12 lab graph.

sink-model-latency matches sink-alerts. New since the modules: the two-stage cost operator and a source uid.

Deriving the job

Keys

Each key follows from its state: request_id a lifecycle, tenant_id a bill, model a dashboard and an alert. The 40% tenant breaks tenant_id: a key cannot be split, so acme pins one subtask whose trailing watermark holds every tenant (module 11). Salt the key, aggregate partials, merge.

A key "must be derived from deterministic computations", and Python's string hash() is seeded per worker process; zlib.crc32(request_id.encode()) % 8 is the same in every worker and in an audit replay. Salt every tenant, not a hot list: 3,200 keys cost nothing.

The trap is stage 2. Stage 1 is module 6's window with allowed_lateness(2 min), so a late firing emits a corrected partial; a stage-2 window summing partials would add it on top of the original, a double count inside Flink. Stage 2 replaces a partial, never adds one:

LATENESS_MS = 2 * 60 * 1000

class MergePartials(KeyedProcessFunction):            # keyed by tenant_id; uid "tenant-cost"
    def open(self, ctx):
        self.parts = ctx.get_map_state(MapStateDescriptor(
            "partials", Types.PICKLED_BYTE_ARRAY(), Types.PICKLED_BYTE_ARRAY()))

    def process_element(self, p, ctx):                # (tenant_id, salt, window_end, micro_usd, requests)
        self.parts.put((p.window_end, p.salt), (p.micro_usd, p.requests))   # replace, never add
        ts = ctx.timer_service()
        if ts.current_watermark() >= p.window_end - 1:
            yield self.total(p.tenant_id, p.window_end)   # main firing already sent: a correction
        else:
            ts.register_event_time_timer(p.window_end - 1)
        ts.register_event_time_timer(p.window_end + LATENESS_MS)

    def on_timer(self, t, ctx):
        if t % 60_000 == 59_999:                      # window_end - 1: every main partial is in
            yield self.total(ctx.get_current_key(), t + 1)
        else:                                         # window_end + lateness: stage 1 is done
            for k in [k for k in self.parts.keys() if k[0] == t - LATENESS_MS]:
                self.parts.remove(k)

    def total(self, tenant_id, window_end):
        v = [x for k, x in self.parts.items() if k[0] == window_end]
        return TenantCostMinute(tenant_id, window_end, sum(x[0] for x in v), sum(x[1] for x in v))

The window_end − 1 timer fires only once every stage-1 subtask has emitted its partials and then forwarded its watermark.

Time

Event time is completed_ts (module 4); bound 20 s, idleness 1 min (module 5); allowed lateness 2 min, which is the corrections requirement, with later records sent to the side output as adjustments. The source runs at parallelism 8, because "the Kafka Source does not go automatically in an idle state if the parallelism is higher than the number of partitions". Module 5's watermark alignment is off: against a dead partition, this suite's failure, it pauses every healthy one. Write the alert cost into the contract: alerts fire about 20 s after their minute, but the first minute of a stall misses 60 s.

State

• open is bounded by its timer. Module 3's 10-minute TTL counts processing time, so demote it to a backstop of hours: a TTL shorter than an event-time stall expires the entry before its timer fires, and the completion is dropped as "completion without a start", never billed. Likewise, hold an early completion under the timer instead of dropping it as module 2's code does.
• Record money as integer micro-dollars (Types.LONG()). Stage 2 adds partials in arrival order, and floating-point addition is not associative. Round once, on the invoice.
• ModelLatency samples with random: approximate, not reproducible on replay. Fine for a dashboard, never for an invoice.

Checkpoints

Keep module 7's build_env() as it is: 10 s interval, EXACTLY_ONCE mode, RocksDB with incremental checkpoints, filesystem storage (s3:// in production), RETAIN_ON_CANCELLATION, min pause 2 s, timeout 60 s, one concurrent checkpoint, zero tolerable failures, and fixed-delay restarts (3 × 10 s). Add one setting for backpressure:

config.set_string('execution.checkpointing.unaligned', 'true')
config.set_string('execution.checkpointing.aligned-checkpoint-timeout', '30 s')   # upstream's example

Checkpoints start aligned and go unaligned after 30 s; savepoints are always aligned. The checklist sets the interval: "how much data can you tolerate reprocessing?" (10 s), and "Exactly once sinks, such as Kafka or the FileSink, only make results visible on checkpoint completion". ForSt stays out: experimental as of 2026-09 (module 7).

Sinks and enrichment

Upstream: "To guarantee end-to-end exactly-once record delivery (in addition to exactly-once state semantics), the data sink needs to take part in the checkpointing mechanism." tenant-cost gets both: DeliveryGuarantee.EXACTLY_ONCE (prefix gatewaymeter-tenant-cost, transaction.timeout.ms 15 min) with read_committed, so the topic is an audit log with one record per firing and none per recovery; and an upsert by (tenant_id, window_end) in the billing store, because late firings are real corrections. model-latency and alerts stay AT_LEAST_ONCE: duplicates are cheap there, a 10 s delay is not (module 8).

enrich_price answers from a broadcast price table and calls HTTP only on a miss, via module 9's unordered_wait_with_retry (5 s timeout, capacity 100, 3 retries). A timeout bills list price, flagged price_source="list" for re-rating; a replay may price it differently, so the billing store keeps the price it used.

Scale and operations

Parallelism 32, max parallelism 128 set explicitly (4 key groups per subtask), and a uid on every stateful operator, including the source. The cluster is 5 TaskManagers × 8 slots: a standalone ResourceManager cannot start a TaskManager (module 1), so losing one must still leave 32 slots. Deploys follow module 10's runbook. The JobManager is "a single point of failure within the cluster, and if it crashes, no new jobs can be submitted, and running applications will fail"; HA through ZooKeeper or Kubernetes persists "JobGraphs, user code jars, completed checkpoints".

The job

Every decision above, in one graph. Read the shuffles first — each one is a key you chose and a reason the next operator needs every record for that key in one place — then the three sinks, whose guarantees differ because what reads them differs.

The injection suite

The lab stages the incident from one seeded trace. Every input failure is written into the trace before the job starts, and you add the two TaskManager deaths by hand. "Everything at once" is neither reproducible nor diagnosable: each transition gets a time and an evidence artifact.

In the lab

The trace is 20 requests/s for four minutes of event time from 00:00:00 on 2026-01-01, with acme at 40% and 5% of requests 60 s late; every price lookup takes 2 s for the whole run. It is replayed at its own pace, so T is both seconds into the replay and event time since 00:00:00. The lab job reads each partition with its own source subtask and runs the rest at parallelism 4, with a 15 s idleness timeout as in lab 5.3. Its watermark trails the replay by the 20 s bound plus 5–15 s of pipeline delay (two-second lookups, Python bundles), so a minute fires 25–35 s after it ends. The source fills all eight slots, so the job restarts only once a killed TaskManager is back, and you bring each one back as the platform would. A restart can also land before the JobManager notices the dead TaskManager, and that failed deployment counts as a second failure, so the lab raises module 7's three restart attempts to ten.

The invoice under review is the four minutes ending by 00:03:00 (the first holds late requests stamped before 00:00). invoice_oracle.py computes it from the same trace with the job's own semantics, including the price book and the abandonment timer; invoice_diff.py materializes tenant-cost the way the billing store upserts it and compares the two to the micro-dollar. Expect corrected_minutes above zero, because late data re-fires windows, and repeated_values at zero, because read_committed never shows an aborted transaction. On the course cluster (2026-09-19) the two kills cost four restarts, and the diff matched all 24 minutes to the micro-dollar.

At production scale

The same failures at production scale are a thought experiment with assumed numbers. They are staged differently: here the TaskManager dies during the silence, which exposes idleness as runtime state that a restart forgets. The suite replays 1,200 requests/s (about 6,000 events/s, roughly 20× the daily average) so four failures fit in five minutes; acme sends 480/s. Assume, until your lab measures it, that one PyFlink tenant-cost-partial subtask saturates at about 300 records/s. t = 0 is 10:00:00; event time tracks the wall clock; checkpoints 101 and 102 complete at t = 1 s and 11 s.

The invoice at t = 300 s. Minutes up to 09:59:00 were committed before the suite. Those ending 10:00:00–10:02:00 fired without p5's eighth: acme's 10:01:00 minute read $42.00, then $48.00 at about t = 155 s. The upsert holds $48.00, a single-threaded replay's figure to the cent; an append-and-sum ledger would bill every correction on top. 10:03:00 and 10:04:00 were exact at first firing. Replay produced no duplicate only because frozen event time fired nothing between checkpoint 102 and the kill, so 12.3 kills a second TaskManager just after a firing. And had p5 slept about 50 s longer, its 10:00 events would have gone to the side output.

The module-10 job fares worse. Module 5's watermark alignment pauses the healthy partitions behind the dead one. acme's single subtask sits at 1,000 ms/s busy until its backpressure expires checkpoints (module 9's failover loop). The restart cannot place 32 subtasks on 24 slots. And module 8's append-only consumer bills each correction on top of the minute it corrects.

Where Flink ends

Ten defense questions

1. What survives a TaskManager death? Checkpoint k: all operator state and every offset S_k. Nothing that lived only in a slot.
2. How does recovery rebuild the job? It redeploys, restores k, rewinds to S_k and replays: "Any records that are processed as part of the restarted parallel dataflow are guaranteed to not have affected the previously checkpointed state."
3. Why must the salt be deterministic? Upstream requires deterministic keys, and an audit replay must rebuild the same partials. Python's string hash() differs per worker; crc32 does not.
4. Why can a TenantCostMinute appear twice? A late firing is a correction with a new value; a replay repeats the same value. Transactions hide the replay; only the upsert handles the correction.
5. Replay or retry? Replay re-reads the source against restored state; retry re-issues a timed-out lookup, and recovery every in-flight one. Both are safe only because the lookup is a read.
6. Who decides that a minute is over? The watermark generator you configured, from the data it has seen. Operators take the minimum of their inputs, so a silent partition holds everyone back until idleness releases it. After that, its events arrive late.
7. Transactional sink or upsert? The invoice needs the upsert, because corrections exist. The transaction keeps the topic auditable at a cost of up to 10 s. "Flink is exactly-once" answers neither question.
8. What must be longer than what? transaction.timeout.ms ≫ worst checkpoint + worst restart; open TTL > longest event-time stall; abandonment threshold > longest legitimate request; Kafka retention > worst outage and the audit window.
9. How do you deploy twice a day? Stop-with-savepoint (no --drain), then run -s; never -n unless you can name the state it discards; uids and max parallelism set before the first start; never upgrade from an unaligned checkpoint.
10. Where does Flink end? See the table above.

If a TaskManager dies right now

This is the course's full answer. Survives: checkpoint k, in storage no TaskManager owns: offsets, open requests, timers, partials, reservoirs, the price table, in-flight lookups; and every committed transaction, so everything a read_committed consumer has read. Recomputed: every record after S_k, against state that never saw it, and the in-flight lookups. From outside: a pause (detection + 10 s + catch-up) and a watermark that briefly moves backwards. Emitted twice: nothing on tenant-cost at read_committed, because the open transaction is never committed. Alerts and latency windows since k are published again, and their consumers dedupe them. Only the random p99 sample and list-priced records can differ after recovery. Money is never doubled and never missing. A late minute is corrected within 2 minutes or adjusted on the side.

Try it

Lab 12 is in labs/12-capstone-gatewaymeter/:

• 12.1: Fill in the failure contract from memory, then check it against this page.
• 12.2: Empty both topics, generate the seeded trace and match its published SHA-256, compute the oracle for the four minutes ending by 00:03:00, and submit stage 12 with a 2 s price lookup.
• 12.3: Replay the trace and follow the timeline above: kill an occupied TaskManager at T+125 and a second one just after the minute ending 00:02 fires, bringing each back. Capture job status, numRestarts, watermarks, the last minute visible at read_committed, lag and checkpoints at each row.
• 12.4: Read tenant-cost at read_committed, run invoice_diff.py against the oracle, and write the defense: the contract, the dataflow, the ten questions, and the evidence (checkpoint history, numRestarts, the invoice diff).

Optional module 13, Which Runtime Does an Agent Belong In?, puts GatewayMeter beside the Temporal course's AgentRun and asks which runtime each part of an agent system belongs in.

Lab page: lab-12-capstone-gatewaymeter.html — the exercises, every source file, and the run-it-locally steps.