[FLINK-11196] Extend S3 EntropyInjector to use key replacement (instead of key removal) when creating checkpoint metadata files.

[markcho]

[FLINK-11196] Extend S3 EntropyInjector to use key replacement (instead of key removal) when creating checkpoint metadata files.

What is the purpose of the change

Extend S3 entropy injection feature to optionally replace the entropy marker in the path (instead of just removing the marker) when writing checkpoint metadata files.

It introduces a new configuration s3.entropy.replacement which specifies the string that will be use to replace the entropy marker with (only when writing checkpoint metadata files). Checkpoint state files entropy injection behavior does not change.

If s3.entropy.replacement is not configured, then the behavior will be the same as it is today. Entropy marker will be removed from checkpoint metadata file path and entropy will be injected for checkpoint state file paths. This PR is an optional extension to the current feature. Only if s3.entropy.replacement is set to a non-empty string, it will use the value to replace the entropy marker when writing checkpoint metadata files.

Verifying this change

• Unit tests
• Integration tests with entropy injection enabled (manually verified)

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, Yarn/Mesos, ZooKeeper: yes
• The S3 file system connector: yes

Documentation

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