Upgrading Apache Flink Jobs Without Losing Streaming State

Upgrading a streaming application differs from deploying a typical web application. A web service can often restart and load the latest data from a database. A stateful streaming job must also preserve the progress and intermediate state it has built while processing events.

Consider an Apache Flink job that reads payment events, maintains running totals, and detects duplicate transactions. After running for several weeks, the job may contain millions of keyed state entries, active event-time timers, and information needed to continue processing correctly.

If you upgrade the job without a suitable recovery plan, it may fail to restore its state, lose processing progress, or produce incorrect results.

Apache Flink provides checkpoints and savepoints to help manage stateful upgrades. However, a successful upgrade also depends on state compatibility, operator identifiers, application changes, and the deployment process.

This article explains how to prepare a Flink job for an upgrade, restore its state safely, and troubleshoot common migration problems.

Understanding Flink State

Apache Flink maintains state when a streaming application needs information from previously processed events.

For example, a payment-processing job might calculate the total amount spent by each customer:

DataStream<Payment> payments = ...;

payments
    .keyBy(Payment::getCustomerId)
    .process(new CustomerSpendingFunction());

The processing function may store a running total for each customer. Flink keeps this information as keyed state, allowing the job to continue its calculations as new payments arrive.

Flink applications can use several kinds of state:

Keyed state: State associated with a key, such as a customer ID or account number.
Operator state: State associated with a parallel operator instance.
Timers: Scheduled callbacks used for event-time or processing-time operations.
Source state: Information needed by a source connector to track its progress, depending on the connector and its implementation.

When an application restarts, Flink needs compatible state and source progress information to resume processing correctly.

Simply deploying a new JAR file does not guarantee that the existing state can be restored.

Checkpoints vs. Savepoints

Checkpoints and savepoints both preserve Flink application state, but they serve different operational purposes.

Feature	Checkpoint	Savepoint
Primary purpose	Fault recovery during normal execution	Controlled recovery, upgrades, and migrations
Typical creation	Managed automatically	Usually triggered deliberately
Lifecycle	Managed according to checkpoint configuration	Managed explicitly by the operator
Upgrade use	Can be useful when the deployment workflow supports checkpoint-based restoration	Common choice for planned application upgrades
Storage	Configured checkpoint storage	Configured savepoint storage

Checkpoints are normally part of the application's regular fault-tolerance mechanism. Flink periodically captures a consistent snapshot that can be used to recover after a failure.

Savepoints are useful when an operator wants to stop or migrate a job in a controlled way.

For a planned upgrade, a savepoint provides a clear recovery artifact that can be referenced when starting the new deployment.

The exact stop and restore commands depend on the Flink version and deployment environment, such as standalone, Kubernetes, or a managed Flink service.

Preparing the Job for an Upgrade

Before stopping a production job, verify that the new application version can restore the state created by the existing version.

Step 1: Identify the Current State

Review the state used by the application, including:

State descriptors and their names.
State types and serializers.
Operator IDs and UIDs.
Key selectors and key types.
Event-time timers and window configuration.
Source connector state and sink behavior.
External dependencies that may affect recovery.

This inventory is especially important when the upgrade changes the application's data model or processing topology.

For example, changing a customer key from an integer to a string may affect the way keyed state is partitioned and restored. Changing a state descriptor name can also prevent Flink from matching existing state to the corresponding operator.

Step 2: Enable and Verify Checkpointing

Checkpointing should be configured and tested before an upgrade is attempted.

A simplified Java configuration might look like this:

StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();

env.enableCheckpointing(60_000);

env.getCheckpointConfig()
    .setCheckpointStorage(
        "s3://my-flink-bucket/checkpoints/"
    );

This example enables checkpoints every 60 seconds and configures checkpoint storage using an S3 path.

The actual storage configuration depends on the Flink version, installed filesystem plugins, and deployment environment. Credentials and access permissions must also be configured correctly.

In production, verify that checkpoints complete successfully and that the stored snapshots remain accessible after a restart.

A checkpoint interval alone does not guarantee recoverability. The application also needs suitable state backends, storage permissions, and compatible source and sink connectors.

Step 3: Create a Savepoint

For a planned upgrade, create a savepoint using the appropriate mechanism for the deployment.

For example, the Flink CLI supports a savepoint operation in compatible versions:

flink savepoint <job-id> \
  s3://my-flink-bucket/savepoints/

Replace <job-id> with the actual running job ID and use a savepoint directory supported by the deployment.

Wait for the operation to complete successfully and record the resulting savepoint location. Do not assume that a command was successful merely because it was submitted.

The savepoint should be retained until the upgraded job has been verified and the rollback window has closed.

Maintaining State Compatibility

State compatibility is one of the most important parts of a Flink upgrade.

Flink must understand the saved state created by the previous application version. If the new version changes the state representation in an incompatible way, restoration may fail.

Preserve State Descriptors

Consider a keyed state descriptor:

ValueStateDescriptor<Long> descriptor =
    new ValueStateDescriptor<>(
        "customer-total",
        Long.class
    );

ValueState<Long> totalState =
    getRuntimeContext().getState(descriptor);

The state descriptor name is part of the application's state identity. Changing customer-total to total-by-customer without a migration plan can prevent Flink from matching the old state to the new descriptor.

When renaming or restructuring state, verify the supported migration mechanism for the relevant state type and Flink version.

Preserve Operator UIDs

