[FLINK-37553] Make ForStKeyedStateBackend update NamespaceSerializer during restoring state

[mayuehappy]

…during restoring state

What is the purpose of the change

Make ForStKeyedStateBackend update NamespaceSerializer during restoring state

Brief change log

• add updateNamespaceSerializer in RegisteredKeyValueStateBackendMetaInfo
• invoke RegisteredKeyValueStateBackendMetaInfo#updateNamespaceSerializer when creating State

Verifying this change

(Please pick either of the following options)

This change is a trivial rework

This change added tests and can be verified as follows:

add unit test ForStStateMigrationTest#testStateNamespaceSerializerChanged

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)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (checkpointing)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not documented)

[flinkbot]

CI report:

• c786cfd Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[davidradl]

nit: is the New in the method name required?
registerNewSerializerForRestoredState -> registerSerializerForRestoredState

[Zakelly]

Well this is the name of existing interface, we'd better keep this in current PR.

[davidradl]

I am curious about the difference between isCompatibleAfterMigration() and isIncompatible(). If the difference is meaningful, should we include these different incompatibilities in the error message to help make the message more useful?

It would even better if the compatibility checks returned the reason for the incompatibility. Maybe throwing an exception in a checkIncompatibility method and construct different messages in the Exception might be better as we then log out the root cause of the incompatibility. WDYT?

[Zakelly]

@mayuehappy We could distinguish the messages for two cases. For the isCompatibleAfterMigration, we may say 'The new namespace serializer () can be compatible with the old one () after migration. But ForSt does not support state migration yet.'

[mayuehappy]

@Zakelly Thanks for the reply. As I remember, NamespaceSerializer should not support migration, even RocksdbStateBackend does. So do we still need to distinguish here?

[Zakelly]

@mayuehappy we can keep it if so.

[fredia]

@mayuehappy Thanks for the PR, LGTM