[FLINK-10411] Make ClusterEntrypoint more compositional #6743

tillrohrmann · 2018-09-24T14:53:54Z

What is the purpose of the change

Introduce the ClusterComponent which encapsulates the logic for starting the cluster
components, Dispatcher, RestServerEndpoint and the ResourceManager. The individual
components are created by using a factory instance. The ClusterEntrypoint is now
only responsible for managing the required services needed by the ClusterComponent.
This design should make the testing of these components easier, improve reusability
and reduce code duplication.

Move the logic of when to exit the JVM process out of the ClusterEntrypoint
so that the caller is now responsible to make this call. This improves the
usage of the ClusterEntrypoint for testing purposes.

Brief change log

Introduce ClusterComponent responsible for managing the cluster components: Dispatcher, RestServerEndpoint and ResourceManager
Introduce factories for the cluster components
Update existing ClusterEntrypoints to use the new ClusterComponent

Verifying this change

Most changes are covered by existing tests since it is mainly restructuring code
Added ClassPathJobGraphRetrieverTest to test ClassPathJobGraphRetriever

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: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes)
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)

StefanRRichter

I think the refactoring is good! Being new to this code, I was wondering if we could not simplify ClusterEntrypoint even more be separating the creation of collaborator object even more from the construction of the entry point. I think that is should be possible to avoid lazy initialization and to get the main fields final. We should also be able to avoid the need for locking in entry point and cluster component. Probably that could be improved in a separate PR if you think it is a good improvement.

StefanRRichter · 2018-09-25T09:39:56Z

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

@@ -163,9 +135,6 @@
 	@GuardedBy("lock")
 	private ClusterInformation clusterInformation;


I think this field could become just a local variable.

Yes, we don't need it at all.

StefanRRichter · 2018-09-25T09:47:23Z

flink-runtime/src/main/java/org/apache/flink/runtime/rest/JobRestEndpointFactory.java

+	INSTANCE;
+
+	@Override
+	public WebMonitorEndpoint<RestfulGateway> createRestEndpoint(Configuration configuration, LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever, LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever, TransientBlobService transientBlobService, Executor executor, MetricQueryServiceRetriever metricQueryServiceRetriever, LeaderElectionService leaderElectionService, FatalErrorHandler fatalErrorHandler) throws Exception {


Add some newlines to break down the length of parameter list.

StefanRRichter · 2018-09-25T09:51:23Z

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

+				schedulerConfiguration,
+				taskManagerParameters,
+				taskManagerContainerSpec),
+			new FileJobGraphRetriever(configuration.getString(JOB_GRAPH_FILE_PATH, "job.graph")));


We could use a helper method that does construct FileJobGraphRetriever from configuration because this is duplicated. At least "job.graph" could become a string constant somewhere.

Yes, I'll move it into the FileJobGraphRetriever.

StefanRRichter · 2018-09-25T09:53:16Z

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