Flink uses operator identifiers to associate saved state with operators during restoration.

Explicit UIDs help keep those identities stable across code changes:

payments
    .keyBy(Payment::getCustomerId)
    .process(new CustomerSpendingFunction())
    .uid("customer-spending-processor");

Assign stable UIDs to stateful operators and preserve them when the logical operator remains the same.

Avoid generating UIDs dynamically or changing them unnecessarily. If an operator is intentionally replaced or its state ownership changes, plan that migration explicitly.

Check Serializer Compatibility

State serializers determine how values are stored and reconstructed.

Changing a state class, field type, or serializer may affect whether the saved state can be read by the new version.

For example, replacing a state value represented as a string with a custom object may require a compatible serializer or an explicit migration strategy.

Before upgrading, test the actual state schema changes using a representative savepoint. Do not assume that a source-code change is compatible simply because the new code compiles.

Restoring the Upgraded Job

Once the savepoint has been created and the new application version is ready, deploy the new job from that savepoint.

A simplified CLI command is:

flink run \
  -s s3://my-flink-bucket/savepoints/savepoint-abc123 \
  -d \
  application-v2.jar

Here, -s specifies the savepoint path and -d requests detached submission.

The exact command-line options and behavior can vary by Flink release. Verify the syntax supported by the installed CLI before using it in a deployment script.

The new job must use compatible state definitions and operators. If state is intentionally being discarded, use an explicit and reviewed procedure rather than silently ignoring restore errors.

After submission, verify that the job reaches the expected running state and that it continues processing from the correct source position.

Changing State During an Upgrade

Not every upgrade is a simple replacement of one JAR file with another.

A release may add a new state field, change a calculation, introduce a new operator, or modify the job's parallelism.

Adding a State Field

Adding a field to a state object can be compatible in some situations, but the result depends on the serializer and its compatibility rules.

Test the new state schema against a savepoint generated by the old application. Confirm that existing records can be restored and that newly introduced fields receive appropriate values.

Changing Operator Parallelism

Changing parallelism can affect how state is redistributed among operator instances.

Flink supports state redistribution for many state types, but the outcome depends on the operator, state type, and implementation. Some custom operator state may require additional care.

Test the intended parallelism change using the same kind of savepoint that will be used in production.

Changing the Processing Logic

A job can restore its state successfully and still produce incorrect results if the new processing logic interprets that state differently.

For example, changing how transaction totals are calculated may require more than restoring the previous total. The new logic must have a defined interpretation of the historical state.

State restoration and business correctness are separate checks. Both must pass before the upgrade is considered successful.

Common Upgrade Problems

Problem	Likely cause	Recommended action
State cannot be restored	Incompatible state schema or serializer	Review state descriptors and serializer compatibility
Operator state is missing	Operator UID changed or state mapping differs	Compare operator identifiers between releases
Savepoint cannot be accessed	Incorrect path or missing storage permissions	Verify the path, credentials, and filesystem configuration
Job starts but processes unexpected data	Source progress or business logic changed	Verify source restoration and compare output with expected results
Checkpoints keep failing	Storage, timeout, or state-size problems	Inspect checkpoint diagnostics and resource usage
New job fails during initialization	Missing connector, plugin, or configuration	Verify runtime dependencies and deployment settings

Testing the Upgrade Before Production

A successful compilation is not sufficient evidence that a stateful job can be upgraded safely.

Use a staging environment to test the complete process.

Run the current application with representative events.
Wait for a successful checkpoint and create a savepoint.
Deploy the new application using that savepoint.
Verify that the new job restores its state and source progress.
Compare its output with expected business results.
Trigger another checkpoint and verify that it completes.
Test failure recovery using the new application version.

The test should include realistic state volume, not just a few sample records. Large state snapshots can reveal restoration delays, memory pressure, and storage problems that do not appear in small tests.

For applications that write to external systems, verify sink behavior as well. Exactly-once processing guarantees depend on the source, checkpoint configuration, sink implementation, and external system's capabilities.

Advantages and Limitations

Using savepoints and compatible state definitions allows teams to perform controlled upgrades without rebuilding all streaming state from the beginning. It also supports planned changes to application code and deployment configuration.

However, savepoints do not make every upgrade automatically compatible. State schema changes, custom serializers, operator topology changes, and connector differences can require additional migration work.

Large savepoints can also take time to create, transfer, and restore. The deployment process should account for these delays, along with the time needed to verify application correctness.

Best Practices

Create and test savepoints before planned production upgrades.
Keep checkpoint storage durable and accessible to the replacement job.
Assign stable UIDs to stateful operators.
Avoid unnecessary changes to state descriptor names and serializers.
Test schema and parallelism changes against real savepoints.
Verify source positions, sink behavior, and business-level output after restoration.
Keep the previous application version and savepoint available during the rollback window.
Document the upgrade and recovery procedure so that another engineer can execute it.
Monitor checkpoint completion, restore duration, processing lag, and application errors after deployment.

Summary

Upgrading an Apache Flink job without losing streaming state requires a controlled process. Checkpoints provide regular fault recovery, while savepoints are useful for planned upgrades and migrations.

The most important work happens before deployment: identifying state dependencies, preserving operator identities, checking serializer compatibility, and testing restoration with a representative savepoint.

A successful upgrade is not just one that starts running. It must also resume from the expected source position, preserve the required state, and continue producing correct results.