[hotfix][runtime] Code deduplication in ResourceManagerFactory and its implementations.

flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneJobClusterEntryPoint.java

@@ -80,7 +80,7 @@ public final class StandaloneJobClusterEntryPoint extends ClusterEntrypoint {
 				new DefaultDispatcherRunnerFactory(
 						ApplicationDispatcherLeaderProcessFactoryFactory
 								.create(configuration, SessionDispatcherFactory.INSTANCE, program)),
-				StandaloneResourceManagerFactory.INSTANCE,
+				StandaloneResourceManagerFactory.getInstance(),
 				JobRestEndpointFactory.INSTANCE);
 	}
 

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/entrypoint/KubernetesResourceManagerFactory.java

@@ -38,6 +38,7 @@ import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServices;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServicesConfiguration;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.ConfigurationException;
 
 import javax.annotation.Nullable;
 
@@ -57,7 +58,7 @@ public class KubernetesResourceManagerFactory extends ActiveResourceManagerFacto
 	}
 
 	@Override
-	public ResourceManager<KubernetesWorkerNode> createActiveResourceManager(
+	public ResourceManager<KubernetesWorkerNode> createResourceManager(
 			Configuration configuration,
 			ResourceID resourceId,
 			RpcService rpcService,
@@ -66,14 +67,9 @@ public class KubernetesResourceManagerFactory extends ActiveResourceManagerFacto
 			FatalErrorHandler fatalErrorHandler,
 			ClusterInformation clusterInformation,
 			@Nullable String webInterfaceUrl,
-			ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception {
-		final ResourceManagerRuntimeServicesConfiguration rmServicesConfiguration =
-			ResourceManagerRuntimeServicesConfiguration.fromConfiguration(
-				configuration, KubernetesWorkerResourceSpecFactory.INSTANCE);
-		final ResourceManagerRuntimeServices rmRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(
-			rmServicesConfiguration,
-			highAvailabilityServices,
-			rpcService.getScheduledExecutor());
+			ResourceManagerMetricGroup resourceManagerMetricGroup,
+			ResourceManagerRuntimeServices resourceManagerRuntimeServices) {
+
 		final KubernetesResourceManagerConfiguration kubernetesResourceManagerConfiguration =
 			new KubernetesResourceManagerConfiguration(
 				configuration.getString(KubernetesConfigOptions.CLUSTER_ID),
@@ -85,13 +81,19 @@ public class KubernetesResourceManagerFactory extends ActiveResourceManagerFacto
 			configuration,
 			highAvailabilityServices,
 			heartbeatServices,
-			rmRuntimeServices.getSlotManager(),
+			resourceManagerRuntimeServices.getSlotManager(),
 			ResourceManagerPartitionTrackerImpl::new,
-			rmRuntimeServices.getJobLeaderIdService(),
+			resourceManagerRuntimeServices.getJobLeaderIdService(),
 			clusterInformation,
 			fatalErrorHandler,
 			resourceManagerMetricGroup,
 			KubeClientFactory.fromConfiguration(configuration),
 			kubernetesResourceManagerConfiguration);
 	}
+
+	@Override
+	protected ResourceManagerRuntimeServicesConfiguration createResourceManagerRuntimeServicesConfiguration(
+		Configuration configuration) throws ConfigurationException {
+		return ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration, KubernetesWorkerResourceSpecFactory.INSTANCE);
+	}
 }

flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManagerFactory.java

@@ -36,6 +36,7 @@ import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServices;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServicesConfiguration;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.ConfigurationException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -62,7 +63,7 @@ public class MesosResourceManagerFactory extends ActiveResourceManagerFactory<Re
 	}
 
 	@Override
