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

flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DefaultDispatcherResourceManagerComponentFactory.java

@@ -176,6 +176,10 @@ public DispatcherResourceManagerComponent create(
             log.debug("Starting Dispatcher REST endpoint.");
             webMonitorEndpoint.start();
 
+            configuration.setInteger(RestOptions.PORT, webMonitorEndpoint.getRestPort());
+            configuration.setString(
+                    RestOptions.ADDRESS, webMonitorEndpoint.getServerAddress().getHostString());
+
             final String hostname = RpcUtils.getHostname(rpcService);
 
             resourceManagerService =

flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/ClientHighAvailabilityServices.java

@@ -20,6 +20,8 @@
 
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
 
+import java.net.UnknownHostException;
+
 /**
  * {@code ClientHighAvailabilityServices} provides services those are required on client-side. At
  * the moment only the REST endpoint leader retriever is required because all communication between
@@ -32,5 +34,5 @@ public interface ClientHighAvailabilityServices extends AutoCloseable {
      *
      * @return the leader retriever for cluster's rest endpoint.
      */
-    LeaderRetrievalService getClusterRestEndpointLeaderRetriever();
+    LeaderRetrievalService getClusterRestEndpointLeaderRetriever() throws UnknownHostException;
 }

flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java

@@ -28,6 +28,7 @@
 import org.apache.flink.util.concurrent.FutureUtils;
 
 import java.io.IOException;
+import java.net.UnknownHostException;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
@@ -202,7 +203,8 @@ default LeaderElectionService getClusterRestEndpointLeaderElectionService() {
     }
 
     @Override
