Scale and Upgrade

A savepoint is a checkpoint you own. Module 7 gave GatewayMeter checkpoints so a dead TaskManager could not lose a tenant's running total; this module uses the same snapshot mechanism for the changes you schedule: more slots, a new operator, a new field on RequestRecord, a new Flink. Every one of those goes through a savepoint, and there are exactly two ways to lose state on the way: a max parallelism you cannot raise, and an operator id you never set.

Checkpoint or savepoint

Upstream's distinction is about ownership, and its words are kept exact. A checkpoint "is created, owned, and released by Flink - without user interaction"; its design goals are "being as lightweight to create and ii) being as fast to restore from as possible", and those optimizations may assume "that the job code doesn't change between the execution attempts." Savepoints, by contrast, "are created, owned and deleted solely by the user. That means, Flink does not delete savepoints neither after job termination nor after restore." Their design "focuses more on portability and operational flexibility, especially with respect to changes to the job", and "the use case for savepoints is for planned, manual operations."

The capabilities table in checkpoints_vs_savepoints.md is the operational summary (✓ supported, ! works without an official guarantee, ✗ unsupported), as of 2026-09 (selected rows):

Read the last column as a warning: module 7's unaligned checkpoints carry in-flight records, so they can rescale (and restore a non-arbitrary job upgrade or a patch-version upgrade) but cannot be the basis of an arbitrary job upgrade or a Flink minor-version upgrade. Everything in this module starts from a savepoint.

Taking one

The savepoint is a directory of state files plus a _metadata file, written to a location "accessible by both the JobManager(s) and TaskManager(s)"; the default comes from execution.checkpointing.savepoint-dir, and "if you neither configure a default nor specify a custom target directory, triggering the savepoint will fail."

bin/flink savepoint :jobId [:targetDirectory]              # take one, keep running
bin/flink savepoint --type [native/canonical] :jobId       # default: canonical
bin/flink savepoint :jobId [:targetDirectory] -detached    # return the trigger id at once
bin/flink savepoint -d :savepointPath                      # dispose

Two formats. The canonical format is "a format that has been unified across all state backends, which lets you take a savepoint with one state backend and then restore it using another. This is the most stable format, that is targeted at maintaining the most compatibility with previous versions, schemas, modifications etc." The native format "creates a snapshot in the format specific for the used state backend (e.g. SST files for RocksDB)" and exists because canonical "often it is slow to take and restore from." Rule: native for a routine rescale of a large RocksDB state; canonical for anything that changes the backend, the schema or the Flink version.

Before an upgrade you want the job stopped by the savepoint, not after it:

bin/flink stop --type [native/canonical] --savepointPath [:targetDirectory] :jobId

The CLI page describes what stop does: "all sources will be requested to send the last checkpoint barrier that will trigger a savepoint, and after the successful completion of that savepoint, they will finish by calling their cancel() method." That final barrier is the module 8 commit point: the EXACTLY_ONCE KafkaSink on tenant-cost commits its open transaction on that snapshot, so the last TenantCostMinute written before the stop is visible to the billing consumer and nothing is left in an aborted transaction. A savepoint taken while running does not do this. Upstream's warning, kept exact: "Starting from Flink 1.15 intermediate savepoints (savepoints other than created with stop-with-savepoint) are not used for recovery and do not commit any side effects." Resume from an intermediate savepoint and "we might commit transactions that might've never happened."

Do not add --drain when you intend to resume. It emits a MAX_WATERMARK so every event-time timer fires and every open window closes; "if you want to resume the job at a later point in time, then do not drain the pipeline because it could lead to incorrect results when the job is resumed." For GatewayMeter that would close every tenant's current minute early and bill it short. The older flink cancel --withSavepoint is deprecated in favour of stop.

Resuming

bin/flink run -s :savepointPath -p 32 [:runArgs]        # path to the directory or its _metadata
bin/flink run -s :savepointPath -n [:runArgs]           # -n = --allowNonRestoredState
bin/flink run -s :savepointPath -claimMode :mode [:runArgs]

