[FLINK-10411] Introduce DispatcherResourceManagerComponentFactory

This commit introduces the DispatcherResourceManagerComponentFactory which is used
to create a DispatcherResourceManagerComponent. That way, it is possible to eagerly
initialize all fields of the DispatcherResourceManagerComponent making it possible
to make all fields final and remove the lock.

This closes apache#6743.

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

@@ -23,7 +23,7 @@
 import org.apache.flink.client.program.ProgramInvocationException;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
-import org.apache.flink.runtime.entrypoint.JobGraphRetriever;
+import org.apache.flink.runtime.entrypoint.component.JobGraphRetriever;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.util.FlinkException;

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

@@ -19,11 +19,11 @@
 package org.apache.flink.container.entrypoint;
 
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.entrypoint.ClusterComponent;
 import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
 import org.apache.flink.runtime.entrypoint.FlinkParseException;
-import org.apache.flink.runtime.entrypoint.JobClusterComponent;
 import org.apache.flink.runtime.entrypoint.JobClusterEntrypoint;
+import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory;
+import org.apache.flink.runtime.entrypoint.component.JobDispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory;
@@ -61,8 +61,8 @@ public final class StandaloneJobClusterEntryPoint extends JobClusterEntrypoint {
  }
 
  @Override
- protected ClusterComponent<?> createClusterComponent(Configuration configuration) {
- return new JobClusterComponent(
+ protected DispatcherResourceManagerComponentFactory<?> createDispatcherResourceManagerComponentFactory(Configuration configuration) {
+ return new JobDispatcherResourceManagerComponentFactory(
  StandaloneResourceManagerFactory.INSTANCE,
  new ClassPathJobGraphRetriever(jobClassName, savepointRestoreSettings, programArguments));
  }

flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java

@@ -28,11 +28,11 @@
 import org.apache.flink.runtime.clusterframework.BootstrapTools;
 import org.apache.flink.runtime.clusterframework.ContainerSpecification;
 import org.apache.flink.runtime.concurrent.FutureUtils;
-import org.apache.flink.runtime.entrypoint.ClusterComponent;
 import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
-import org.apache.flink.runtime.entrypoint.FileJobGraphRetriever;
-import org.apache.flink.runtime.entrypoint.JobClusterComponent;
 import org.apache.flink.runtime.entrypoint.JobClusterEntrypoint;
+import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory;
+import org.apache.flink.runtime.entrypoint.component.FileJobGraphRetriever;
+import org.apache.flink.runtime.entrypoint.component.JobDispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.util.EnvironmentInformation;
 import org.apache.flink.runtime.util.JvmShutdownSafeguard;
 import org.apache.flink.runtime.util.SignalHandler;
@@ -109,8 +109,8 @@ protected CompletableFuture<Void> stopClusterServices(boolean cleanupHaData) {
  }
 
  @Override
- protected ClusterComponent<?> createClusterComponent(Configuration configuration) {
- return new JobClusterComponent(
+ protected DispatcherResourceManagerComponentFactory<?> createDispatcherResourceManagerComponentFactory(Configuration configuration) {
+ return new JobDispatcherResourceManagerComponentFactory(
  new MesosResourceManagerFactory(
  mesosServices,
  schedulerConfiguration,

flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java

@@ -28,10 +28,10 @@
 import org.apache.flink.runtime.clusterframework.BootstrapTools;
 import org.apache.flink.runtime.clusterframework.ContainerSpecification;
 import org.apache.flink.runtime.concurrent.FutureUtils;
-import org.apache.flink.runtime.entrypoint.ClusterComponent;
 import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
-import org.apache.flink.runtime.entrypoint.SessionClusterComponent;
 import org.apache.flink.runtime.entrypoint.SessionClusterEntrypoint;
+import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory;
+import org.apache.flink.runtime.entrypoint.component.SessionDispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.util.EnvironmentInformation;
 import org.apache.flink.runtime.util.JvmShutdownSafeguard;
 import org.apache.flink.runtime.util.SignalHandler;
@@ -108,12 +108,13 @@ protected CompletableFuture<Void> stopClusterServices(boolean cleanupHaData) {
  }
 
  @Override
- protected ClusterComponent<?> createClusterComponent(Configuration configuration) {
- return new SessionClusterComponent(new MesosResourceManagerFactory(
- mesosServices,
- mesosConfig,
- taskManagerParameters,
- taskManagerContainerSpec));
+ protected DispatcherResourceManagerComponentFactory<?> createDispatcherResourceManagerComponentFactory(Configuration configuration) {
+ return new SessionDispatcherResourceManagerComponentFactory(
+ new MesosResourceManagerFactory(
+ mesosServices,
+ mesosConfig,
+ taskManagerParameters,
+ taskManagerContainerSpec));
  }
 
  public static void main(String[] args) {

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/JobDispatcherFactory.java

@@ -21,7 +21,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.blob.BlobServer;
 import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
-import org.apache.flink.runtime.entrypoint.JobGraphRetriever;
+import org.apache.flink.runtime.entrypoint.component.JobGraphRetriever;
 import org.apache.flink.runtime.heartbeat.HeartbeatServices;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.jobgraph.JobGraph;

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

@@ -38,6 +38,8 @@
 import org.apache.flink.runtime.concurrent.ScheduledExecutor;
 import org.apache.flink.runtime.dispatcher.ArchivedExecutionGraphStore;
 import org.apache.flink.runtime.dispatcher.MiniDispatcher;
+import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent;
+import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
 import org.apache.flink.runtime.heartbeat.HeartbeatServices;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
@@ -109,7 +111,7 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
  private final AtomicBoolean isShutDown = new AtomicBoolean(false);
 
  @GuardedBy("lock")
- private ClusterComponent<?> clusterComponent;
+ private DispatcherResourceManagerComponent<?> clusterComponent;
 
  @GuardedBy("lock")
  private MetricRegistryImpl metricRegistry;
@@ -204,9 +206,9 @@ private void runCluster(Configuration configuration) throws Exception {
  configuration.setString(JobManagerOptions.ADDRESS, commonRpcService.getAddress());
  configuration.setInteger(JobManagerOptions.PORT, commonRpcService.getPort());
 
- clusterComponent = createClusterComponent(configuration);
+ final DispatcherResourceManagerComponentFactory<?> dispatcherResourceManagerComponentFactory = createDispatcherResourceManagerComponentFactory(configuration);
 
- clusterComponent.startComponent(
+ clusterComponent = dispatcherResourceManagerComponentFactory.create(
  configuration,
  commonRpcService,
  haServices,
@@ -460,7 +462,7 @@ private void cleanupDirectories() throws IOException {
  // Abstract methods
  // --------------------------------------------------
 
- protected abstract ClusterComponent<?> createClusterComponent(Configuration configuration);
+ protected abstract DispatcherResourceManagerComponentFactory<?> createDispatcherResourceManagerComponentFactory(Configuration configuration);
 
  protected abstract ArchivedExecutionGraphStore createSerializableExecutionGraphStore(
  Configuration configuration,

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

@@ -19,6 +19,8 @@
 package org.apache.flink.runtime.entrypoint;
 
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory;
+import org.apache.flink.runtime.entrypoint.component.SessionDispatcherResourceManagerComponentFactory;
 import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
 import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory;
 import org.apache.flink.runtime.util.EnvironmentInformation;
@@ -35,8 +37,8 @@ public StandaloneSessionClusterEntrypoint(Configuration configuration) {
  }
 
  @Override
- protected ClusterComponent<?> createClusterComponent(Configuration configuration) {
- return new SessionClusterComponent(StandaloneResourceManagerFactory.INSTANCE);
+ protected DispatcherResourceManagerComponentFactory<?> createDispatcherResourceManagerComponentFactory(Configuration configuration) {
+ return new SessionDispatcherResourceManagerComponentFactory(StandaloneResourceManagerFactory.INSTANCE);
  }
 
  public static void main(String[] args) {