-    default LeaderRetrievalService getClusterRestEndpointLeaderRetriever() {
+    default LeaderRetrievalService getClusterRestEndpointLeaderRetriever()
+            throws UnknownHostException {
         // for backwards compatibility we delegate to getWebMonitorLeaderRetriever
         // all implementations of this interface should override
         // getClusterRestEndpointLeaderRetriever, though

flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServicesUtils.java

@@ -128,11 +128,9 @@ public static HighAvailabilityServices createHighAvailabilityServices(
                                 RpcServiceUtils.createWildcardName(Dispatcher.DISPATCHER_NAME),
                                 addressResolution,
                                 configuration);
-                final String webMonitorAddress =
-                        getWebMonitorAddress(configuration, addressResolution);
 
                 return new StandaloneHaServices(
-                        resourceManagerRpcUrl, dispatcherRpcUrl, webMonitorAddress);
+                        resourceManagerRpcUrl, dispatcherRpcUrl, configuration);
             case ZOOKEEPER:
                 return createZooKeeperHaServices(configuration, executor, fatalErrorHandler);
             case KUBERNETES:

flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServices.java

@@ -19,12 +19,17 @@
 package org.apache.flink.runtime.highavailability.nonha.standalone;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
 import org.apache.flink.runtime.highavailability.nonha.AbstractNonHaServices;
 import org.apache.flink.runtime.leaderelection.LeaderElectionService;
 import org.apache.flink.runtime.leaderelection.StandaloneLeaderElectionService;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
 import org.apache.flink.runtime.leaderretrieval.StandaloneLeaderRetrievalService;
+import org.apache.flink.runtime.rpc.AddressResolution;
+
+import java.net.UnknownHostException;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -45,23 +50,20 @@ public class StandaloneHaServices extends AbstractNonHaServices {
     /** The fix address of the Dispatcher. */
     private final String dispatcherAddress;
 
-    private final String clusterRestEndpointAddress;
+    private final Configuration configuration;
 
     /**
      * Creates a new services class for the fix pre-defined leaders.
      *
      * @param resourceManagerAddress The fix address of the ResourceManager
-     * @param clusterRestEndpointAddress
+     * @param configuration
      */
     public StandaloneHaServices(
-            String resourceManagerAddress,
-            String dispatcherAddress,
-            String clusterRestEndpointAddress) {
+            String resourceManagerAddress, String dispatcherAddress, Configuration configuration) {
         this.resourceManagerAddress =
                 checkNotNull(resourceManagerAddress, "resourceManagerAddress");
         this.dispatcherAddress = checkNotNull(dispatcherAddress, "dispatcherAddress");
-        this.clusterRestEndpointAddress =
-                checkNotNull(clusterRestEndpointAddress, clusterRestEndpointAddress);
+        this.configuration = configuration;
     }
 
     // ------------------------------------------------------------------------
@@ -134,10 +136,13 @@ public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
     }
 
     @Override
-    public LeaderRetrievalService getClusterRestEndpointLeaderRetriever() {
+    public LeaderRetrievalService getClusterRestEndpointLeaderRetriever()
+            throws UnknownHostException {
+        String clusterRestEndpointAddress =
+                HighAvailabilityServicesUtils.getWebMonitorAddress(
+                        configuration, AddressResolution.NO_ADDRESS_RESOLUTION);
         synchronized (lock) {
             checkNotShutdown();
-
             return new StandaloneLeaderRetrievalService(
                     clusterRestEndpointAddress, DEFAULT_LEADER_ID);
         }

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java

@@ -25,7 +25,6 @@
 import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
 import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.configuration.UnmodifiableConfiguration;
 import org.apache.flink.core.plugin.PluginManager;
 import org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils;
 import org.apache.flink.util.Preconditions;
@@ -41,7 +40,7 @@ public class MiniClusterConfiguration {
 
     static final int DEFAULT_IO_POOL_SIZE = 4;
 
-    private final UnmodifiableConfiguration configuration;
+    private final Configuration configuration;
 
     private final int numTaskManagers;
 
@@ -72,7 +71,7 @@ public MiniClusterConfiguration(
         this.pluginManager = pluginManager;
     }
 
-    private UnmodifiableConfiguration generateConfiguration(final Configuration configuration) {
+    private Configuration generateConfiguration(final Configuration configuration) {
         final Configuration modifiedConfig = new Configuration(configuration);
 
         TaskExecutorResourceUtils.adjustForLocalExecution(modifiedConfig);
@@ -94,7 +93,7 @@ private UnmodifiableConfiguration generateConfiguration(final Configuration conf
             modifiedConfig.set(AkkaOptions.ASK_TIMEOUT_DURATION, Duration.ofMinutes(5L));
         }
 
-        return new UnmodifiableConfiguration(modifiedConfig);
+        return new Configuration(modifiedConfig);
     }
 
     // ------------------------------------------------------------------------
@@ -146,7 +145,7 @@ public String getTaskManagerBindAddress() {
                 : configuration.getString(TaskManagerOptions.BIND_HOST, "localhost");
     }
 
-    public UnmodifiableConfiguration getConfiguration() {
+    public Configuration getConfiguration() {
         return configuration;
     }
 

flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServicesTest.java

@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.highavailability.nonha.standalone;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.leaderelection.LeaderContender;
 import org.apache.flink.runtime.leaderelection.LeaderElectionService;
@@ -39,16 +40,14 @@ public class StandaloneHaServicesTest extends TestLogger {
 
     private final String dispatcherAddress = "dispatcher";
     private final String resourceManagerAddress = "resourceManager";
-    private final String webMonitorAddress = "webMonitor";
 
     private StandaloneHaServices standaloneHaServices;
 
     @Before
     public void setupTest() {
-
+        Configuration config = new Configuration();
         standaloneHaServices =
-                new StandaloneHaServices(
-                        resourceManagerAddress, dispatcherAddress, webMonitorAddress);
+                new StandaloneHaServices(resourceManagerAddress, dispatcherAddress, config);
     }
 
     @After

flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/TestingMiniCluster.java

@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.minicluster;
 
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
 import org.apache.flink.runtime.blob.BlobServer;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.dispatcher.DispatcherGateway;
@@ -156,11 +157,13 @@ protected HighAvailabilityServices createHighAvailabilityServices(
         final List<DispatcherResourceManagerComponent> result =
                 new ArrayList<>(numberDispatcherResourceManagerComponents);
 
+        String configuredPort = configuration.getString(RestOptions.BIND_PORT);
         for (int i = 0; i < numberDispatcherResourceManagerComponents; i++) {
             // FLINK-24038 relies on the fact that there is only one leader election instance per
             // JVM that is freed when the JobManager stops. This is simulated in the
             // TestingMiniCluster by providing individual HighAvailabilityServices per
             // DispatcherResourceManagerComponent to allow running more-than-once JobManager tests
+
             final HighAvailabilityServices thisHaServices =
                     createHighAvailabilityServices(configuration, getIOExecutor());
             final DispatcherResourceManagerComponent dispatcherResourceManagerComponent =
@@ -200,6 +203,10 @@ protected HighAvailabilityServices createHighAvailabilityServices(
                                     });
             FutureUtils.assertNoException(shutDownFuture);
             result.add(dispatcherResourceManagerComponent);
+            if (i + 1 < numberDispatcherResourceManagerComponents) {
+                // Reset the port to original
+                configuration.setString(RestOptions.BIND_PORT, configuredPort);
+            }
         }
 
         return result;