The job gets a new JobID. Resume, per the upgrading page, means "the state of its operators is initialized with the operator state persisted in the savepoint", and "the started application continues processing from exactly this point on" — the Kafka source's offsets included, so gateway-events is re-read from the savepoint's position, not from the current head.

Who owns the files after the restore is the claim mode. In the default NO_CLAIM, Flink "will leave the files in user's control and never delete any of the files", and to stop depending on them "it will force the first (successful) checkpoint to be a full checkpoint as opposed to an incremental one." The consequence is the last line of this module's runbook: "once a checkpoint succeeds you can manually delete the original snapshot. You can not do this earlier, because without any completed checkpoints Flink will - upon failure - try to recover from the initial snapshot." In CLAIM mode Flink "treats it like a checkpoint" and may delete it; do not start two jobs from a claimed savepoint.

State is matched by uid

The savepoint is "a map of Operator ID -> State for each stateful operator". Upstream: "It is highly recommended that you specify operator IDs via the uid(String) method. These IDs are used to scope the state of each operator." And the sentence that decides failure 2 below: "If you do not specify the IDs manually they will be generated automatically. You can automatically restore from the savepoint as long as these IDs do not change. The generated IDs depend on the structure of your program and are sensitive to program changes." The production checklist says how: "by traversing the JobGraph and hashing specific operator properties", which "is also very fragile, as changes to the JobGraph (e.g., exchanging an operator) results in new UUIDs."

The upstream snapshot documents the call in Java only:

// Java, from ops/state/savepoints.md
DataStream<String> stream = env
  .addSource(new StatefulSource()).uid("source-id")
  .shuffle()
  .map(new StatefulMapper()).uid("mapper-id")
  .print();                                     // stateless: auto-generated id, not in the savepoint

The local docs have no Python example of uid; the code below assumes PyFlink's DataStream method of the same name, which lab 10.1 verifies on the compose cluster. set_max_parallelism does appear in the Python docs (env.set_max_parallelism(256), dev/table/data_stream_api.md). Stage 10 sets every frozen uid from the spine:

# gatewaymeter/job.py, stage 10 — abridged: enrich_price (module 9), allowed lateness,
# the late side output and the window function (module 6) are unchanged and omitted here
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(int(os.environ.get("GATEWAYMETER_PARALLELISM", "4")))
env.set_max_parallelism(128)        # decided once, at the first start; see "Rescaling"

events = env.from_source(source, watermark_strategy, "gateway-events") \
    .uid("gateway-events")         # the Kafka offsets are this operator's state

records = (events
    .map(parse_event, output_type=EVENT_TYPE).uid("parse")
    .key_by(lambda e: e.request_id)
    .process(RequestAssembler(), output_type=REQUEST_RECORD_TYPE).uid("assemble"))

(records
    .key_by(lambda r: r.tenant_id)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(TenantCost(), output_type=TENANT_COST_MINUTE_TYPE).uid("tenant-cost")
    .sink_to(tenant_cost_sink).uid("sink-tenant-cost"))

Upstream's FAQ answers the rest. Should every operator get one? "As a rule of thumb, yes", because "some of Flink's built-in operators like the Window operator are also stateful and it is not obvious which built-in operators are actually stateful" — tenant-cost is a window operator. Add a stateful operator: "it will be initialized without any state." Delete one: "If you restore from a savepoint that contains state for an operator that has been deleted, this will therefore fail", unless you pass --allowNonRestoredState. Reorder without ids: "the auto generated IDs of the stateful operators will most likely change after the reordering." (Java only: ExecutionConfig#disableAutoGeneratedUIDs fails submission when any operator lacks one.)

Rescaling: key groups and the ceiling

Keyed state moves in units, not keys. Upstream, exact: "Key Groups are the atomic unit by which Flink can redistribute Keyed State; there are exactly as many Key Groups as the defined maximum parallelism. During execution each parallel instance of a keyed operator works with the keys for one or more Key Groups." A rescale hands whole key groups to new subtasks; a tenant's total moves with its group.

