[FLINK-37351][runtime] Ensure Writer/Committer Colocation for Non-Pre-Commit Topologies

[rionmonster]

What is the purpose of the change

This pull request addresses the issue detailed in FLINK-37351 which aims to ensure writer/committer colocation, specifically for non-pre-commit topologies, which should improve overall performance.

Brief change log

• Added a process within the existing SinkTransformationTranslator.addCommittingTopology() function to detect non-pre-commit topologies and define a colocation-group key for the transformation (via setCoLocationGroupKey(...)).

Verifying this change

This change added a series of tests in SinkV2TransformationTranslatorITCase to verify that colocation wasn't being applied (via a red-green cycle) as well as two other tests to confirm colocation was working for expected cases:

• testWriterAndCommitterColocatedWithoutPreCommitTopology to verify writer/committer colocation for non-pre-commit topologies
• testWriterAndCommitterNotColocatedWithPreCommitTopology to verify pre-commit sinks are not colocated
• testUserSpecifiedCoLocationGroupIsRespected to verify user-specified colocation groups are preserved

Example Tests

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: no/don't know
• 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

Reviewer Requested

@AHeise

[flinkbot]

CI report:

• 0cdc92f Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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