+	@GuardedBy("lock")
+	private JobManagerMetricGroup jobManagerMetricGroup;
+
+	public ClusterComponent(DispatcherFactory<T> dispatcherFactory, ResourceManagerFactory<?> resourceManagerFactory, RestEndpointFactory<?> restEndpointFactory) {


linebreaks.

StefanRRichter · 2018-09-25T11:36:36Z

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

+	INSTANCE;
+
+	@Override
+	public ResourceManager<YarnWorkerNode> createResourceManager(Configuration configuration, ResourceID resourceId, RpcService rpcService, HighAvailabilityServices highAvailabilityServices, HeartbeatServices heartbeatServices, MetricRegistry metricRegistry, FatalErrorHandler fatalErrorHandler, ClusterInformation clusterInformation, @Nullable String webInterfaceUrl) throws Exception {


StefanRRichter · 2018-09-25T12:03:00Z

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

+	@Nonnull
+	private final String[] programArguments;
+
+	public ClassPathJobGraphRetriever(@Nonnull String jobClassName, @Nonnull SavepointRestoreSettings savepointRestoreSettings, @Nonnull String[] programArguments) {


StefanRRichter · 2018-09-25T13:32:58Z

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

+	}
+
+	@Override
+	public MiniDispatcher createDispatcher(Configuration configuration, RpcService rpcService, HighAvailabilityServices highAvailabilityServices, ResourceManagerGateway resourceManagerGateway, BlobServer blobServer, HeartbeatServices heartbeatServices, JobManagerMetricGroup jobManagerMetricGroup, @Nullable String metricQueryServicePath, ArchivedExecutionGraphStore archivedExecutionGraphStore, FatalErrorHandler fatalErrorHandler, @Nullable String restAddress, HistoryServerArchivist historyServerArchivist) throws Exception {


linebreaks.

StefanRRichter · 2018-09-25T13:35:05Z

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

+ * Component which starts a {@link Dispatcher}, {@link ResourceManager} and {@link WebMonitorEndpoint}
+ * in the same process.
+ */
+public class ClusterComponent<T extends Dispatcher> implements AutoCloseableAsync {


As somebody looking into this code for the fist time, I found that the name ClusterComponent for this class was not really helpful to understand what it does or what it is good for. Maybe there is a name that makes this a it clearer?

True, I'm also not super happy with the name. Before, I named it DispatcherComponent but this is also not perfect. Maybe it should be named DispatcherResourceManagerComponent/DispatcherResourceManagerRpcEndpoints?

Maybe ClusterCoreCollaborators? It also does not give info about what it contains, but gives a hint that it is holding references to core collaborators and you can dig into the class for more details?

Or a different view, if we cannot even find a name for it, then maybe it should not exists in the current form? (given that: one class = one concept/responsibility)

StefanRRichter · 2018-09-25T13:44:24Z

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

+		try {
+			clusterEntrypoint.startCluster();
+		} catch (ClusterEntrypointException e) {
+			LOG.error(String.format("Could not start cluster entrypoint %s.", clusterEntrypointName, e));


LOG.error(String.format("Could not start cluster entrypoint %s.", clusterEntrypointName), e);

Good catch!

StefanRRichter · 2018-09-25T13:50:41Z

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

 		return terminationFuture;
 	}

-	protected void startCluster() {
+	protected void startCluster() throws ClusterEntrypointException {


Could currently be private

Yes, you're right. I actually use it in a follow-up PR for testing purposes. Would, thus, just keep it like this.

StefanRRichter

Had some minor fixes and suggestions, but overall looking good!

tillrohrmann · 2018-09-27T13:16:04Z

Thanks a lot for your review @StefanRRichter. I agree that instantiating the ClusterComponent in an eager way is a good improvement. I'll add another commit for that.

Introduce a ClusterComponent which encapsulates the logic for starting the cluster components, Dispatcher, RestServerEndpoint and the ResourceManager. The individual components are created by using a factory instance. The ClusterEntrypoint is now only responsible for managing the required services needed by the ClusterComponent. This design should make the testing of these components easier, improve reusability and reduce code duplication.

Move the logic of when to exit the JVM process out of the ClusterEntrypoint so that the caller is now responsible to make this call. This improves the usage of the ClusterEntrypoint for testing purposes.

…omponent

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.

tillrohrmann mentioned this pull request Sep 24, 2018

[FLINK-10401] Port ProcessFailureCancelingITCase to new code base #6749

Closed

StefanRRichter reviewed Sep 25, 2018

View reviewed changes

StefanRRichter approved these changes Sep 25, 2018

View reviewed changes

tillrohrmann force-pushed the refactorClusterEntrypoint branch from 06b1e2c to 8bb823c Compare September 27, 2018 13:19

tillrohrmann added 4 commits September 27, 2018 16:19

[FLINK-10411] Move System.exit out of ClusterEntrypoint

9166ee9

Move the logic of when to exit the JVM process out of the ClusterEntrypoint so that the caller is now responsible to make this call. This improves the usage of the ClusterEntrypoint for testing purposes.

[FLINK-10411] Rename ClusterComponent into DispatcherResourceManagerC…

430ae07

…omponent

tillrohrmann force-pushed the refactorClusterEntrypoint branch from 8bb823c to 8d38759 Compare September 27, 2018 14:19

asfgit closed this in 9f5fd07 Sep 28, 2018

tillrohrmann deleted the refactorClusterEntrypoint branch September 28, 2018 09:46

yanghua mentioned this pull request Sep 28, 2018

[FLINK-10291] Generate JobGraph with fixed/configurable JobID in StandaloneJobClusterEntrypoint #6733

Closed

rmetzger added the component=Runtime/Coordination label Mar 18, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-10411] Make ClusterEntrypoint more compositional #6743

[FLINK-10411] Make ClusterEntrypoint more compositional #6743

tillrohrmann commented Sep 24, 2018

StefanRRichter left a comment •

edited

StefanRRichter Sep 25, 2018 •

edited

tillrohrmann Sep 27, 2018

StefanRRichter Sep 25, 2018

StefanRRichter Sep 25, 2018

tillrohrmann Sep 27, 2018

StefanRRichter Sep 25, 2018

StefanRRichter Sep 25, 2018

StefanRRichter Sep 25, 2018

StefanRRichter Sep 25, 2018

StefanRRichter Sep 25, 2018

tillrohrmann Sep 27, 2018

StefanRRichter Sep 27, 2018

StefanRRichter Sep 27, 2018 •

edited

StefanRRichter Sep 25, 2018

tillrohrmann Sep 27, 2018

StefanRRichter Sep 25, 2018

tillrohrmann Sep 27, 2018

StefanRRichter left a comment •

edited

tillrohrmann commented Sep 27, 2018

		@@ -163,9 +135,6 @@
		@GuardedBy("lock")
		private ClusterInformation clusterInformation;

[FLINK-10411] Make ClusterEntrypoint more compositional #6743

[FLINK-10411] Make ClusterEntrypoint more compositional #6743

Conversation

tillrohrmann commented Sep 24, 2018

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

StefanRRichter left a comment • edited

Choose a reason for hiding this comment

StefanRRichter Sep 25, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

StefanRRichter Sep 27, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

StefanRRichter left a comment • edited

Choose a reason for hiding this comment

tillrohrmann commented Sep 27, 2018

StefanRRichter left a comment •

edited

StefanRRichter Sep 25, 2018 •

edited

StefanRRichter Sep 27, 2018 •

edited

StefanRRichter left a comment •

edited