Max parallelism is therefore the number of key groups, and it is fixed at first start. The checklist: "There is currently no way to change the maximum parallelism of an operator after a job has started without discarding that operators state." The constraint is 0 < parallelism <= max parallelism <= 2^15. The default, when you set nothing, is decided "when the job is first started":

• "128 : for all parallelism <= 128."
• "MIN(nextPowerOfTwo(parallelism + (parallelism / 2)), 2^15) : for all parallelism > 128."

(The parallelism page phrases the same rule as "roughly operatorParallelism + (operatorParallelism / 2) with a lower bound of 128 and an upper bound of 32768", and adds: "Changing the maximum parallelism explicitly when recovery from original job will lead to state incompatibility.") The recommendation is a judgment, not a number: "choose max parallelism that is high enough to fit your future needs in scalability, while keeping it low enough to maintain reasonable performance", since backends keep metadata that "grows linearly with max parallelism". GatewayMeter sets 128 explicitly: the same value the default would compute, but written down, so nobody later believes it was 4.

Operator (non-keyed) state redistributes differently: with even-split, "the list is evenly divided into as many sublists as there are parallel operators"; with union, "each operator gets the complete list of state elements." The Kafka source's partition offsets are the operator state GatewayMeter has; state.md notes "Operator state is still not supported in Python DataStream API", so you rescale it but never write it in this course.

GatewayMeter stage 10: 4 to 32

The cluster grows from four TaskManagers with two slots to 32 slots, a production cluster; the lab's eight slots take the same steps from 4 to 8 (lab 10.2). The procedure: stop-with-savepoint (canonical, since the next step may also carry a code change), redeploy the job with GATEWAYMETER_PARALLELISM=32, resume with -s. Three things go wrong on the way.

Failure 1: max parallelism left to the default

With parallelism 4 and nothing set, the formula gives 128 key groups. That is enough for 32, and not enough for the year after:

Raising the ceiling means "discarding that operators state": every TenantCost window in flight and every open request restarts empty, and the minutes in flight are billed short. The docs quote no exception text for the refused case; the lab reads it off the JobManager log.

Failure 2: tenant-cost without a uid

Module 9 inserted enrich_price between assemble and the cost window. If tenant-cost had no uid, its auto-generated id — a hash of "specific operator properties" traversed from the graph — changes, and the savepoint's entry no longer matches any operator.

The -n row is the dollar sign. Upstream: "Improper usage of this feature could result in significant issues with the correctness of the application", and because "operator UIDs are reassigned based on topological order by default", state may also be handed to the wrong operator rather than dropped. Nothing in the Web UI distinguishes the -n job from a healthy one; the missing money shows up on the invoice.

Failure 3: a new field on RequestRecord

RequestRecord gains tool_calls. What restores depends on where the type lives:

Upstream's evolution support is narrow: "Currently, schema evolution is supported only for POJO and Avro types." POJO rules: fields can be removed; "New fields can be added. The new field will be initialized to the default value for its type"; "Declared fields types cannot change"; the class name and namespace cannot change. Keys never evolve: "Schema evolution of keys is not supported." And for anything Flink cannot inspect: "Kryo cannot be used for schema evolution." State TTL, separately, can be switched on or off across a restore since 2.2.0 without a StateMigrationException (state_migration.md).

PyFlink is where the page goes quiet. schema_evolution.md is scoped to "state serializers that are generated by Flink's own type serialization framework", names POJO and Avro, and says nothing about Python. The PyFlink examples in the docs declare state as Types.PICKLED_BYTE_ARRAY(), which is opaque bytes to Flink: no documented migration applies, so a pickled OpenRequest that changes shape is on you. The course's rule: declare state with explicit Types (the Types.ROW(...) shape the docs show) so the type is at least visible, keep the key type frozen, and carry a version field in any pickled payload so your process_element can migrate an old value when it first reads it. Migrating state outside the job — the State Processor API — is Java-only and writes only canonical savepoints.

Upgrading Flink itself

