[FLINK-40297][runtime] Route TTL-aware value migration through the migrate hook

[weiqingy]

This is the second PR of the FLIP-527 implementation, split into a stack of small, independently reviewable PRs under the umbrella issue FLINK-37732. Landing order:

Step	Sub-task	Scope
PR-1	FLINK-40296	Object-level migrate hook on TypeSerializerSnapshot
PR-2 (this PR)	FLINK-40297	Route TTL-aware value migration through the hook
PR-3	FLINK-40298	Opt-in name-based schema evolution for RowData
PR-4	FLINK-40299	End-to-end state migration coverage on RocksDB

Draft until PR-1 merges: this branch contains PR-1's commit, so the diff against master shows both. It will be rebased onto master and marked ready once PR-1 lands. PR-1 and PR-2 are behavior-neutral, since the hook defaults to returning its argument and nothing overrides it until PR-3.

What is the purpose of the change

TtlAwareSerializer.migrateValueFromPriorSerializer is the single entry point through which the RocksDB state backend migrates state values on restore. AbstractRocksDBState, RocksDBListState and RocksDBMapState all call it, each after unwrapping the state shape it owns: the value serializer, the list element serializer, or the map value serializer.

Today it deserializes with the prior serializer and re-serializes with the new one, giving a serializer no opportunity to adapt the value in between. This routes it through the migrate hook added in PR-1, so a serializer that overrides the hook takes effect. Behavior is unchanged, because nothing overrides it yet and the default returns the value unchanged.

Two details are worth calling out for review, because both are contract decisions rather than mechanics.

The hook receives the persisted prior snapshot, not a re-derived one. The prior serializer reaching this method is itself previousSerializerSnapshot.restoreSerializer(), so calling snapshotConfiguration() on it is a snapshot to serializer to snapshot round trip. That round trip is lossy: PojoSerializerSnapshot inserts a null Field for a field that no longer exists on the class, and re-snapshotting substitutes a synthetic name for it. An override reconciling fields by name, which is what FLIP-527 adds in PR-3, would then see a fabricated schema. The backend already fetches the persisted snapshot above the migration loop, so this threads it down and each caller descends it alongside the serializer it already descends.

The descent unwraps the TtlAware decorator first, and an unexpected snapshot type now fails rather than falling back. Registering a new serializer mutates the previous snapshot's nested snapshots in place, so a list or map state's persisted element or value snapshot is a TtlAwareSerializerSnapshot rather than the snapshot the checkpoint wrote. An earlier revision of this change assumed otherwise and fell back silently to the re-derived snapshot on exactly those paths; the assertion added here is what surfaced it. Only an absent snapshot falls back now, which is the case where nothing was persisted and the re-derived snapshot is all that exists.

Brief change log

• Unwrap the prior value to its bare, non-TTL form, pass it through migrate, and re-wrap when this serializer is TTL-enabled, preserving the prior TTL timestamp when the prior value carried one
• Thread the persisted prior snapshot from the backend down to the hook, descending it alongside the serializer at each of the three call sites
• Fail with a clear message when the persisted snapshot is of an unexpected type, instead of silently re-deriving one
• Pin the nullability of the hook's value parameter, and note that old and new snapshots need not expose the same nested snapshot types

Verifying this change

This change added tests and can be verified as follows:

• TtlAwareSerializerTest covers all four combinations of prior and current TTL being enabled, asserting migrated bytes and the preserved timestamp. These were written to pass against the pre-change code, so they demonstrate that behavior is unchanged rather than asserting it
• Tests pin which snapshot reaches the hook, by object identity: a re-derived snapshot has the same class and a different identity, so identity is the only discriminator. Removing either the TtlAware unwrap or the TTL descent fails these
• Tests assert the type guards fire, in both directions and behind the TtlAware layer. Softening either guard alone still fails
• StateSerializerProviderTest pins the in-place mutation of nested snapshots that the descent depends on, so a change to that mechanism fails here rather than silently downstream
• Covered end to end by the existing RocksDB state migration suites, which exercise this method through all three state shapes in both TTL directions

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no. TypeSerializerSnapshot is @PublicEvolving but only its javadoc changed. Three internal backend method signatures widened; none is on an interface or carries a stability annotation, and japicmp is green.
• The serializers: yes
• The runtime per-record code paths (performance sensitive): no for the steady-state record path. This method runs once per state entry during restore, so the snapshot lookup is hoisted out of the loop and guard messages are built only on failure.
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes, this is on the checkpoint and savepoint restore path, though behavior is unchanged
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no, it is preparatory for FLIP-527
• If yes, how is the feature documented? JavaDocs

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5)

[flinkbot]

CI report:

• 9103aec Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build