-	public ResourceManager<RegisteredMesosWorkerNode> createActiveResourceManager(
+	public ResourceManager<RegisteredMesosWorkerNode> createResourceManager(
 			Configuration configuration,
 			ResourceID resourceId,
 			RpcService rpcService,
@@ -71,13 +72,8 @@ public class MesosResourceManagerFactory extends ActiveResourceManagerFactory<Re
 			FatalErrorHandler fatalErrorHandler,
 			ClusterInformation clusterInformation,
 			@Nullable String webInterfaceUrl,
-			ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception {
-		final ResourceManagerRuntimeServicesConfiguration rmServicesConfiguration =
-			ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration, MesosWorkerResourceSpecFactory.INSTANCE);
-		final ResourceManagerRuntimeServices rmRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(
-			rmServicesConfiguration,
-			highAvailabilityServices,
-			rpcService.getScheduledExecutor());
+			ResourceManagerMetricGroup resourceManagerMetricGroup,
+			ResourceManagerRuntimeServices resourceManagerRuntimeServices) throws Exception {
 
 		final MesosTaskManagerParameters taskManagerParameters = MesosUtils.createTmParameters(configuration, LOG);
 		final ContainerSpecification taskManagerContainerSpec = MesosUtils.createContainerSpec(configuration);
@@ -87,9 +83,9 @@ public class MesosResourceManagerFactory extends ActiveResourceManagerFactory<Re
 			resourceId,
 			highAvailabilityServices,
 			heartbeatServices,
-			rmRuntimeServices.getSlotManager(),
+			resourceManagerRuntimeServices.getSlotManager(),
 			ResourceManagerPartitionTrackerImpl::new,
-			rmRuntimeServices.getJobLeaderIdService(),
+			resourceManagerRuntimeServices.getJobLeaderIdService(),
 			clusterInformation,
 			fatalErrorHandler,
 			configuration,
@@ -100,4 +96,10 @@ public class MesosResourceManagerFactory extends ActiveResourceManagerFactory<Re
 			webInterfaceUrl,
 			resourceManagerMetricGroup);
 	}
+
+	@Override
+	protected ResourceManagerRuntimeServicesConfiguration createResourceManagerRuntimeServicesConfiguration(
+		Configuration configuration) throws ConfigurationException {
+		return ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration, MesosWorkerResourceSpecFactory.INSTANCE);
+	}
 }

flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/StandaloneSessionClusterEntrypoint.java

