You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/02/15 10:42:54 UTC
[11/22] flink git commit: [FLINK-8608] [flip6] Implement
MiniDispatcher for job mode
[FLINK-8608] [flip6] Implement MiniDispatcher for job mode
The MiniDispatcher is responsible for submitting the single job with which
a job mode cluster is started. Once the job has completed and if the cluster
has been started in detached mode, the MiniDispatcher will terminate.
In order to reduce code duplication, the MiniDispatcher is a sub class of the
Dispatcher which is started with a single job submitted job graph store.
This closes #5431.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/84418d0b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/84418d0b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/84418d0b
Branch: refs/heads/master
Commit: 84418d0b9183fbfeb3919fcb0d2f17708f786628
Parents: 2a15417
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Feb 14 15:44:06 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Feb 15 08:44:19 2018 +0100
----------------------------------------------------------------------
.../client/deployment/ClusterDescriptor.java | 5 +-
.../Flip6StandaloneClusterDescriptor.java | 5 +-
.../deployment/StandaloneClusterDescriptor.java | 5 +-
.../client/cli/util/DummyClusterDescriptor.java | 5 +-
.../entrypoint/MesosJobClusterEntrypoint.java | 16 +-
.../MesosSessionClusterEntrypoint.java | 16 +-
.../flink/runtime/dispatcher/Dispatcher.java | 112 ++++--
.../MemoryArchivedExecutionGraphStore.java | 92 +++++
.../runtime/dispatcher/MiniDispatcher.java | 120 ++++++
.../SingleJobSubmittedJobGraphStore.java | 77 ++++
.../dispatcher/StandaloneDispatcher.java | 36 +-
.../runtime/entrypoint/ClusterEntrypoint.java | 383 +++++++++++++++++--
.../entrypoint/JobClusterEntrypoint.java | 301 ++-------------
.../entrypoint/SessionClusterEntrypoint.java | 202 +---------
.../jobmaster/JobMasterRestEndpoint.java | 62 ---
.../jobmaster/MiniDispatcherRestEndpoint.java | 64 ++++
.../rest/handler/job/BlobServerPortHandler.java | 2 +-
.../rest/handler/job/JobSubmitHandler.java | 2 +-
.../legacy/files/StaticFileServerHandler.java | 2 +-
.../TaskExecutorRegistrationSuccess.java | 2 +-
.../taskexecutor/slot/TaskSlotTable.java | 2 +-
.../runtime/webmonitor/WebMonitorEndpoint.java | 6 +-
.../runtime/webmonitor/WebMonitorUtils.java | 2 +-
.../runtime/dispatcher/DispatcherTest.java | 54 +--
.../MemoryArchivedExecutionGraphStore.java | 92 -----
.../runtime/dispatcher/MiniDispatcherTest.java | 211 ++++++++++
.../yarn/TestingYarnClusterDescriptor.java | 5 +-
.../java/org/apache/flink/yarn/YARNITCase.java | 5 +-
.../yarn/AbstractYarnClusterDescriptor.java | 51 +--
.../flink/yarn/Flip6YarnClusterDescriptor.java | 24 ++
.../flink/yarn/YarnClusterDescriptor.java | 5 +-
.../flink/yarn/AbstractYarnClusterTest.java | 8 +
32 files changed, 1173 insertions(+), 801 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java
index aadb5fe..f9f5d4b 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterDescriptor.java
@@ -56,12 +56,15 @@ public interface ClusterDescriptor<T> extends AutoCloseable {
*
* @param clusterSpecification Initial cluster specification with which the Flink cluster is launched
* @param jobGraph JobGraph with which the job cluster is started
+ * @param detached true if the cluster should be stopped after the job completion without serving the result,
+ * otherwise false
* @return Cluster client to talk to the Flink cluster
* @throws ClusterDeploymentException if the cluster could not be deployed
*/
ClusterClient<T> deployJobCluster(
final ClusterSpecification clusterSpecification,
- final JobGraph jobGraph) throws ClusterDeploymentException;
+ final JobGraph jobGraph,
+ final boolean detached) throws ClusterDeploymentException;
/**
* Terminates the cluster with the given cluster id.
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-clients/src/main/java/org/apache/flink/client/deployment/Flip6StandaloneClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/Flip6StandaloneClusterDescriptor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/Flip6StandaloneClusterDescriptor.java
index 8096da8..0a3286a 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/Flip6StandaloneClusterDescriptor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/Flip6StandaloneClusterDescriptor.java
@@ -58,7 +58,10 @@ public class Flip6StandaloneClusterDescriptor implements ClusterDescriptor<Stand
}
@Override
- public RestClusterClient<StandaloneClusterId> deployJobCluster(ClusterSpecification clusterSpecification, JobGraph jobGraph) {
+ public RestClusterClient<StandaloneClusterId> deployJobCluster(
+ ClusterSpecification clusterSpecification,
+ JobGraph jobGraph,
+ boolean detached) {
throw new UnsupportedOperationException("Can't deploy a standalone FLIP-6 per-job cluster.");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java
index 62908fe..83bb136 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/StandaloneClusterDescriptor.java
@@ -57,7 +57,10 @@ public class StandaloneClusterDescriptor implements ClusterDescriptor<Standalone
}
@Override
- public StandaloneClusterClient deployJobCluster(ClusterSpecification clusterSpecification, JobGraph jobGraph) {
+ public StandaloneClusterClient deployJobCluster(
+ ClusterSpecification clusterSpecification,
+ JobGraph jobGraph,
+ boolean detached) {
throw new UnsupportedOperationException("Can't deploy a standalone per-job cluster.");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
index 0ceb95e..df2f3f7 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/util/DummyClusterDescriptor.java
@@ -52,7 +52,10 @@ public class DummyClusterDescriptor<T> implements ClusterDescriptor<T> {
}
@Override
- public ClusterClient<T> deployJobCluster(ClusterSpecification clusterSpecification, JobGraph jobGraph) {
+ public ClusterClient<T> deployJobCluster(
+ ClusterSpecification clusterSpecification,
+ JobGraph jobGraph,
+ boolean detached) {
return clusterClient;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
index 21864dc..b0054bb 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosJobClusterEntrypoint.java
@@ -25,7 +25,6 @@ import org.apache.flink.mesos.runtime.clusterframework.MesosTaskManagerParameter
import org.apache.flink.mesos.runtime.clusterframework.services.MesosServices;
import org.apache.flink.mesos.runtime.clusterframework.services.MesosServicesUtils;
import org.apache.flink.mesos.util.MesosConfiguration;
-import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.ContainerSpecification;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -114,17 +113,6 @@ public class MesosJobClusterEntrypoint extends JobClusterEntrypoint {
}
@Override
- protected void startClusterComponents(
- Configuration configuration,
- RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- BlobServer blobServer,
- HeartbeatServices heartbeatServices,
- MetricRegistry metricRegistry) throws Exception {
- super.startClusterComponents(configuration, rpcService, highAvailabilityServices, blobServer, heartbeatServices, metricRegistry);
- }
-
- @Override
protected ResourceManager<?> createResourceManager(
Configuration configuration,
ResourceID resourceId,
@@ -179,11 +167,11 @@ public class MesosJobClusterEntrypoint extends JobClusterEntrypoint {
}
@Override
- protected void stopClusterComponents(boolean cleanupHaData) throws Exception {
+ protected void stopClusterServices(boolean cleanupHaData) throws FlinkException {
Throwable exception = null;
try {
- super.stopClusterComponents(cleanupHaData);
+ super.stopClusterServices(cleanupHaData);
} catch (Throwable t) {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
index 68cf7c6..306d5b2 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosSessionClusterEntrypoint.java
@@ -25,7 +25,6 @@ import org.apache.flink.mesos.runtime.clusterframework.MesosTaskManagerParameter
import org.apache.flink.mesos.runtime.clusterframework.services.MesosServices;
import org.apache.flink.mesos.runtime.clusterframework.services.MesosServicesUtils;
import org.apache.flink.mesos.util.MesosConfiguration;
-import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.clusterframework.ContainerSpecification;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -104,17 +103,6 @@ public class MesosSessionClusterEntrypoint extends SessionClusterEntrypoint {
}
@Override
- protected void startClusterComponents(
- Configuration configuration,
- RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- BlobServer blobServer,
- HeartbeatServices heartbeatServices,
- MetricRegistry metricRegistry) throws Exception {
- super.startClusterComponents(configuration, rpcService, highAvailabilityServices, blobServer, heartbeatServices, metricRegistry);
- }
-
- @Override
protected ResourceManager<?> createResourceManager(
Configuration configuration,
ResourceID resourceId,
@@ -153,11 +141,11 @@ public class MesosSessionClusterEntrypoint extends SessionClusterEntrypoint {
}
@Override
- protected void stopClusterComponents(boolean cleanupHaData) throws Exception {
+ protected void stopClusterServices(boolean cleanupHaData) throws FlinkException {
Throwable exception = null;
try {
- super.stopClusterComponents(cleanupHaData);
+ super.stopClusterServices(cleanupHaData);
} catch (Throwable t) {
exception = t;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index c83dce0..92ccc9b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -102,19 +102,23 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
private final ArchivedExecutionGraphStore archivedExecutionGraphStore;
+ private final JobManagerRunnerFactory jobManagerRunnerFactory;
+
@Nullable
protected final String restAddress;
- protected Dispatcher(
+ public Dispatcher(
RpcService rpcService,
String endpointId,
Configuration configuration,
HighAvailabilityServices highAvailabilityServices,
+ SubmittedJobGraphStore submittedJobGraphStore,
ResourceManagerGateway resourceManagerGateway,
BlobServer blobServer,
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
ArchivedExecutionGraphStore archivedExecutionGraphStore,
+ JobManagerRunnerFactory jobManagerRunnerFactory,
FatalErrorHandler fatalErrorHandler,
@Nullable String restAddress) throws Exception {
super(rpcService, endpointId);
@@ -126,12 +130,12 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
this.blobServer = Preconditions.checkNotNull(blobServer);
this.metricRegistry = Preconditions.checkNotNull(metricRegistry);
this.fatalErrorHandler = Preconditions.checkNotNull(fatalErrorHandler);
+ this.submittedJobGraphStore = Preconditions.checkNotNull(submittedJobGraphStore);
this.jobManagerSharedServices = JobManagerSharedServices.fromConfiguration(
configuration,
this.blobServer);
- this.submittedJobGraphStore = highAvailabilityServices.getSubmittedJobGraphStore();
this.runningJobsRegistry = highAvailabilityServices.getRunningJobsRegistry();
jobManagerRunners = new HashMap<>(16);
@@ -141,6 +145,8 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
this.restAddress = restAddress;
this.archivedExecutionGraphStore = Preconditions.checkNotNull(archivedExecutionGraphStore);
+
+ this.jobManagerRunnerFactory = Preconditions.checkNotNull(jobManagerRunnerFactory);
}
//------------------------------------------------------
@@ -226,7 +232,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
final JobManagerRunner jobManagerRunner;
try {
- jobManagerRunner = createJobManagerRunner(
+ jobManagerRunner = jobManagerRunnerFactory.createJobManagerRunner(
ResourceID.generate(),
jobGraph,
configuration,
@@ -237,7 +243,8 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
jobManagerSharedServices,
metricRegistry,
new DispatcherOnCompleteActions(jobGraph.getJobID()),
- fatalErrorHandler);
+ fatalErrorHandler,
+ restAddress);
jobManagerRunner.start();
} catch (Exception e) {
@@ -507,16 +514,18 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
});
}
- private void onFatalError(Throwable throwable) {
+ protected void onFatalError(Throwable throwable) {
log.error("Fatal error occurred in dispatcher {}.", getAddress(), throwable);
+
fatalErrorHandler.onFatalError(throwable);
}
- private void jobReachedGloballyTerminalState(ArchivedExecutionGraph archivedExecutionGraph) {
+ protected void jobReachedGloballyTerminalState(ArchivedExecutionGraph archivedExecutionGraph) {
Preconditions.checkArgument(
archivedExecutionGraph.getState().isGloballyTerminalState(),
- "Job " + archivedExecutionGraph.getJobID() + " is in state " +
- archivedExecutionGraph.getState() + " which is not globally terminal.");
+ "Job %s is in state %s which is not globally terminal.",
+ archivedExecutionGraph.getJobID(),
+ archivedExecutionGraph.getState());
try {
archivedExecutionGraphStore.put(archivedExecutionGraph);
@@ -537,18 +546,13 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
}
}
- protected abstract JobManagerRunner createJobManagerRunner(
- ResourceID resourceId,
- JobGraph jobGraph,
- Configuration configuration,
- RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- HeartbeatServices heartbeatServices,
- BlobServer blobServer,
- JobManagerSharedServices jobManagerSharedServices,
- MetricRegistry metricRegistry,
- OnCompletionActions onCompleteActions,
- FatalErrorHandler fatalErrorHandler) throws Exception;
+ protected void jobFinishedByOther(JobID jobId) {
+ try {
+ removeJob(jobId, false);
+ } catch (Exception e) {
+ log.warn("Could not properly remove job {} from the dispatcher.", jobId, e);
+ }
+ }
//------------------------------------------------------
// Leader contender
@@ -671,13 +675,67 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
log.info("Job {} was finished by other JobManager.", jobId);
runAsync(
- () -> {
- try {
- removeJob(jobId, false);
- } catch (Exception e) {
- log.warn("Could not properly remove job {} from the dispatcher.", jobId, e);
- }
- });
+ () -> Dispatcher.this.jobFinishedByOther(jobId));
+ }
+ }
+
+ //------------------------------------------------------
+ // Factories
+ //------------------------------------------------------
+
+ /**
+ * Factory for a {@link JobManagerRunner}.
+ */
+ @FunctionalInterface
+ public interface JobManagerRunnerFactory {
+ JobManagerRunner createJobManagerRunner(
+ ResourceID resourceId,
+ JobGraph jobGraph,
+ Configuration configuration,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ BlobServer blobServer,
+ JobManagerSharedServices jobManagerServices,
+ MetricRegistry metricRegistry,
+ OnCompletionActions onCompleteActions,
+ FatalErrorHandler fatalErrorHandler,
+ @Nullable String restAddress) throws Exception;
+ }
+
+ /**
+ * Singleton default factory for {@link JobManagerRunner}.
+ */
+ public enum DefaultJobManagerRunnerFactory implements JobManagerRunnerFactory {
+ INSTANCE;
+
+ @Override
+ public JobManagerRunner createJobManagerRunner(
+ ResourceID resourceId,
+ JobGraph jobGraph,
+ Configuration configuration,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ BlobServer blobServer,
+ JobManagerSharedServices jobManagerServices,
+ MetricRegistry metricRegistry,
+ OnCompletionActions onCompleteActions,
+ FatalErrorHandler fatalErrorHandler,
+ @Nullable String restAddress) throws Exception {
+ return new JobManagerRunner(
+ resourceId,
+ jobGraph,
+ configuration,
+ rpcService,
+ highAvailabilityServices,
+ heartbeatServices,
+ blobServer,
+ jobManagerServices,
+ metricRegistry,
+ onCompleteActions,
+ fatalErrorHandler,
+ restAddress);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryArchivedExecutionGraphStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryArchivedExecutionGraphStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryArchivedExecutionGraphStore.java
new file mode 100644
index 0000000..bcb7df2
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryArchivedExecutionGraphStore.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.jobgraph.JobStatus;
+import org.apache.flink.runtime.messages.webmonitor.JobDetails;
+import org.apache.flink.runtime.messages.webmonitor.JobsOverview;
+import org.apache.flink.runtime.webmonitor.WebMonitorUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * {@link ArchivedExecutionGraphStore} implementation which stores the {@link ArchivedExecutionGraph}
+ * in memory.
+ */
+public class MemoryArchivedExecutionGraphStore implements ArchivedExecutionGraphStore {
+
+ private final Map<JobID, ArchivedExecutionGraph> serializableExecutionGraphs = new HashMap<>(4);
+
+ @Override
+ public int size() {
+ return serializableExecutionGraphs.size();
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionGraph get(JobID jobId) {
+ return serializableExecutionGraphs.get(jobId);
+ }
+
+ @Override
+ public void put(ArchivedExecutionGraph serializableExecutionGraph) throws IOException {
+ serializableExecutionGraphs.put(serializableExecutionGraph.getJobID(), serializableExecutionGraph);
+ }
+
+ @Override
+ public JobsOverview getStoredJobsOverview() {
+ Collection<JobStatus> allJobStatus = serializableExecutionGraphs.values().stream()
+ .map(ArchivedExecutionGraph::getState)
+ .collect(Collectors.toList());
+
+ return JobsOverview.create(allJobStatus);
+ }
+
+ @Override
+ public Collection<JobDetails> getAvailableJobDetails() {
+ return serializableExecutionGraphs.values().stream()
+ .map(WebMonitorUtils::createDetailsForJob)
+ .collect(Collectors.toList());
+ }
+
+ @Nullable
+ @Override
+ public JobDetails getAvailableJobDetails(JobID jobId) {
+ final ArchivedExecutionGraph archivedExecutionGraph = serializableExecutionGraphs.get(jobId);
+
+ if (archivedExecutionGraph != null) {
+ return WebMonitorUtils.createDetailsForJob(archivedExecutionGraph);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ serializableExecutionGraphs.clear();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
new file mode 100644
index 0000000..26d357e
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+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.JobClusterEntrypoint;
+import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.heartbeat.HeartbeatServices;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.rpc.RpcService;
+import org.apache.flink.util.FlinkException;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Mini Dispatcher which is instantiated as the dispatcher component by the {@link JobClusterEntrypoint}.
+ *
+ * <p>The mini dispatcher is initialized with a single {@link JobGraph} which it runs.
+ *
+ * <p>Depending on the {@link ClusterEntrypoint.ExecutionMode}, the mini dispatcher will directly
+ * terminate after job completion if its execution mode is {@link ClusterEntrypoint.ExecutionMode#DETACHED}.
+ */
+public class MiniDispatcher extends Dispatcher {
+
+ private final JobClusterEntrypoint.ExecutionMode executionMode;
+
+ public MiniDispatcher(
+ RpcService rpcService,
+ String endpointId,
+ Configuration configuration,
+ HighAvailabilityServices highAvailabilityServices,
+ ResourceManagerGateway resourceManagerGateway,
+ BlobServer blobServer,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ ArchivedExecutionGraphStore archivedExecutionGraphStore,
+ JobManagerRunnerFactory jobManagerRunnerFactory,
+ FatalErrorHandler fatalErrorHandler,
+ @Nullable String restAddress,
+ JobGraph jobGraph,
+ JobClusterEntrypoint.ExecutionMode executionMode) throws Exception {
+ super(
+ rpcService,
+ endpointId,
+ configuration,
+ highAvailabilityServices,
+ new SingleJobSubmittedJobGraphStore(jobGraph),
+ resourceManagerGateway,
+ blobServer,
+ heartbeatServices,
+ metricRegistry,
+ archivedExecutionGraphStore,
+ jobManagerRunnerFactory,
+ fatalErrorHandler,
+ restAddress);
+
+ this.executionMode = executionMode;
+ }
+
+ @Override
+ public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) {
+ final CompletableFuture<Acknowledge> acknowledgeCompletableFuture = super.submitJob(jobGraph, timeout);
+
+ acknowledgeCompletableFuture.whenComplete(
+ (Acknowledge ignored, Throwable throwable) -> {
+ if (throwable != null) {
+ onFatalError(new FlinkException(
+ "Failed to submit job " + jobGraph.getJobID() + " in job mode.",
+ throwable));
+ }
+ });
+
+ return acknowledgeCompletableFuture;
+ }
+
+ @Override
+ protected void jobReachedGloballyTerminalState(ArchivedExecutionGraph archivedExecutionGraph) {
+ super.jobReachedGloballyTerminalState(archivedExecutionGraph);
+
+ if (executionMode == ClusterEntrypoint.ExecutionMode.DETACHED) {
+ // shut down since we don't have to wait for the execution result retrieval
+ shutDown();
+ }
+ }
+
+ @Override
+ protected void jobFinishedByOther(JobID jobId) {
+ super.jobFinishedByOther(jobId);
+
+ // shut down since we have done our job
+ shutDown();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/SingleJobSubmittedJobGraphStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/SingleJobSubmittedJobGraphStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/SingleJobSubmittedJobGraphStore.java
new file mode 100644
index 0000000..26d3abc
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/SingleJobSubmittedJobGraphStore.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.SubmittedJobGraph;
+import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.Preconditions;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * {@link SubmittedJobGraphStore} implementation for a single job.
+ */
+public class SingleJobSubmittedJobGraphStore implements SubmittedJobGraphStore {
+
+ private final JobGraph jobGraph;
+
+ public SingleJobSubmittedJobGraphStore(JobGraph jobGraph) {
+ this.jobGraph = Preconditions.checkNotNull(jobGraph);
+ }
+
+ @Override
+ public void start(SubmittedJobGraphListener jobGraphListener) throws Exception {
+ // noop
+ }
+
+ @Override
+ public void stop() throws Exception {
+ // noop
+ }
+
+ @Override
+ public SubmittedJobGraph recoverJobGraph(JobID jobId) throws Exception {
+ if (jobGraph.getJobID().equals(jobId)) {
+ return new SubmittedJobGraph(jobGraph, null);
+ } else {
+ throw new FlinkException("Could not recover job graph " + jobId + '.');
+ }
+ }
+
+ @Override
+ public void putJobGraph(SubmittedJobGraph jobGraph) throws Exception {
+ if (!jobGraph.getJobId().equals(jobGraph.getJobId())) {
+ throw new FlinkException("Cannot put additional jobs into this submitted job graph store.");
+ }
+ }
+
+ @Override
+ public void removeJobGraph(JobID jobId) throws Exception {
+ // ignore
+ }
+
+ @Override
+ public Collection<JobID> getJobIds() throws Exception {
+ return Collections.singleton(jobGraph.getJobID());
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
index 6592e98..a7d21f3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
@@ -20,13 +20,9 @@ package org.apache.flink.runtime.dispatcher;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.blob.BlobServer;
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.jobmanager.OnCompletionActions;
-import org.apache.flink.runtime.jobmaster.JobManagerRunner;
-import org.apache.flink.runtime.jobmaster.JobManagerSharedServices;
import org.apache.flink.runtime.jobmaster.JobMaster;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
@@ -51,6 +47,7 @@ public class StandaloneDispatcher extends Dispatcher {
HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
ArchivedExecutionGraphStore archivedExecutionGraphStore,
+ JobManagerRunnerFactory jobManagerRunnerFactory,
FatalErrorHandler fatalErrorHandler,
@Nullable String restAddress) throws Exception {
super(
@@ -58,40 +55,13 @@ public class StandaloneDispatcher extends Dispatcher {
endpointId,
configuration,
highAvailabilityServices,
+ highAvailabilityServices.getSubmittedJobGraphStore(),
resourceManagerGateway,
blobServer,
heartbeatServices,
metricRegistry,
archivedExecutionGraphStore,
- fatalErrorHandler,
- restAddress);
- }
-
- @Override
- protected JobManagerRunner createJobManagerRunner(
- ResourceID resourceId,
- JobGraph jobGraph,
- Configuration configuration,
- RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- HeartbeatServices heartbeatServices,
- BlobServer blobServer,
- JobManagerSharedServices jobManagerSharedServices,
- MetricRegistry metricRegistry,
- OnCompletionActions onCompleteActions,
- FatalErrorHandler fatalErrorHandler) throws Exception {
- // create the standard job manager runner
- return new JobManagerRunner(
- resourceId,
- jobGraph,
- configuration,
- rpcService,
- highAvailabilityServices,
- heartbeatServices,
- blobServer,
- jobManagerSharedServices,
- metricRegistry,
- onCompleteActions,
+ jobManagerRunnerFactory,
fatalErrorHandler,
restAddress);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index dfb1b4c..584fcae 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -20,25 +20,48 @@ package org.apache.flink.runtime.entrypoint;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.WebOptions;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobServer;
+import org.apache.flink.runtime.blob.TransientBlobCache;
+import org.apache.flink.runtime.blob.TransientBlobService;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.concurrent.ScheduledExecutor;
+import org.apache.flink.runtime.dispatcher.ArchivedExecutionGraphStore;
+import org.apache.flink.runtime.dispatcher.Dispatcher;
+import org.apache.flink.runtime.dispatcher.DispatcherGateway;
+import org.apache.flink.runtime.dispatcher.DispatcherId;
+import org.apache.flink.runtime.dispatcher.MiniDispatcher;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
import org.apache.flink.runtime.metrics.MetricRegistryImpl;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.akka.AkkaRpcService;
import org.apache.flink.runtime.security.SecurityConfiguration;
import org.apache.flink.runtime.security.SecurityContext;
import org.apache.flink.runtime.security.SecurityUtils;
+import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
+import org.apache.flink.runtime.webmonitor.retriever.LeaderGatewayRetriever;
+import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
+import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaQueryServiceRetriever;
+import org.apache.flink.runtime.webmonitor.retriever.impl.RpcGatewayRetriever;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
@@ -47,12 +70,15 @@ import akka.actor.ActorSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
import java.io.IOException;
+import java.net.InetSocketAddress;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
import scala.concurrent.duration.FiniteDuration;
@@ -63,6 +89,10 @@ import scala.concurrent.duration.FiniteDuration;
*/
public abstract class ClusterEntrypoint implements FatalErrorHandler {
+ public static final ConfigOption<String> EXECUTION_MODE = ConfigOptions
+ .key("internal.cluster.execution-mode")
+ .defaultValue(ExecutionMode.NORMAL.toString());
+
protected static final Logger LOG = LoggerFactory.getLogger(ClusterEntrypoint.class);
protected static final int SUCCESS_RETURN_CODE = 0;
@@ -76,20 +106,43 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
private final CompletableFuture<Boolean> terminationFuture;
+ private final AtomicBoolean isTerminating = new AtomicBoolean(false);
+
@GuardedBy("lock")
- private MetricRegistryImpl metricRegistry = null;
+ private MetricRegistryImpl metricRegistry;
@GuardedBy("lock")
- private HighAvailabilityServices haServices = null;
+ private HighAvailabilityServices haServices;
@GuardedBy("lock")
- private BlobServer blobServer = null;
+ private BlobServer blobServer;
@GuardedBy("lock")
- private HeartbeatServices heartbeatServices = null;
+ private HeartbeatServices heartbeatServices;
@GuardedBy("lock")
- private RpcService commonRpcService = null;
+ private RpcService commonRpcService;
+
+ @GuardedBy("lock")
+ private ResourceManager<?> resourceManager;
+
+ @GuardedBy("lock")
+ private Dispatcher dispatcher;
+
+ @GuardedBy("lock")
+ private LeaderRetrievalService dispatcherLeaderRetrievalService;
+
+ @GuardedBy("lock")
+ private LeaderRetrievalService resourceManagerRetrievalService;
+
+ @GuardedBy("lock")
+ private WebMonitorEndpoint<?> webMonitorEndpoint;
+
+ @GuardedBy("lock")
+ private ArchivedExecutionGraphStore archivedExecutionGraphStore;
+
+ @GuardedBy("lock")
+ private TransientBlobCache transientBlobCache;
protected ClusterEntrypoint(Configuration configuration) {
this.configuration = Preconditions.checkNotNull(configuration);
@@ -108,24 +161,18 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
SecurityContext securityContext = installSecurityContext(configuration);
- securityContext.runSecured(new Callable<Void>() {
- @Override
- public Void call() throws Exception {
- runCluster(configuration);
+ securityContext.runSecured((Callable<Void>) () -> {
+ runCluster(configuration);
- return null;
- }
+ return null;
});
} catch (Throwable t) {
LOG.error("Cluster initialization failed.", t);
- try {
- shutDown(false);
- } catch (Throwable st) {
- LOG.error("Could not properly shut down cluster entrypoint.", st);
- }
-
- System.exit(STARTUP_FAILURE_RETURN_CODE);
+ shutDownAndTerminate(
+ STARTUP_FAILURE_RETURN_CODE,
+ ApplicationStatus.FAILED,
+ false);
}
}
@@ -163,33 +210,134 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
blobServer,
heartbeatServices,
metricRegistry);
+
+ // TODO: Make shutDownAndTerminate non blocking to not use the global executor
+ dispatcher.getTerminationFuture().whenCompleteAsync(
+ (Boolean success, Throwable throwable) -> {
+ if (throwable != null) {
+ LOG.info("Could not properly terminate the Dispatcher.", throwable);
+ }
+
+ shutDownAndTerminate(
+ SUCCESS_RETURN_CODE,
+ ApplicationStatus.SUCCEEDED,
+ true);
+ });
}
}
protected void initializeServices(Configuration configuration) throws Exception {
- assert(Thread.holdsLock(lock));
LOG.info("Initializing cluster services.");
- final String bindAddress = configuration.getString(JobManagerOptions.ADDRESS);
- final String portRange = getRPCPortRange(configuration);
+ synchronized (lock) {
+ final String bindAddress = configuration.getString(JobManagerOptions.ADDRESS);
+ final String portRange = getRPCPortRange(configuration);
- commonRpcService = createRpcService(configuration, bindAddress, portRange);
+ commonRpcService = createRpcService(configuration, bindAddress, portRange);
- // update the configuration used to create the high availability services
- configuration.setString(JobManagerOptions.ADDRESS, commonRpcService.getAddress());
- configuration.setInteger(JobManagerOptions.PORT, commonRpcService.getPort());
+ // update the configuration used to create the high availability services
+ configuration.setString(JobManagerOptions.ADDRESS, commonRpcService.getAddress());
+ configuration.setInteger(JobManagerOptions.PORT, commonRpcService.getPort());
- haServices = createHaServices(configuration, commonRpcService.getExecutor());
- blobServer = new BlobServer(configuration, haServices.createBlobStore());
- blobServer.start();
- heartbeatServices = createHeartbeatServices(configuration);
- metricRegistry = createMetricRegistry(configuration);
+ haServices = createHaServices(configuration, commonRpcService.getExecutor());
+ blobServer = new BlobServer(configuration, haServices.createBlobStore());
+ blobServer.start();
+ heartbeatServices = createHeartbeatServices(configuration);
+ metricRegistry = createMetricRegistry(configuration);
- // TODO: This is a temporary hack until we have ported the MetricQueryService to the new RpcEndpoint
- // start the MetricQueryService
- final ActorSystem actorSystem = ((AkkaRpcService) commonRpcService).getActorSystem();
- metricRegistry.startQueryService(actorSystem, null);
+ // TODO: This is a temporary hack until we have ported the MetricQueryService to the new RpcEndpoint
+ // start the MetricQueryService
+ final ActorSystem actorSystem = ((AkkaRpcService) commonRpcService).getActorSystem();
+ metricRegistry.startQueryService(actorSystem, null);
+ }
+ }
+
+ protected void startClusterComponents(
+ Configuration configuration,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ BlobServer blobServer,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry) throws Exception {
+ synchronized (lock) {
+ archivedExecutionGraphStore = createSerializableExecutionGraphStore(configuration, rpcService.getScheduledExecutor());
+
+ dispatcherLeaderRetrievalService = highAvailabilityServices.getDispatcherLeaderRetriever();
+
+ resourceManagerRetrievalService = highAvailabilityServices.getResourceManagerLeaderRetriever();
+
+ final ClusterInformation clusterInformation = new ClusterInformation(
+ rpcService.getAddress(),
+ blobServer.getPort());
+
+ transientBlobCache = new TransientBlobCache(
+ configuration,
+ new InetSocketAddress(
+ clusterInformation.getBlobServerHostname(),
+ clusterInformation.getBlobServerPort()));
+
+ LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever = new RpcGatewayRetriever<>(
+ rpcService,
+ DispatcherGateway.class,
+ DispatcherId::new,
+ 10,
+ Time.milliseconds(50L));
+
+ LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever = new RpcGatewayRetriever<>(
+ rpcService,
+ ResourceManagerGateway.class,
+ ResourceManagerId::new,
+ 10,
+ Time.milliseconds(50L));
+
+ // TODO: Remove once we have ported the MetricFetcher to the RpcEndpoint
+ final ActorSystem actorSystem = ((AkkaRpcService) rpcService).getActorSystem();
+ final Time timeout = Time.milliseconds(configuration.getLong(WebOptions.TIMEOUT));
+
+ webMonitorEndpoint = createRestEndpoint(
+ configuration,
+ dispatcherGatewayRetriever,
+ resourceManagerGatewayRetriever,
+ transientBlobCache,
+ rpcService.getExecutor(),
+ new AkkaQueryServiceRetriever(actorSystem, timeout),
+ highAvailabilityServices.getWebMonitorLeaderElectionService());
+
+ LOG.debug("Starting Dispatcher REST endpoint.");
+ webMonitorEndpoint.start();
+
+ resourceManager = createResourceManager(
+ configuration,
+ ResourceID.generate(),
+ rpcService,
+ highAvailabilityServices,
+ heartbeatServices,
+ metricRegistry,
+ this,
+ clusterInformation,
+ webMonitorEndpoint.getRestAddress());
+
+ dispatcher = createDispatcher(
+ configuration,
+ rpcService,
+ highAvailabilityServices,
+ resourceManager.getSelfGateway(ResourceManagerGateway.class),
+ blobServer,
+ heartbeatServices,
+ metricRegistry,
+ archivedExecutionGraphStore,
+ this,
+ webMonitorEndpoint.getRestAddress());
+
+ LOG.debug("Starting ResourceManager.");
+ resourceManager.start();
+ resourceManagerRetrievalService.start(resourceManagerGatewayRetriever);
+
+ LOG.debug("Starting Dispatcher.");
+ dispatcher.start();
+ dispatcherLeaderRetrievalService.start(dispatcherGatewayRetriever);
+ }
}
/**
@@ -228,7 +376,7 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
return new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(configuration));
}
- protected void shutDown(boolean cleanupHaData) throws FlinkException {
+ private void shutDown(boolean cleanupHaData) throws FlinkException {
LOG.info("Stopping {}.", getClass().getSimpleName());
Throwable exception = null;
@@ -236,11 +384,29 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
synchronized (lock) {
try {
- stopClusterComponents(cleanupHaData);
+ stopClusterComponents();
+ } catch (Throwable t) {
+ exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ }
+
+ try {
+ stopClusterServices(cleanupHaData);
} catch (Throwable t) {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
}
+ terminationFuture.complete(true);
+ }
+
+ if (exception != null) {
+ throw new FlinkException("Could not properly shut down the cluster entrypoint.", exception);
+ }
+ }
+
+ protected void stopClusterServices(boolean cleanupHaData) throws FlinkException {
+ Throwable exception = null;
+
+ synchronized (lock) {
if (metricRegistry != null) {
try {
metricRegistry.shutdown();
@@ -276,8 +442,6 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
}
}
-
- terminationFuture.complete(true);
}
if (exception != null) {
@@ -285,6 +449,68 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
}
}
+ protected void stopClusterComponents() throws Exception {
+ synchronized (lock) {
+ Throwable exception = null;
+
+ if (webMonitorEndpoint != null) {
+ webMonitorEndpoint.shutdown(Time.seconds(10L));
+ }
+
+ if (dispatcherLeaderRetrievalService != null) {
+ try {
+ dispatcherLeaderRetrievalService.stop();
+ } catch (Throwable t) {
+ exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ }
+ }
+
+ if (dispatcher != null) {
+ try {
+ dispatcher.shutDown();
+ } catch (Throwable t) {
+ exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ }
+ }
+
+ if (resourceManagerRetrievalService != null) {
+ try {
+ resourceManagerRetrievalService.stop();
+ } catch (Throwable t) {
+ exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ }
+ }
+
+ if (resourceManager != null) {
+ try {
+ resourceManager.shutDown();
+ } catch (Throwable t) {
+ exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ }
+ }
+
+ if (archivedExecutionGraphStore != null) {
+ try {
+ archivedExecutionGraphStore.close();
+ } catch (Throwable t) {
+ exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ }
+ }
+
+ if (transientBlobCache != null) {
+ try {
+ transientBlobCache.close();
+ } catch (Throwable t) {
+ exception = ExceptionUtils.firstOrSuppressed(t, exception);
+ }
+ }
+
+ if (exception != null) {
+ throw new FlinkException("Could not properly shut down the session cluster entry point.", exception);
+ }
+ }
+ }
+
@Override
public void onFatalError(Throwable exception) {
LOG.error("Fatal error occurred in the cluster entrypoint.", exception);
@@ -292,16 +518,74 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
System.exit(RUNTIME_FAILURE_RETURN_CODE);
}
- protected abstract void startClusterComponents(
+ // --------------------------------------------------
+ // Internal methods
+ // --------------------------------------------------
+
+ private void shutDownAndTerminate(
+ int returnCode,
+ ApplicationStatus applicationStatus,
+ boolean cleanupHaData) {
+
+ LOG.info("Shut down and terminate {} with return code {} and application status {}.",
+ getClass().getSimpleName(),
+ returnCode,
+ applicationStatus);
+
+ if (isTerminating.compareAndSet(false, true)) {
+ try {
+ shutDown(cleanupHaData);
+ } catch (Throwable t) {
+ LOG.info("Could not properly shut down cluster entrypoint.", t);
+ }
+
+ System.exit(returnCode);
+ } else {
+ LOG.debug("Concurrent termination call detected. Ignoring termination call with return code {} and application status {}.",
+ returnCode,
+ applicationStatus);
+ }
+ }
+
+ // --------------------------------------------------
+ // Abstract methods
+ // --------------------------------------------------
+
+ protected abstract Dispatcher createDispatcher(
Configuration configuration,
RpcService rpcService,
HighAvailabilityServices highAvailabilityServices,
+ ResourceManagerGateway resourceManagerGateway,
BlobServer blobServer,
HeartbeatServices heartbeatServices,
- MetricRegistry metricRegistry) throws Exception;
+ MetricRegistry metricRegistry,
+ ArchivedExecutionGraphStore archivedExecutionGraphStore,
+ FatalErrorHandler fatalErrorHandler,
+ @Nullable String restAddress) throws Exception;
- protected void stopClusterComponents(boolean cleanupHaData) throws Exception {
- }
+ protected abstract ResourceManager<?> createResourceManager(
+ Configuration configuration,
+ ResourceID resourceId,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ FatalErrorHandler fatalErrorHandler,
+ ClusterInformation clusterInformation,
+ @Nullable String webInterfaceUrl) throws Exception;
+
+ protected abstract WebMonitorEndpoint<?> createRestEndpoint(
+ Configuration configuration,
+ LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever,
+ LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
+ TransientBlobService transientBlobService,
+ Executor executor,
+ MetricQueryServiceRetriever metricQueryServiceRetriever,
+ LeaderElectionService leaderElectionService) throws Exception;
+
+ protected abstract ArchivedExecutionGraphStore createSerializableExecutionGraphStore(
+ Configuration configuration,
+ ScheduledExecutor scheduledExecutor) throws IOException;
protected static ClusterConfiguration parseArguments(String[] args) {
ParameterTool parameterTool = ParameterTool.fromArgs(args);
@@ -314,4 +598,19 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
protected static Configuration loadConfiguration(ClusterConfiguration clusterConfiguration) {
return GlobalConfiguration.loadConfiguration(clusterConfiguration.getConfigDir());
}
+
+ /**
+ * Execution mode of the {@link MiniDispatcher}.
+ */
+ public enum ExecutionMode {
+ /**
+ * Waits until the job result has been served.
+ */
+ NORMAL,
+
+ /**
+ * Directly stops after the job has finished.
+ */
+ DETACHED
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
index e03c318..dc211d8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
@@ -18,52 +18,33 @@
package org.apache.flink.runtime.entrypoint;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.WebOptions;
import org.apache.flink.runtime.blob.BlobServer;
-import org.apache.flink.runtime.blob.TransientBlobCache;
import org.apache.flink.runtime.blob.TransientBlobService;
-import org.apache.flink.runtime.clusterframework.ApplicationStatus;
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
-import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.concurrent.ScheduledExecutor;
+import org.apache.flink.runtime.dispatcher.ArchivedExecutionGraphStore;
+import org.apache.flink.runtime.dispatcher.Dispatcher;
+import org.apache.flink.runtime.dispatcher.DispatcherGateway;
+import org.apache.flink.runtime.dispatcher.MemoryArchivedExecutionGraphStore;
+import org.apache.flink.runtime.dispatcher.MiniDispatcher;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.jobmanager.OnCompletionActions;
-import org.apache.flink.runtime.jobmaster.JobManagerRunner;
-import org.apache.flink.runtime.jobmaster.JobManagerSharedServices;
-import org.apache.flink.runtime.jobmaster.JobMasterGateway;
-import org.apache.flink.runtime.jobmaster.JobMasterId;
-import org.apache.flink.runtime.jobmaster.JobMasterRestEndpoint;
+import org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint;
import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.metrics.MetricRegistry;
-import org.apache.flink.runtime.resourcemanager.ResourceManager;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
-import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
-import org.apache.flink.runtime.rpc.akka.AkkaRpcService;
-import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
import org.apache.flink.runtime.webmonitor.retriever.LeaderGatewayRetriever;
import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
-import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaQueryServiceRetriever;
-import org.apache.flink.runtime.webmonitor.retriever.impl.RpcGatewayRetriever;
-import org.apache.flink.util.ConfigurationException;
-import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.Preconditions;
-
-import akka.actor.ActorSystem;
import javax.annotation.Nullable;
-import java.net.InetSocketAddress;
+import java.io.IOException;
import java.util.concurrent.Executor;
/**
@@ -71,123 +52,24 @@ import java.util.concurrent.Executor;
*/
public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
- private ResourceManager<?> resourceManager;
-
- private JobManagerSharedServices jobManagerSharedServices;
-
- private JobMasterRestEndpoint jobMasterRestEndpoint;
-
- private LeaderRetrievalService jobMasterRetrievalService;
-
- private LeaderRetrievalService resourceManagerRetrievalService;
-
- private TransientBlobCache transientBlobCache;
-
- private JobManagerRunner jobManagerRunner;
-
public JobClusterEntrypoint(Configuration configuration) {
super(configuration);
}
@Override
- protected void startClusterComponents(
+ protected MiniDispatcherRestEndpoint createRestEndpoint(
Configuration configuration,
- RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- BlobServer blobServer,
- HeartbeatServices heartbeatServices,
- MetricRegistry metricRegistry) throws Exception {
-
- jobManagerSharedServices = JobManagerSharedServices.fromConfiguration(configuration, blobServer);
-
- resourceManagerRetrievalService = highAvailabilityServices.getResourceManagerLeaderRetriever();
-
- final LeaderGatewayRetriever<JobMasterGateway> jobMasterGatewayRetriever = new RpcGatewayRetriever<>(
- rpcService,
- JobMasterGateway.class,
- JobMasterId::new,
- 10,
- Time.milliseconds(50L));
-
- final LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever = new RpcGatewayRetriever<>(
- rpcService,
- ResourceManagerGateway.class,
- ResourceManagerId::new,
- 10,
- Time.milliseconds(50L));
-
- // TODO: Remove once we have ported the MetricFetcher to the RpcEndpoint
- final ActorSystem actorSystem = ((AkkaRpcService) rpcService).getActorSystem();
- final Time timeout = Time.milliseconds(configuration.getLong(WebOptions.TIMEOUT));
-
- final ClusterInformation clusterInformation = new ClusterInformation(rpcService.getAddress(), blobServer.getPort());
-
- transientBlobCache = new TransientBlobCache(
- configuration,
- new InetSocketAddress(clusterInformation.getBlobServerHostname(), clusterInformation.getBlobServerPort()));
-
- jobMasterRestEndpoint = createJobMasterRestEndpoint(
- configuration,
- jobMasterGatewayRetriever,
- resourceManagerGatewayRetriever,
- transientBlobCache,
- rpcService.getExecutor(),
- new AkkaQueryServiceRetriever(actorSystem, timeout),
- highAvailabilityServices.getWebMonitorLeaderElectionService());
-
- LOG.debug("Starting JobMaster REST endpoint.");
- jobMasterRestEndpoint.start();
-
- resourceManager = createResourceManager(
- configuration,
- ResourceID.generate(),
- rpcService,
- highAvailabilityServices,
- heartbeatServices,
- metricRegistry,
- this,
- clusterInformation,
- jobMasterRestEndpoint.getRestAddress());
-
- jobManagerRunner = createJobManagerRunner(
- configuration,
- ResourceID.generate(),
- rpcService,
- highAvailabilityServices,
- jobManagerSharedServices,
- heartbeatServices,
- blobServer,
- metricRegistry,
- this,
- jobMasterRestEndpoint.getRestAddress());
-
- LOG.debug("Starting ResourceManager.");
- resourceManager.start();
- resourceManagerRetrievalService.start(resourceManagerGatewayRetriever);
-
- LOG.debug("Starting JobManager.");
- jobManagerRunner.start();
-
- jobMasterRetrievalService = highAvailabilityServices.getJobManagerLeaderRetriever(
- jobManagerRunner.getJobGraph().getJobID(),
- jobManagerRunner.getAddress());
- jobMasterRetrievalService.start(jobMasterGatewayRetriever);
- }
-
- protected JobMasterRestEndpoint createJobMasterRestEndpoint(
- Configuration configuration,
- GatewayRetriever<JobMasterGateway> jobMasterGatewayRetriever,
- GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
+ LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever,
+ LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
TransientBlobService transientBlobService,
Executor executor,
MetricQueryServiceRetriever metricQueryServiceRetriever,
- LeaderElectionService leaderElectionService) throws ConfigurationException {
-
+ LeaderElectionService leaderElectionService) throws Exception {
final RestHandlerConfiguration restHandlerConfiguration = RestHandlerConfiguration.fromConfiguration(configuration);
- return new JobMasterRestEndpoint(
+ return new MiniDispatcherRestEndpoint(
RestServerEndpointConfiguration.fromConfiguration(configuration),
- jobMasterGatewayRetriever,
+ dispatcherGatewayRetriever,
configuration,
restHandlerConfiguration,
resourceManagerGatewayRetriever,
@@ -196,155 +78,50 @@ public abstract class JobClusterEntrypoint extends ClusterEntrypoint {
metricQueryServiceRetriever,
leaderElectionService,
this);
+ }
+ @Override
+ protected ArchivedExecutionGraphStore createSerializableExecutionGraphStore(
+ Configuration configuration,
+ ScheduledExecutor scheduledExecutor) throws IOException {
+ return new MemoryArchivedExecutionGraphStore();
}
- protected JobManagerRunner createJobManagerRunner(
+ @Override
+ protected Dispatcher createDispatcher(
Configuration configuration,
- ResourceID resourceId,
RpcService rpcService,
HighAvailabilityServices highAvailabilityServices,
- JobManagerSharedServices jobManagerSharedServices,
- HeartbeatServices heartbeatServices,
+ ResourceManagerGateway resourceManagerGateway,
BlobServer blobServer,
+ HeartbeatServices heartbeatServices,
MetricRegistry metricRegistry,
+ ArchivedExecutionGraphStore archivedExecutionGraphStore,
FatalErrorHandler fatalErrorHandler,
@Nullable String restAddress) throws Exception {
final JobGraph jobGraph = retrieveJobGraph(configuration);
- return new JobManagerRunner(
- resourceId,
- jobGraph,
- configuration,
+ final String executionModeValue = configuration.getString(EXECUTION_MODE);
+
+ final ExecutionMode executionMode = ExecutionMode.valueOf(executionModeValue);
+
+ return new MiniDispatcher(
rpcService,
+ Dispatcher.DISPATCHER_NAME,
+ configuration,
highAvailabilityServices,
- heartbeatServices,
+ resourceManagerGateway,
blobServer,
- jobManagerSharedServices,
+ heartbeatServices,
metricRegistry,
- new TerminatingOnCompleteActions(jobGraph.getJobID()),
+ archivedExecutionGraphStore,
+ Dispatcher.DefaultJobManagerRunnerFactory.INSTANCE,
fatalErrorHandler,
- restAddress);
- }
-
- @Override
- protected void stopClusterComponents(boolean cleanupHaData) throws Exception {
- Throwable exception = null;
-
- if (jobMasterRestEndpoint != null) {
- jobMasterRestEndpoint.shutdown(Time.seconds(10L));
- }
-
- if (jobMasterRetrievalService != null) {
- try {
- jobMasterRetrievalService.stop();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (jobManagerRunner != null) {
- try {
- jobManagerRunner.shutdown();
- } catch (Throwable t) {
- exception = t;
- }
- }
-
- if (jobManagerSharedServices != null) {
- try {
- jobManagerSharedServices.shutdown();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (resourceManagerRetrievalService != null) {
- try {
- resourceManagerRetrievalService.stop();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (resourceManager != null) {
- try {
- resourceManager.shutDown();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (transientBlobCache != null) {
- try {
- transientBlobCache.close();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (exception != null) {
- throw new FlinkException("Could not properly shut down the job cluster entry point.", exception);
- }
- }
-
- private void shutDownAndTerminate(
- boolean cleanupHaData,
- ApplicationStatus status,
- @Nullable String optionalDiagnostics) {
- try {
- if (resourceManager != null) {
- resourceManager.shutDownCluster(status, optionalDiagnostics);
- }
-
- shutDown(cleanupHaData);
- } catch (Throwable t) {
- LOG.error("Could not properly shut down cluster entrypoint.", t);
- }
-
- System.exit(SUCCESS_RETURN_CODE);
+ restAddress,
+ jobGraph,
+ executionMode);
}
- protected abstract ResourceManager<?> createResourceManager(
- Configuration configuration,
- ResourceID resourceId,
- RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- HeartbeatServices heartbeatServices,
- MetricRegistry metricRegistry,
- FatalErrorHandler fatalErrorHandler,
- ClusterInformation clusterInformation,
- @Nullable String webInterfaceUrl) throws Exception;
-
protected abstract JobGraph retrieveJobGraph(Configuration configuration) throws FlinkException;
-
- private final class TerminatingOnCompleteActions implements OnCompletionActions {
-
- private final JobID jobId;
-
- private TerminatingOnCompleteActions(JobID jobId) {
- this.jobId = Preconditions.checkNotNull(jobId);
- }
-
- @Override
- public void jobReachedGloballyTerminalState(ArchivedExecutionGraph executionGraph) {
- LOG.info("Job({}) finished.", jobId);
-
- final ErrorInfo errorInfo = executionGraph.getFailureInfo();
-
- if (errorInfo == null) {
- shutDownAndTerminate(true, ApplicationStatus.SUCCEEDED, null);
- } else {
- shutDownAndTerminate(true, ApplicationStatus.FAILED, errorInfo.getExceptionAsString());
- }
- }
-
- @Override
- public void jobFinishedByOther() {
- LOG.info("Job({}) was finished by another JobManager.", jobId);
-
- shutDownAndTerminate(false, ApplicationStatus.UNKNOWN, "Job was finished by another master");
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
index 0f41c9a..764356d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
@@ -22,48 +22,33 @@ import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ConfigurationUtils;
import org.apache.flink.configuration.JobManagerOptions;
-import org.apache.flink.configuration.WebOptions;
import org.apache.flink.runtime.blob.BlobServer;
-import org.apache.flink.runtime.blob.TransientBlobCache;
import org.apache.flink.runtime.blob.TransientBlobService;
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.ScheduledExecutor;
import org.apache.flink.runtime.dispatcher.ArchivedExecutionGraphStore;
import org.apache.flink.runtime.dispatcher.Dispatcher;
import org.apache.flink.runtime.dispatcher.DispatcherGateway;
-import org.apache.flink.runtime.dispatcher.DispatcherId;
import org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint;
import org.apache.flink.runtime.dispatcher.FileArchivedExecutionGraphStore;
import org.apache.flink.runtime.dispatcher.StandaloneDispatcher;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.metrics.MetricRegistry;
-import org.apache.flink.runtime.resourcemanager.ResourceManager;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
-import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
-import org.apache.flink.runtime.rpc.akka.AkkaRpcService;
import org.apache.flink.runtime.webmonitor.retriever.LeaderGatewayRetriever;
import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
-import org.apache.flink.runtime.webmonitor.retriever.impl.AkkaQueryServiceRetriever;
-import org.apache.flink.runtime.webmonitor.retriever.impl.RpcGatewayRetriever;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.FlinkException;
import org.apache.flink.shaded.guava18.com.google.common.base.Ticker;
-import akka.actor.ActorSystem;
-
import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.concurrent.Executor;
/**
@@ -71,110 +56,12 @@ import java.util.concurrent.Executor;
*/
public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
- private ResourceManager<?> resourceManager;
-
- private Dispatcher dispatcher;
-
- private LeaderRetrievalService dispatcherLeaderRetrievalService;
-
- private LeaderRetrievalService resourceManagerRetrievalService;
-
- private DispatcherRestEndpoint dispatcherRestEndpoint;
-
- private ArchivedExecutionGraphStore archivedExecutionGraphStore;
-
- private TransientBlobCache transientBlobCache;
-
public SessionClusterEntrypoint(Configuration configuration) {
super(configuration);
}
@Override
- protected void startClusterComponents(
- Configuration configuration,
- RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- BlobServer blobServer,
- HeartbeatServices heartbeatServices,
- MetricRegistry metricRegistry) throws Exception {
-
- archivedExecutionGraphStore = createSerializableExecutionGraphStore(configuration, rpcService.getScheduledExecutor());
-
- dispatcherLeaderRetrievalService = highAvailabilityServices.getDispatcherLeaderRetriever();
-
- resourceManagerRetrievalService = highAvailabilityServices.getResourceManagerLeaderRetriever();
-
- LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever = new RpcGatewayRetriever<>(
- rpcService,
- DispatcherGateway.class,
- DispatcherId::new,
- 10,
- Time.milliseconds(50L));
-
- LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever = new RpcGatewayRetriever<>(
- rpcService,
- ResourceManagerGateway.class,
- ResourceManagerId::new,
- 10,
- Time.milliseconds(50L));
-
- // TODO: Remove once we have ported the MetricFetcher to the RpcEndpoint
- final ActorSystem actorSystem = ((AkkaRpcService) rpcService).getActorSystem();
- final Time timeout = Time.milliseconds(configuration.getLong(WebOptions.TIMEOUT));
-
- final ClusterInformation clusterInformation = new ClusterInformation(
- rpcService.getAddress(),
- blobServer.getPort());
-
- transientBlobCache = new TransientBlobCache(
- configuration,
- new InetSocketAddress(clusterInformation.getBlobServerHostname(), clusterInformation.getBlobServerPort()));
-
- dispatcherRestEndpoint = createDispatcherRestEndpoint(
- configuration,
- dispatcherGatewayRetriever,
- resourceManagerGatewayRetriever,
- transientBlobCache,
- rpcService.getExecutor(),
- new AkkaQueryServiceRetriever(actorSystem, timeout),
- highAvailabilityServices.getWebMonitorLeaderElectionService());
-
- LOG.debug("Starting Dispatcher REST endpoint.");
- dispatcherRestEndpoint.start();
-
- resourceManager = createResourceManager(
- configuration,
- ResourceID.generate(),
- rpcService,
- highAvailabilityServices,
- heartbeatServices,
- metricRegistry,
- this,
- clusterInformation,
- dispatcherRestEndpoint.getRestAddress());
-
- dispatcher = createDispatcher(
- configuration,
- rpcService,
- highAvailabilityServices,
- resourceManager.getSelfGateway(ResourceManagerGateway.class),
- blobServer,
- heartbeatServices,
- metricRegistry,
- archivedExecutionGraphStore,
- this,
- dispatcherRestEndpoint.getRestAddress());
-
- LOG.debug("Starting ResourceManager.");
- resourceManager.start();
- resourceManagerRetrievalService.start(resourceManagerGatewayRetriever);
-
- LOG.debug("Starting Dispatcher.");
- dispatcher.start();
- dispatcherLeaderRetrievalService.start(dispatcherGatewayRetriever);
- }
-
- private ArchivedExecutionGraphStore createSerializableExecutionGraphStore(
+ protected ArchivedExecutionGraphStore createSerializableExecutionGraphStore(
Configuration configuration,
ScheduledExecutor scheduledExecutor) throws IOException {
final File tmpDir = new File(ConfigurationUtils.parseTempDirectories(configuration)[0]);
@@ -191,74 +78,14 @@ public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
}
@Override
- protected void stopClusterComponents(boolean cleanupHaData) throws Exception {
- Throwable exception = null;
-
- if (dispatcherRestEndpoint != null) {
- dispatcherRestEndpoint.shutdown(Time.seconds(10L));
- }
-
- if (dispatcherLeaderRetrievalService != null) {
- try {
- dispatcherLeaderRetrievalService.stop();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (dispatcher != null) {
- try {
- dispatcher.shutDown();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (resourceManagerRetrievalService != null) {
- try {
- resourceManagerRetrievalService.stop();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (resourceManager != null) {
- try {
- resourceManager.shutDown();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (archivedExecutionGraphStore != null) {
- try {
- archivedExecutionGraphStore.close();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (transientBlobCache != null) {
- try {
- transientBlobCache.close();
- } catch (Throwable t) {
- exception = ExceptionUtils.firstOrSuppressed(t, exception);
- }
- }
-
- if (exception != null) {
- throw new FlinkException("Could not properly shut down the session cluster entry point.", exception);
- }
- }
-
- protected DispatcherRestEndpoint createDispatcherRestEndpoint(
- Configuration configuration,
- LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever,
- LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
+ protected DispatcherRestEndpoint createRestEndpoint(
+ Configuration configuration,
+ LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever,
+ LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
TransientBlobService transientBlobService,
- Executor executor,
- MetricQueryServiceRetriever metricQueryServiceRetriever,
- LeaderElectionService leaderElectionService) throws Exception {
+ Executor executor,
+ MetricQueryServiceRetriever metricQueryServiceRetriever,
+ LeaderElectionService leaderElectionService) throws Exception {
final RestHandlerConfiguration restHandlerConfiguration = RestHandlerConfiguration.fromConfiguration(configuration);
@@ -275,6 +102,7 @@ public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
this);
}
+ @Override
protected Dispatcher createDispatcher(
Configuration configuration,
RpcService rpcService,
@@ -298,18 +126,8 @@ public abstract class SessionClusterEntrypoint extends ClusterEntrypoint {
heartbeatServices,
metricRegistry,
archivedExecutionGraphStore,
+ Dispatcher.DefaultJobManagerRunnerFactory.INSTANCE,
fatalErrorHandler,
restAddress);
}
-
- protected abstract ResourceManager<?> createResourceManager(
- Configuration configuration,
- ResourceID resourceId,
- RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- HeartbeatServices heartbeatServices,
- MetricRegistry metricRegistry,
- FatalErrorHandler fatalErrorHandler,
- ClusterInformation clusterInformation,
- @Nullable String webInterfaceUrl) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterRestEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterRestEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterRestEndpoint.java
deleted file mode 100644
index b0fa46d..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterRestEndpoint.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.jobmaster;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.blob.TransientBlobService;
-import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
-import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
-import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
-import org.apache.flink.runtime.rpc.FatalErrorHandler;
-import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
-import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
-import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
-
-import java.util.concurrent.Executor;
-
-/**
- * REST endpoint for the {@link JobMaster}.
- */
-public class JobMasterRestEndpoint extends WebMonitorEndpoint<JobMasterGateway> {
-
- public JobMasterRestEndpoint(
- RestServerEndpointConfiguration endpointConfiguration,
- GatewayRetriever<JobMasterGateway> leaderRetriever,
- Configuration clusterConfiguration,
- RestHandlerConfiguration restConfiguration,
- GatewayRetriever<ResourceManagerGateway> resourceManagerRetriever,
- TransientBlobService transientBlobService,
- Executor executor,
- MetricQueryServiceRetriever metricQueryServiceRetriever,
- LeaderElectionService leaderElectionService,
- FatalErrorHandler fatalErrorHandler) {
- super(
- endpointConfiguration,
- leaderRetriever,
- clusterConfiguration,
- restConfiguration,
- resourceManagerRetriever,
- transientBlobService,
- executor,
- metricQueryServiceRetriever,
- leaderElectionService,
- fatalErrorHandler);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
new file mode 100644
index 0000000..54ad526
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniDispatcherRestEndpoint.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.jobmaster;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.TransientBlobService;
+import org.apache.flink.runtime.entrypoint.JobClusterEntrypoint;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
+import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.WebMonitorEndpoint;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever;
+
+import java.util.concurrent.Executor;
+
+/**
+ * REST endpoint for the {@link JobClusterEntrypoint}.
+ */
+public class MiniDispatcherRestEndpoint extends WebMonitorEndpoint<RestfulGateway> {
+
+ public MiniDispatcherRestEndpoint(
+ RestServerEndpointConfiguration endpointConfiguration,
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Configuration clusterConfiguration,
+ RestHandlerConfiguration restConfiguration,
+ GatewayRetriever<ResourceManagerGateway> resourceManagerRetriever,
+ TransientBlobService transientBlobService,
+ Executor executor,
+ MetricQueryServiceRetriever metricQueryServiceRetriever,
+ LeaderElectionService leaderElectionService,
+ FatalErrorHandler fatalErrorHandler) {
+ super(
+ endpointConfiguration,
+ leaderRetriever,
+ clusterConfiguration,
+ restConfiguration,
+ resourceManagerRetriever,
+ transientBlobService,
+ executor,
+ metricQueryServiceRetriever,
+ leaderElectionService,
+ fatalErrorHandler);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/BlobServerPortHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/BlobServerPortHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/BlobServerPortHandler.java
index 1d14563..4b5fa89 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/BlobServerPortHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/BlobServerPortHandler.java
@@ -45,7 +45,7 @@ public final class BlobServerPortHandler extends AbstractRestHandler<DispatcherG
public BlobServerPortHandler(
CompletableFuture<String> localRestAddress,
- GatewayRetriever<DispatcherGateway> leaderRetriever,
+ GatewayRetriever<? extends DispatcherGateway> leaderRetriever,
Time timeout,
Map<String, String> headers) {
super(localRestAddress, leaderRetriever, timeout, headers, BlobServerPortHeaders.getInstance());
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandler.java
index efb162e..ac0a17b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobSubmitHandler.java
@@ -46,7 +46,7 @@ public final class JobSubmitHandler extends AbstractRestHandler<DispatcherGatewa
public JobSubmitHandler(
CompletableFuture<String> localRestAddress,
- GatewayRetriever<DispatcherGateway> leaderRetriever,
+ GatewayRetriever<? extends DispatcherGateway> leaderRetriever,
Time timeout,
Map<String, String> headers) {
super(localRestAddress, leaderRetriever, timeout, headers, JobSubmitHeaders.getInstance());
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/files/StaticFileServerHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/files/StaticFileServerHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/files/StaticFileServerHandler.java
index 650647b..5159e12 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/files/StaticFileServerHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/files/StaticFileServerHandler.java
@@ -110,7 +110,7 @@ public class StaticFileServerHandler<T extends RestfulGateway> extends RedirectH
private final File rootPath;
public StaticFileServerHandler(
- GatewayRetriever<T> retriever,
+ GatewayRetriever<? extends T> retriever,
CompletableFuture<String> localJobManagerAddressFuture,
Time timeout,
File rootPath) throws IOException {
http://git-wip-us.apache.org/repos/asf/flink/blob/84418d0b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java
index dc8d599..1b1c1e5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java
@@ -44,7 +44,7 @@ public final class TaskExecutorRegistrationSuccess extends RegistrationResponse.
/**
* Create a new {@code TaskExecutorRegistrationSuccess} message.
- *
+ *
* @param registrationId The ID that the ResourceManager assigned the registration.
* @param resourceManagerResourceId The unique ID that identifies the ResourceManager.
* @param heartbeatInterval The interval in which the ResourceManager will heartbeat the TaskExecutor.