Upgrading Flink Jobs: Turn Parallel Runs and Diffs into Evidence
Separate state compatibility, result correctness, and performance regressions to make streaming migrations diagnosable and reversible.
Read in Chinese →Engineering notes with synthetic examples. Version scope and references are included in each article.
A Flink job starting after an upgrade proves that the new program can run. It does not prove that all old state was restored or that output semantics are unchanged. Migration acceptance needs three independent forms of evidence: recoverable state, correct business results, and acceptable operational behavior.
This article discusses migration to Flink 1.15 as a baseline. When crossing several versions, check the compatibility notes for each runtime and connector. “We use a savepoint” is not a guarantee that any two versions support a direct upgrade.
Freeze what you are comparing
Combining a runtime upgrade, business-logic changes, schema changes, and a source migration in one release makes differences difficult to explain. Separate them when practical. If they must ship together, define an expected result for each category of change.
A migration inventory should include Flink and Java versions, connector versions, operator UIDs, state names, serializers, key types, maximum parallelism, checkpoint settings, time zones, external dependency versions, and sink semantics. Many apparent runtime-upgrade failures are actually changed defaults or dependency conflicts.
Flink savepoints associate state with operator identity. The documentation recommends explicit UIDs to reduce the effect of topology changes on automatically generated identifiers. Flink 1.15: Savepoints
Stable UIDs are only one requirement. Whether a state-field change is compatible depends on the serializer. Changes to a key’s schema in particular cannot be treated as ordinary field evolution. Flink 1.15: State Schema Evolution
Share facts between parallel runs, not side effects
The old and new jobs should read the same logical input interval and write to isolated result sets. Kafka consumers can use separate groups with aligned partition boundaries. File replay should use a frozen inventory and object versions. A shadow job must not trigger duplicate charges, notifications, or irreversible writes.
External dimension data is an easily missed variable. If the old job reads user label X in the morning and the new job reads Y in the afternoon, identical code can still produce different results. Freeze dimension snapshots, record external responses, or pin versions during validation. When that is impossible, track these nondeterministic fields separately instead of attributing every difference to the upgrade.
Make the warm-up boundary explicit as well. A new job starting with empty deduplication state, windows, and caches will naturally differ from a long-running job. Use comparable initial state, or identify a warm-up interval and compare only an adequately covered event range.
Move from record-level diffs to business meaning
First compare accounting identities: inputs, filtered records, errors, and outputs. Next compare sets by business key. Only then compare fields on records present on both sides.
left_only Present in the old pipeline, absent in the new one
right_only Present in the new pipeline, absent in the old one
field_diff The same business key has different field values
duplicate One side contains multiple records for a business key
Detect duplicates before joining; otherwise a many-to-many join can inflate differences. Compare nested maps by their key-value meaning. Apply floating-point tolerances only where the business permits them. Maintain an explicit ignore list for nonbusiness fields such as processing timestamps and trace IDs. Do not hide every inconvenient difference behind an unexplained regular expression.
For windowed aggregates, compare results for settled windows, not two arbitrary live snapshots. Align the event-time interval, watermark progress, late-data handling, and final-result version. Matching overall totals is insufficient: an increase in one source can cancel a decrease in another.
Verify recovery before cutover and retain repair options afterward
Run at least one rehearsal of “restore the new version from the intended state → process controlled input → fail and recover again.” Inspect timers, deduplication state, and external writes. Before enabling an option that allows unrestored state, identify exactly what would be abandoned. It should not be a generic switch for making startup turn green.
The acceptance report can be short, but its evidence must be traceable:
| Dimension | Evidence |
|---|---|
| Correctness | Comparison interval, sample size, difference categories, explanations |
| State | Savepoint restoration outcome and assertions on critical state |
| Performance | Latency, backlog, CPU, state size, and checkpoints under comparable load |
| Recovery | Time to recover from an injected failure; duplicate and missing event counts |
| Rollback | Old artifacts, recoverable state, input retention, and treatment of external output |
Rolling back the runtime does not undo records already written by the new version. Design idempotency keys or repair batches before migration, and verify that the old version can read the state intended for rollback. Treating rollback as “deploy the old package again” misses its hardest part.
References
- Flink 1.15: Savepoints , for operator UIDs and restoration behavior.
- Flink 1.15: State Schema Evolution , for serialization compatibility constraints.