The documented procedure has two steps: "Take a savepoint in the previous, old Flink version for the jobs you want to migrate" and "Resume your jobs under the new Flink version from the previously taken savepoints", either in-place (stop jobs, shut the cluster down, upgrade, restart) or as a shadow copy (a new installation beside the old one, resume there, retire the old cluster once it runs). Explicit uids are a soft precondition; a hard one is that "all the savepoint data must be accessible from the new installation under the same (absolute) path." The capabilities table permits a minor-version upgrade from a canonical or native savepoint or an aligned checkpoint, never an unaligned one. The savepoint compatibility table in this docs snapshot lists sources 1.8.x–1.20.x resumed with 1.17.x–1.20.x and has no 2.x row (as of 2026-09; see the upstream page); treat a 2.x-to-2.y move as "savepoint, then verify", not as a table lookup.

The upgrade runbook

1. Confirm every stateful operator has a uid and set_max_parallelism is explicit; if either is missing, this is not an upgrade but a rebuild, and the minutes in flight will be short.
2. bin/flink stop --savepointPath <dir> :jobId — canonical unless nothing but parallelism changes; no --drain. Record the printed path.
3. Deploy the new job code (and, for a Flink upgrade, the new cluster with the savepoint reachable at the same absolute path).
4. bin/flink run -s <path> -p 32 ... — never -n unless you can name the operator whose state you are discarding and why it owes nobody money.
5. Verify: currentInputWatermark on tenant-cost advances past the stop time; numRecordsInPerSecond on parse recovers; the first checkpoint completes (it is a full one).
6. Keep the savepoint until that first checkpoint completes; then dispose it, and only then.

Temporal, in one paragraph

Temporal's module 9 kept an old history and new code in agreement: workflow.patched("insert-verify-step") branches inside a replay, and a Worker Deployment Version pins an execution to the build it started on. Flink keeps old state and a new graph in agreement: the savepoint is a map from operator id to state, restored into whatever graph you submit. Both fail on the same omission, a unit of code or state with no name: a patch id that changes fails the Workflow Task, and an operator without a uid fails the restore. The difference is which one is silent. A Temporal non-determinism error is loud and blocks progress; a Flink restore with -n runs, at full throughput, on empty totals.

If a TaskManager dies right now

The savepoint is immutable and user-owned: Flink "will not delete its files" and the restore never mutates them. If the death happens during the stop, the savepoint is not complete, so it is not a savepoint; the old job fails over from its last checkpoint under its restart strategy exactly as in module 7, and you stop again. If it happens during the restore, before the first checkpoint, the new job recovers from the savepoint itself, which is why you keep it. A restore that fails (failure 1, failure 2 without -n) leaves the stopped job stopped and the savepoint intact, so the numbers that survive are all of them, up to the stop; nothing is recomputed, and nothing can be emitted twice, because stop-with-savepoint committed the sink's last transaction and the resume re-reads gateway-events from the savepoint's offsets under the same exactly-once end-to-end path module 8 built. The one outcome with a dollar sign is not a death at all: a restore with -n that succeeds.

Try it

Lab 10 lives in labs/10-scale-and-upgrade/ and has four exercises:

• 10.1 — run stage 10 until windows and checkpoints complete, stop it with a savepoint (flink stop --savepointPath), and record the path; stage 10 fixes max parallelism at 128 and gives every stateful operator a uid.
• 10.2 — the compose cluster has eight slots, so restore from the savepoint at parallelism 8 and compare keyed totals before and after. The 4 → 32 rescale above needs a cluster with 32 free slots; on the lab cluster it stays a thought experiment.
• 10.3 — in two throwaway copies, change .uid("tenant-cost") and try the restore; then run with GM_MAX_PARALLELISM=4, savepoint, and request parallelism 8. Capture both refusals, and do not reach for --allowNonRestoredState until you can name the state it would discard.
• 10.4 — run the lab's tests, add a versioned field to a pickled state value with an explicit read-time migration, and contrast it with an incompatible key-type change.

Lab page: lab-10-scale-and-upgrade.html — the exercises, every source file, and the run-it-locally steps.