[FLINK-23143][state/changelog] Support state migration for ChangelogS…

[masteryhx]

What is the purpose of the change

This pull request makes Changelog support state migration.

Brief change log

• added upgradeKeyedState(final TypeSerializer<N> namespaceSerializer, StateDescriptor<S, V> stateDescriptor) in AbstractKeyedStateBackend
• added upgrade( @Nonnull String stateName, @Nonnull TypeSerializer<T> byteOrderedElementSerializer) in PriorityQueueSetFactory

Verifying this change

• added ChangelogStateBackendMigrationTest

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: no
• The S3 file system connector: no

Documentation

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

[flinkbot]

CI report:

• 6790a53 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[rkhachatryan]

Thanks for updating the PR @masteryhx and writing the design document.

I've left some comments, PTAL (let's see if the document needs to be updated).

Please also check the cause of the build failure.

[rkhachatryan]

Thanks for updating the PR @masteryhx
I've left some comments, PTAL.

[rkhachatryan]

Thanks for updating the PR @masteryhx
I've left some comments, PTAL.

There is also a TODO associated with this ticket:

        // todo: support changing ttl (FLINK-23143)

which is currently not addressed.

[masteryhx]

Thanks for updating the PR @masteryhx I've left some comments, PTAL.

There is also a TODO associated with this ticket:

        // todo: support changing ttl (FLINK-23143)

which is currently not addressed.

As we discussed offline, it should be supported in the delegated state backend.
I have added some tests in StateBackendMigrationTestBase.

[rkhachatryan]

Thanks for updating the PR @masteryhx
Production code changes LGTM.

I have some concerns regarding the test coverage, PTAL at the comments.

Besides that, commit 86bb0c2 seems not belonging to this PR;
and commits d49670b and 05d91de should be merged together.
So in the end I think there should be just two commits:

• [refactor][state] Rename createInternalState to createOrUpdateInternalState
• [FLINK-23143][state/changelog] Support state migration for ChangelogStateBackend

WDYT?

[masteryhx]

Thanks for updating the PR @masteryhx Production code changes LGTM.

I have some concerns regarding the test coverage, PTAL at the comments.

Besides that, commit 86bb0c2 seems not belonging to this PR; and commits d49670b and 05d91de should be merged together. So in the end I think there should be just two commits:

• [refactor][state] Rename createInternalState to createOrUpdateInternalState
• [FLINK-23143][state/changelog] Support state migration for ChangelogStateBackend

WDYT?

I agree.
I have also created a seprate pr for updating the documentation about changing changelog from enabling to disabling
as you could see #20486

[rkhachatryan]

Thanks a lot for working on this PR @masteryhx,
LGTM.