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
- wrap serializers (and maybe other objects) in getOrCreateKeyedState
- store wrapping objects in a new map keyed by state name
- pass wrapped objects to delegatedBackend.createInternalState
- on 1st user access, lookup wrapper and upgrade its wrapped serializer
This should be done for both KV/PQ states.
cc: Yun Tang