@@ -37,7 +37,7 @@ public class StandaloneSessionClusterEntrypoint extends SessionClusterEntrypoint
 
 	@Override
 	protected DefaultDispatcherResourceManagerComponentFactory createDispatcherResourceManagerComponentFactory(Configuration configuration) {
-		return DefaultDispatcherResourceManagerComponentFactory.createSessionComponentFactory(StandaloneResourceManagerFactory.INSTANCE);
+		return DefaultDispatcherResourceManagerComponentFactory.createSessionComponentFactory(StandaloneResourceManagerFactory.getInstance());
 	}
 
 	public static void main(String[] args) {

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

@@ -417,7 +417,7 @@ public class MiniCluster implements JobExecutorService, AutoCloseableAsync {
 
 	@Nonnull
 	DispatcherResourceManagerComponentFactory createDispatcherResourceManagerComponentFactory() {
-		return DefaultDispatcherResourceManagerComponentFactory.createSessionComponentFactory(StandaloneResourceManagerFactory.INSTANCE);
+		return DefaultDispatcherResourceManagerComponentFactory.createSessionComponentFactory(StandaloneResourceManagerFactory.getInstance());
 	}
 
 	@VisibleForTesting

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ActiveResourceManagerFactory.java

@@ -41,7 +41,7 @@ import javax.annotation.Nullable;
  *
  * @param <T> type of the {@link ResourceIDRetrievable}
  */
-public abstract class ActiveResourceManagerFactory<T extends ResourceIDRetrievable> implements ResourceManagerFactory<T> {
+public abstract class ActiveResourceManagerFactory<T extends ResourceIDRetrievable> extends ResourceManagerFactory<T> {
 
 	@Override
 	public ResourceManager<T> createResourceManager(
@@ -54,7 +54,7 @@ public abstract class ActiveResourceManagerFactory<T extends ResourceIDRetrievab
 			ClusterInformation clusterInformation,
 			@Nullable String webInterfaceUrl,
 			ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception {
-		return createActiveResourceManager(
+		return super.createResourceManager(
 			createActiveResourceManagerConfiguration(configuration),
 			resourceId,
 			rpcService,
@@ -66,19 +66,8 @@ public abstract class ActiveResourceManagerFactory<T extends ResourceIDRetrievab
 			resourceManagerMetricGroup);
 	}
 
-	private static Configuration createActiveResourceManagerConfiguration(Configuration originalConfiguration) {
+	private Configuration createActiveResourceManagerConfiguration(Configuration originalConfiguration) {
 		return TaskExecutorProcessUtils.getConfigurationMapLegacyTaskManagerHeapSizeToConfigOption(
 			originalConfiguration, TaskManagerOptions.TOTAL_PROCESS_MEMORY);
 	}
-
-	protected abstract ResourceManager<T> createActiveResourceManager(
-		Configuration configuration,
-		ResourceID resourceId,
-		RpcService rpcService,
-		HighAvailabilityServices highAvailabilityServices,
-		HeartbeatServices heartbeatServices,
-		FatalErrorHandler fatalErrorHandler,
-		ClusterInformation clusterInformation,
-		@Nullable String webInterfaceUrl,
-		ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception;
 }

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java

@@ -27,6 +27,7 @@ import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.metrics.groups.ResourceManagerMetricGroup;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.ConfigurationException;
 
 import javax.annotation.Nullable;
 
@@ -35,16 +36,57 @@ import javax.annotation.Nullable;
  *
  * @param <T> type of the workers of the ResourceManager
  */
-public interface ResourceManagerFactory<T extends ResourceIDRetrievable> {
-
-	ResourceManager<T> createResourceManager(
-		Configuration configuration,
-		ResourceID resourceId,
-		RpcService rpcService,
-		HighAvailabilityServices highAvailabilityServices,
-		HeartbeatServices heartbeatServices,
-		FatalErrorHandler fatalErrorHandler,
-		ClusterInformation clusterInformation,
-		@Nullable String webInterfaceUrl,
-		ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception;
+public abstract class ResourceManagerFactory<T extends ResourceIDRetrievable> {
+
+	public ResourceManager<T> createResourceManager(
+			Configuration configuration,
+			ResourceID resourceId,
+			RpcService rpcService,
+			HighAvailabilityServices highAvailabilityServices,
+			HeartbeatServices heartbeatServices,
+			FatalErrorHandler fatalErrorHandler,
+			ClusterInformation clusterInformation,
+			@Nullable String webInterfaceUrl,
+			ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception {
+
+		final ResourceManagerRuntimeServices resourceManagerRuntimeServices = createResourceManagerRuntimeServices(
+			configuration, rpcService, highAvailabilityServices);
+
+		return createResourceManager(
+			configuration,
+			resourceId,
+			rpcService,
+			highAvailabilityServices,
+			heartbeatServices,
+			fatalErrorHandler,
+			clusterInformation,
+			webInterfaceUrl,
+			resourceManagerMetricGroup,
+			resourceManagerRuntimeServices);
+	}
+
+	protected abstract ResourceManager<T> createResourceManager(
+			Configuration configuration,
+			ResourceID resourceId,
+			RpcService rpcService,
+			HighAvailabilityServices highAvailabilityServices,
+			HeartbeatServices heartbeatServices,
+			FatalErrorHandler fatalErrorHandler,
+			ClusterInformation clusterInformation,
+			@Nullable String webInterfaceUrl,
+			ResourceManagerMetricGroup resourceManagerMetricGroup,
+			ResourceManagerRuntimeServices resourceManagerRuntimeServices) throws Exception;
+
+	private ResourceManagerRuntimeServices createResourceManagerRuntimeServices(
+			Configuration configuration,
+			RpcService rpcService,
+			HighAvailabilityServices highAvailabilityServices) throws ConfigurationException {
+		return ResourceManagerRuntimeServices.fromConfiguration(
+			createResourceManagerRuntimeServicesConfiguration(configuration),
+			highAvailabilityServices,
+			rpcService.getScheduledExecutor());
+	}
+
+	protected abstract ResourceManagerRuntimeServicesConfiguration createResourceManagerRuntimeServicesConfiguration(
+			Configuration configuration) throws ConfigurationException;
 }

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactory.java

@@ -30,32 +30,35 @@ import org.apache.flink.runtime.io.network.partition.ResourceManagerPartitionTra
 import org.apache.flink.runtime.metrics.groups.ResourceManagerMetricGroup;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.ConfigurationException;
 
 import javax.annotation.Nullable;
 
 /**
  * {@link ResourceManagerFactory} which creates a {@link StandaloneResourceManager}.
  */
-public enum StandaloneResourceManagerFactory implements ResourceManagerFactory<ResourceID> {
-	INSTANCE;
+public final class StandaloneResourceManagerFactory extends ResourceManagerFactory<ResourceID> {
+
+	private static final StandaloneResourceManagerFactory INSTANCE = new StandaloneResourceManagerFactory();
+
+	private StandaloneResourceManagerFactory() {}
+
+	public static StandaloneResourceManagerFactory getInstance() {
+		return INSTANCE;
+	}
 
 	@Override
-	public ResourceManager<ResourceID> createResourceManager(
-			Configuration configuration,
-			ResourceID resourceId,
-			RpcService rpcService,
-			HighAvailabilityServices highAvailabilityServices,
-			HeartbeatServices heartbeatServices,
-			FatalErrorHandler fatalErrorHandler,
-			ClusterInformation clusterInformation,
-			@Nullable String webInterfaceUrl,
-			ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception {
-		final ResourceManagerRuntimeServicesConfiguration resourceManagerRuntimeServicesConfiguration =
-			ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration, ArbitraryWorkerResourceSpecFactory.INSTANCE);
-		final ResourceManagerRuntimeServices resourceManagerRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(
-			resourceManagerRuntimeServicesConfiguration,
-			highAvailabilityServices,
-			rpcService.getScheduledExecutor());
+	protected ResourceManager<ResourceID> createResourceManager(
+		Configuration configuration,
+		ResourceID resourceId,
+		RpcService rpcService,
+		HighAvailabilityServices highAvailabilityServices,
+		HeartbeatServices heartbeatServices,
+		FatalErrorHandler fatalErrorHandler,
+		ClusterInformation clusterInformation,
+		@Nullable String webInterfaceUrl,
+		ResourceManagerMetricGroup resourceManagerMetricGroup,
+		ResourceManagerRuntimeServices resourceManagerRuntimeServices) {
 
 		final Time standaloneClusterStartupPeriodTime = ConfigurationUtils.getStandaloneClusterStartupPeriodTime(configuration);
 
@@ -73,4 +76,10 @@ public enum StandaloneResourceManagerFactory implements ResourceManagerFactory<R
 			standaloneClusterStartupPeriodTime,
 			AkkaUtils.getTimeoutAsTime(configuration));
 	}
+
+	@Override
+	protected ResourceManagerRuntimeServicesConfiguration createResourceManagerRuntimeServicesConfiguration(
+			Configuration configuration) throws ConfigurationException {
+		return ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration, ArbitraryWorkerResourceSpecFactory.INSTANCE);
+	}
 }

flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java

@@ -128,7 +128,7 @@ public class ProcessFailureCancelingITCase extends TestLogger {
 		config.setInteger(JobManagerOptions.PORT, jobManagerPort);
 
 		final DispatcherResourceManagerComponentFactory resourceManagerComponentFactory = DefaultDispatcherResourceManagerComponentFactory.createSessionComponentFactory(
-			StandaloneResourceManagerFactory.INSTANCE);
+			StandaloneResourceManagerFactory.getInstance());
 		DispatcherResourceManagerComponent dispatcherResourceManagerComponent = null;
 
 		final ScheduledExecutorService ioExecutor = TestingUtils.defaultExecutor();

flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnResourceManagerFactory.java

@@ -32,6 +32,7 @@ import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServices;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServicesConfiguration;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.ConfigurationException;
 import org.apache.flink.yarn.YarnResourceManager;
 import org.apache.flink.yarn.YarnWorkerNode;
 
@@ -51,7 +52,7 @@ public class YarnResourceManagerFactory extends ActiveResourceManagerFactory<Yar
 	}
 
 	@Override
-	public ResourceManager<YarnWorkerNode> createActiveResourceManager(
+	public ResourceManager<YarnWorkerNode> createResourceManager(
 			Configuration configuration,
 			ResourceID resourceId,
 			RpcService rpcService,
@@ -60,13 +61,8 @@ public class YarnResourceManagerFactory extends ActiveResourceManagerFactory<Yar
 			FatalErrorHandler fatalErrorHandler,
 			ClusterInformation clusterInformation,
 			@Nullable String webInterfaceUrl,
-			ResourceManagerMetricGroup resourceManagerMetricGroup) throws Exception {
-		final ResourceManagerRuntimeServicesConfiguration rmServicesConfiguration =
-			ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration, YarnWorkerResourceSpecFactory.INSTANCE);
-		final ResourceManagerRuntimeServices rmRuntimeServices = ResourceManagerRuntimeServices.fromConfiguration(
-			rmServicesConfiguration,
-			highAvailabilityServices,
-			rpcService.getScheduledExecutor());
+			ResourceManagerMetricGroup resourceManagerMetricGroup,
+			ResourceManagerRuntimeServices resourceManagerRuntimeServices) {
 
 		return new YarnResourceManager(
 			rpcService,
@@ -75,12 +71,18 @@ public class YarnResourceManagerFactory extends ActiveResourceManagerFactory<Yar
 			System.getenv(),
 			highAvailabilityServices,
 			heartbeatServices,
-			rmRuntimeServices.getSlotManager(),
+			resourceManagerRuntimeServices.getSlotManager(),
 			ResourceManagerPartitionTrackerImpl::new,
-			rmRuntimeServices.getJobLeaderIdService(),
+			resourceManagerRuntimeServices.getJobLeaderIdService(),
 			clusterInformation,
 			fatalErrorHandler,
 			webInterfaceUrl,
 			resourceManagerMetricGroup);
 	}
+
+	@Override
+	protected ResourceManagerRuntimeServicesConfiguration createResourceManagerRuntimeServicesConfiguration(
+		Configuration configuration) throws ConfigurationException {
+		return ResourceManagerRuntimeServicesConfiguration.fromConfiguration(configuration, YarnWorkerResourceSpecFactory.INSTANCE);
+	}
 }