[FLINK-36001][state] Store operator name and UID in state metadata

[gaborgsomogyi]

What is the purpose of the change

Normally users can define an operator in a Flink application like this where UID and name can be assigned to it:

stream.map(new ValueProcessUDF())
       .uid(“value-process-uid”)
       .name(“value-process-name”)

At the moment checkpoint metadata file is not containing operator UID and name which makes hard for users to find out what is the human readable intention for a specific operator. Please see FLIP-474 further details.

In this PR I've added the 2 mentioned fields into metadata file.

Brief change log

Added operator UID and name to metadata file.

Verifying this change

• Existing + modified + new automated tests
• Manually on cluster (read, write, operator UID change)

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

Documentation

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

[flinkbot]

CI report:

• 142e5f1 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[gaborgsomogyi]

In order to help reviewers the general concept is the following. UID and name fields are optional and as such they're null in the initial part of the code chain. I've kept the null values until MetadataV5Serializer where empty string is stored. The reverse way is to convert empty string as null to keep consistency. The other possibility was to come up with a new optional string storage mechanism which I've tried but in my view it would just add extra complexity with relatively high risk and no gain.

[gaborgsomogyi]

Some extra coverage where only the hash changed.

[gaborgsomogyi]

The intention here with the extra assertions is to make sure that UID and hash are consistent with each other.

[gaborgsomogyi]

Such cases I've considered to add a constructor where UID and name fields are just hardcoded null values (then we wouldn't need so many code changes in tests) but then I've ended up to miss production code parts where the mentioned fields are essential which caused partially working solution. I think having a rock state state story is must have from user's perspective so I've implemented this a more defensive way (compiler blows up not having enough constructor parameters) in order to change all the required execution paths.

[gaborgsomogyi]

cc @gyfora

[gyfora]

Thanks @gaborgsomogyi , overall looks good. Please add 1-2 comments to the critical parts that you also explain in the PR.

[gaborgsomogyi]

Just rebased and added some explanation to the code.

[gaborgsomogyi]

@Zakelly any comments/thought?

[Zakelly]

@Zakelly any comments/thought?

Thanks for the heads up, looking into this...

[Zakelly]

Thanks for the PR, I left some comments, PTAL.

[Zakelly]

Thanks for the update! Overall LGTM.

[fapaul]

@gaborgsomogyi this PR broke the master build by missing the license header for OperatorIDPairTest can you fix that?

[gaborgsomogyi]

Oh gosh, fixing it now.

[gaborgsomogyi]

Just filed: #25329