Runtime state recovery#

For resource locations, build/test invocation, capture, and dependency updates, see Test resources and version updates . This page records the recovery scenarios and their measured scope.

ProtobufRecoveryITCase executes the production serializer in a two-TaskManager MiniCluster with parallelism 2 and max parallelism 16. It uses stable integer stream keys, string MapState user keys, and generated messages as values in ValueState, MapState, and operator ListState. HashMap and RocksDB backends run the same cases. The ValueState implementation is checked at runtime to distinguish the selected backends.

The source stores its input cursor in Flink checkpoint state. Filesystem controls release input and request one failure; they do not supply restored values or a replacement input cursor. The checkpoint test saves eight records, processes four unsaved updates, and fails the source. After restart, the test checks restored operator values, replay from the saved input cursor, and exact per-key, per-map-key, and per-subtask counts through subsequent input. The checkpoint trigger completes before the unsaved updates and failure are permitted.

Separate jobs restore canonical savepoints with fixed operator UIDs and state names. The tests accept unchanged settings, increases in either or both reader limits, and both deterministic-mode transitions. Increased-limit jobs process values exceeding the old size or nesting budget, emit another savepoint, and restore that newly written state. Independent ValueState, MapState, and operator-state cases reject each limit decrease, mixed increase/decrease transitions, and decreases after an earlier increase and new snapshot emission. Rejection assertions inspect the failure cause for Flink’s incompatible state serializer exception.

Schema changes#

ProtobufSchemaEvolutionITCase restores savepoints into separate jobs whose application jar contains only the new generated classes. The runtime schemas keep the same Java class and Protobuf message names:

For each accepted change, the new job checks the restored operator values and continued processing, writes another savepoint, and restores it again. Each processed input also reports the ValueState and MapState values it replaced, and the test decodes them as well; for both state kinds, these include values written by the old schema and values written by the new application before its savepoint. Values written after the restore carry the new field or enum value, while restored values keep what the old schema wrote; an added field therefore has its default there. The reserved job keeps the removed payload as an unknown field, so its savepoint still contains the payload that the original schema reads back. The rejected cases restore original state with incompatible separately for ValueState, MapState values, and operator ListState, changed state with original, and enum-added state with enum-original. Each rejected job must fail, and its failure must name the generated class and the first incompatibility . The incompatible case shows the diagnostic for each state kind on both backends; the two reverse-direction cases register all three state kinds, so the first state that fails reports it. A decreased limit fails with Flink’s StateMigrationException alone. These tests write and restore state with the same development build; the retained-state upgrade check applies the same checks to the state of a retained bundle.

For ProtobufRecoveryITCase and ProtobufSchemaEvolutionITCase, the application and generated classes live in separate test jars that are absent from the test runner’s parent classpath. Each job is constructed with a fresh application loader and submitted with its application jar attached. The application checks that recovered messages belong to its task-side application loader and retain nested values and unknown fields. The test runner checks that neither the application entry point nor its generated message can load through the parent. Generic types are disabled, no module-opening JVM flags are added, and the existing fresh-JVM missing-registration controls remain required. The five production transport shapes additionally retain the native snapshot implementation inside their composite serializers.

DynamicMessage#

ProtobufDynamicMessageRecoveryITCase repeats the checkpoint and schema-change cases with DynamicMessage values and descriptor-bound type information . Its application, DynamicRuntimeJob, keeps the operator UIDs and state names of RuntimeJob and builds its descriptor from the schema variant’s descriptor set. Its jar, application-dynamic.jar, contains no generated class, and the test runner checks that the generated Entry class cannot load through the application loader. The checkpoint case and the accepted changed, reserved, and enum-added restores check the same restored values and continued processing as the generated cases, and an unchanged original restore runs the same checks.

Each observation also reports the root descriptor of the observed value, and the test names the schema variant whose descriptor has that content. After a restore of state that another schema wrote, values from operator ListState, and ValueState and MapState values on HashMap, carry the saved descriptor; RocksDB values carry the new one. After the second restore, of the savepoint that the new schema wrote, every restored value carries the new descriptor. This includes the MapState values that the first restored job left untouched until its savepoint. On HashMap, the new serializer wrote those values from objects that, like the other restored values, are expected to carry the saved descriptor; the test observes them only after the second restore. RocksDB copies their stored bytes into the savepoint.

The rejected cases are those of the generated test. Each rejected job fails with Flink’s StateMigrationException for an incompatible new serializer, without a library diagnostic, as ADR-0005 states. For each one, the same saved state restores into the writer’s own schema and continues processing. Saved state of the other kind fails as well:

ProtobufDynamicMessageTransportITCase runs local jobs with generic types disabled, RocksDB keyed state, and descriptors that no class on the test classpath backs. The original and enum-original schemas have the same Java class and Protobuf full name; their type information differs, and one job transports values of both and counts them in separate ValueStates, which RocksDB writes and reads with each type’s serializer. A stream typed with the original schema fails with the writer’s IOException for values of another root, for enum-original values, and for a nested value of the incompatible schema; the message names the full-name change or the field path and the first incompatibility.

Fixtures and compatibility boundaries#

The repository commits no savepoint fixtures: the 1.0.0 format restart (#140 ) removed the development savepoints and snapshot-format fixtures that 0.x builds wrote, because the current library does not read them. Fixtures written by the published 1.0.0 artifacts are retained under #13 .

The suites run in the existing Flink 2.2.1/2.3.0 JDK 17/21 and Flink 1.20.5 JDK 17 matrix, with paired Protobuf 4.36.2 and 3.25.9 generated code and runtime. The unchanged-jar lanes retain the floor’s application jars, descriptors, and compiled application classes and verify their hashes after execution at the ceiling. They also run each line’s Protobuf 4 jar against these suites with application classes and descriptors generated by protoc 3.25.9. Application jars are built before tests and never recompiled by the tests or attached as publication artifacts.

Each saved-state scenario keeps its Flink version, Protobuf profile, backend, parallelism, and state identity fixed. The suite does not establish backend migration, rescaling, cross-Flink state upgrades, cross-Protobuf restore, or cross-process message-key hashing. CommonTypesRecoveryITCase adds Well-Known Type and OpenTelemetry acceptance , using the same MiniCluster controls with a separate job and values on the ordinary test classpath. It verifies direct Struct ValueState, direct AnyValue MapState values, and an application envelope in operator ListState on both backends. Issue #40’s development bundle tooling captures packaged writers and compiled consumers with retained state for later release checks; published-artifact capture and validation of both release lines remain in #13.