[FLINK-31518][Runtime / REST] Fix StandaloneHaServices#getClusterRestEndpointLeaderRetreiver to return correct rest port

[Samrat002]

What is the purpose of the change

Currently, StandaloneHaServices#getClusterRestEndpointLeaderRetreiver returns wrong rest port when a range of ports is configured in rest.port. This patch addresses the same.

Brief change log

DefaultDispatcherResourceManagerComponentFactory sets the configuration with the rest address and rest port, from which the StandaloneHaServices#getClusterRestEndpointLeaderRetriever retrieves the address and port.

Verifying this change

This change added tests and can be verified as follows:

1. Run a flink job with config rest.port set to a range say 50200-50300.
2. Included below code in a thread launched by the Adaptive Scheduler.

LeaderRetrievalService webMonitorRetrievalService = highAvailabilityServices.getClusterRestEndpointLeaderRetriever();
try {
webMonitorRetrievalService.start(new WebMonitorLeaderListener());
} catch (Exception e) {
throw new RuntimeException(e);
}

private class WebMonitorLeaderListener implements LeaderRetrievalListener {
@OverRide
public void notifyLeaderAddress(final String leaderAddress, final UUID leaderSessionID) {
System.out.println(leaderAddress);
}

3. The leader address printed is the correct one which RestServerEndpoint listens.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
• The serializers: (yes / no / don't know)
• The runtime per-record code paths (performance sensitive): (yes / no / don't know)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
• The S3 file system connector: (yes / no / don't know)

Documentation

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

[flinkbot]

CI report:

• 8c5f8d0 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[Samrat002]

@huwh please review the changes in free time

[huwh]

Thanks for preparing this PR. I left some comments.

Currently there is no internal components in JobManager will call web monitor. So maybe we could not change the HighAvailableService and just update this configuration when WebMonitor is started is enough.

Then you can use ClientHighAvailabilityServices to retrieve the right web address and port.

And looking forward to your opinion @dannycranmer

[huwh]

Do we need update this Address?

"getServerAddress()" may produce NPE here.

[huwh]

Do we really need these change if you use ClientHighAvailabilityServices to retrieve the rest address and port?