[FLINK-23143] Support state migration

• Type: Sub-task
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Fix Version/s: 1.16.0
• Component/s: Runtime / State Backends
• Labels: pull-request-available
• Assignee: Hangxiang Yu
• Reporter: Roman Khachatryan
• Created: 24/Jun/21 11:42
• Updated: 08/Aug/22 14:04
• Resolved: 08/Aug/22 14:04

Description

ChangelogKeyedStateBackend.getOrCreateKeyedState is currently used during recovery; on 1st user access, it doesn't update metadata nor migrate state (as opposed to other backends).

 

The proposed solution is to

1. wrap serializers (and maybe other objects) in getOrCreateKeyedState
2. store wrapping objects in a new map keyed by state name
3. pass wrapped objects to delegatedBackend.createInternalState
4. on 1st user access, lookup wrapper and upgrade its wrapped serializer

This should be done for both KV/PQ states.

 

See also https://github.com/apache/flink/pull/15420#discussion_r656934791

 

cc: yunta

Issue Links

• is blocked by: FLINK-21356 Implement incremental checkpointing and recovery using state changelog
• relates to: FLINK-26853 HeapStateBackend ignores metadata updates in certain cases
• relates to: FLINK-28553 The serializer in StateMap has not been updated when metaInfo of StateTable updated
• links to: GitHub Pull Request #19679
• links to: Support state migration for Changelog