flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From trohrm...@apache.org
Subject [36/50] [abbrv] flink git commit: [FLINK-4408] [JobManager] Introduce JobMasterRunner and implement job submission & setting up the ExecutionGraph
Date Thu, 06 Oct 2016 11:48:56 GMT
[FLINK-4408] [JobManager] Introduce JobMasterRunner and implement job submission & setting up the ExecutionGraph

This closes #2480


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/474ace75
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/474ace75
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/474ace75

Branch: refs/heads/flip-6
Commit: 474ace75f259ff4f949f9a6a9f2adca77cac53a4
Parents: 4d54a8c
Author: Kurt Young <ykt836@gmail.com>
Authored: Thu Sep 8 12:00:13 2016 +0800
Committer: Till Rohrmann <trohrmann@apache.org>
Committed: Thu Oct 6 13:38:44 2016 +0200

----------------------------------------------------------------------
 .../flink/api/common/JobExecutionResult.java    |   2 +-
 .../flink/api/common/JobSubmissionResult.java   |   2 +-
 .../HighAvailabilityServices.java               |  12 +
 .../runtime/highavailability/NonHaServices.java |  16 +-
 .../runtime/jobmanager/OnCompletionActions.java |  31 ++
 .../runtime/jobmanager/scheduler/Scheduler.java |   9 +
 .../runtime/jobmaster/JobManagerRunner.java     | 288 +++++++++++
 .../runtime/jobmaster/JobManagerServices.java   |  73 +++
 .../flink/runtime/jobmaster/JobMaster.java      | 485 ++++++++++++++-----
 .../runtime/jobmaster/JobMasterGateway.java     |  13 +
 .../jobmaster/MiniClusterJobDispatcher.java     | 385 +++++++++++++++
 .../flink/runtime/rpc/FatalErrorHandler.java    |  24 +
 .../runtime/taskexecutor/TaskExecutor.java      |  12 +
 .../TestingHighAvailabilityServices.java        |  39 +-
 .../jobmaster/JobManagerRunnerMockTest.java     | 254 ++++++++++
 .../flink/runtime/rpc/RpcConnectionTest.java    |  17 +-
 16 files changed, 1533 insertions(+), 129 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java b/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java
index cb4ecc5..7286cc5 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java
@@ -32,7 +32,7 @@ import java.util.concurrent.TimeUnit;
 @Public
 public class JobExecutionResult extends JobSubmissionResult {
 
-	private long netRuntime;
+	private final long netRuntime;
 
 	private final Map<String, Object> accumulatorResults;
 

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java b/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java
index c5dc869..b0e7e24 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java
@@ -26,7 +26,7 @@ import org.apache.flink.annotation.Public;
 @Public
 public class JobSubmissionResult {
 
-	private JobID jobID;
+	private final JobID jobID;
 
 	public JobSubmissionResult(JobID jobID) {
 		this.jobID = jobID;

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
index 7634176..d67e927 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java
@@ -19,6 +19,8 @@
 package org.apache.flink.runtime.highavailability;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
+import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
 import org.apache.flink.runtime.leaderelection.LeaderElectionService;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
 
@@ -61,4 +63,14 @@ public interface HighAvailabilityServices {
 	 * @param jobID The identifier of the job running the election.
 	 */
 	LeaderElectionService getJobMasterLeaderElectionService(JobID jobID) throws Exception;
+
+	/**
+	 * Gets the checkpoint recovery factory for the job manager
+	 */
+	CheckpointRecoveryFactory getCheckpointRecoveryFactory() throws Exception;
+
+	/**
+	 * Gets the submitted job graph store for the job manager
+	 */
+	SubmittedJobGraphStore getSubmittedJobGraphStore() throws Exception;
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/NonHaServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/NonHaServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/NonHaServices.java
index 33dc2d7..a2c9cc4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/NonHaServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/NonHaServices.java
@@ -19,13 +19,17 @@
 package org.apache.flink.runtime.highavailability;
 
 import org.apache.flink.api.common.JobID;
-import org.apache.flink.hadoop.shaded.org.jboss.netty.util.internal.ConcurrentHashMap;
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
+import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
+import org.apache.flink.runtime.jobmanager.StandaloneSubmittedJobGraphStore;
+import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
 import org.apache.flink.runtime.leaderelection.LeaderElectionService;
 import org.apache.flink.runtime.leaderelection.StandaloneLeaderElectionService;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
 import org.apache.flink.runtime.leaderretrieval.StandaloneLeaderRetrievalService;
 
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -88,4 +92,14 @@ public class NonHaServices implements HighAvailabilityServices {
 	public LeaderElectionService getJobMasterLeaderElectionService(JobID jobID) throws Exception {
 		return new StandaloneLeaderElectionService();
 	}
+
+	@Override
+	public CheckpointRecoveryFactory getCheckpointRecoveryFactory() throws Exception {
+		return new StandaloneCheckpointRecoveryFactory();
+	}
+
+	@Override
+	public SubmittedJobGraphStore getSubmittedJobGraphStore() throws Exception {
+		return new StandaloneSubmittedJobGraphStore();
+	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/OnCompletionActions.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/OnCompletionActions.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/OnCompletionActions.java
new file mode 100644
index 0000000..6de4253
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/OnCompletionActions.java
@@ -0,0 +1,31 @@
+/*
+ * 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.jobmanager;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+
+public interface OnCompletionActions extends FatalErrorHandler {
+
+	void jobFinished(JobExecutionResult result);
+
+	void jobFailed(Throwable cause);
+
+	void jobFinishedByOther();
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
index b839e0e..aa09314 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/Scheduler.java
@@ -31,6 +31,7 @@ import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 
 import akka.dispatch.Futures;
@@ -57,6 +58,7 @@ import org.apache.flink.util.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.concurrent.ExecutionContext;
+import scala.concurrent.ExecutionContext$;
 
 /**
  * The scheduler is responsible for distributing the ready-to-run tasks among instances and slots.
@@ -110,6 +112,13 @@ public class Scheduler implements InstanceListener, SlotAvailabilityListener, Sl
 	/**
 	 * Creates a new scheduler.
 	 */
+	public Scheduler(ExecutorService executor) {
+		this(ExecutionContext$.MODULE$.fromExecutor(executor));
+	}
+	
+	/**
+	 * Creates a new scheduler.
+	 */
 	public Scheduler(ExecutionContext executionContext) {
 		this.executionContext = executionContext;
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
new file mode 100644
index 0000000..bc2bf9a
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunner.java
@@ -0,0 +1,288 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
+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.jobmanager.scheduler.Scheduler;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.rpc.RpcService;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.UUID;
+import java.util.concurrent.Executor;
+
+/**
+ * The runner for the job manager. It deals with job level leader election and make underlying job manager
+ * properly reacted.
+ */
+public class JobManagerRunner implements LeaderContender, OnCompletionActions {
+
+	private final Logger log = LoggerFactory.getLogger(JobManagerRunner.class);
+
+	/** Lock to ensure that this runner can deal with leader election event and job completion notifies simultaneously */
+	private final Object lock = new Object();
+
+	/** The job graph needs to run */
+	private final JobGraph jobGraph;
+
+	private final OnCompletionActions toNotify;
+
+	/** The execution context which is used to execute futures */
+	private final Executor executionContext;
+
+	// TODO: use this to decide whether the job is finished by other
+	private final CheckpointRecoveryFactory checkpointRecoveryFactory;
+
+	/** Leader election for this job */
+	private final LeaderElectionService leaderElectionService;
+
+	private final JobMaster jobManager;
+
+	/** Leader session id when granted leadership */
+	private UUID leaderSessionID;
+
+	/** flag marking the runner as shut down */
+	private volatile boolean shutdown;
+
+	public JobManagerRunner(
+		final JobGraph jobGraph,
+		final Configuration configuration,
+		final RpcService rpcService,
+		final HighAvailabilityServices haServices,
+		final OnCompletionActions toNotify) throws Exception
+	{
+		this(jobGraph, configuration, rpcService, haServices,
+			JobManagerServices.fromConfiguration(configuration), toNotify);
+	}
+
+	public JobManagerRunner(
+		final JobGraph jobGraph,
+		final Configuration configuration,
+		final RpcService rpcService,
+		final HighAvailabilityServices haServices,
+		final JobManagerServices jobManagerServices,
+		final OnCompletionActions toNotify) throws Exception
+	{
+		this.jobGraph = jobGraph;
+		this.toNotify = toNotify;
+		this.executionContext = rpcService.getExecutor();
+		this.checkpointRecoveryFactory = haServices.getCheckpointRecoveryFactory();
+		this.leaderElectionService = haServices.getJobMasterLeaderElectionService(jobGraph.getJobID());
+		this.leaderSessionID = null;
+
+		this.jobManager = new JobMaster(
+			jobGraph, configuration, rpcService, haServices,
+			jobManagerServices.libraryCacheManager,
+			jobManagerServices.restartStrategyFactory,
+			jobManagerServices.savepointStore,
+			jobManagerServices.timeout,
+			new Scheduler(jobManagerServices.executorService),
+			jobManagerServices.jobManagerMetricGroup,
+			this);
+	}
+
+	//----------------------------------------------------------------------------------------------
+	// Lifecycle management
+	//----------------------------------------------------------------------------------------------
+
+	public void start() throws Exception {
+		jobManager.init();
+		jobManager.start();
+
+		try {
+			leaderElectionService.start(this);
+		}
+		catch (Exception e) {
+			log.error("Could not start the JobManager because the leader election service did not start.", e);
+			throw new Exception("Could not start the leader election service.", e);
+		}
+	}
+
+	public void shutdown() {
+		shutdown(new Exception("The JobManager runner is shutting down"));
+	}
+
+	public void shutdown(Throwable cause) {
+		// TODO what is the cause used for ?
+		shutdownInternally();
+	}
+
+	private void shutdownInternally() {
+		synchronized (lock) {
+			shutdown = true;
+
+			if (leaderElectionService != null) {
+				try {
+					leaderElectionService.stop();
+				} catch (Exception e) {
+					log.error("Could not properly shutdown the leader election service.");
+				}
+			}
+
+			jobManager.shutDown();
+		}
+	}
+
+	//----------------------------------------------------------------------------------------------
+	// Result and error handling methods
+	//----------------------------------------------------------------------------------------------
+
+	/**
+	 * Job completion notification triggered by JobManager
+	 */
+	@Override
+	public void jobFinished(JobExecutionResult result) {
+		try {
+			shutdownInternally();
+		}
+		finally {
+			if (toNotify != null) {
+				toNotify.jobFinished(result);
+			}
+		}
+	}
+
+	/**
+	 * Job completion notification triggered by JobManager
+	 */
+	@Override
+	public void jobFailed(Throwable cause) {
+		try {
+			shutdownInternally();
+		}
+		finally {
+			if (toNotify != null) {
+				toNotify.jobFailed(cause);
+			}
+		}
+	}
+
+	/**
+	 * Job completion notification triggered by self
+	 */
+	@Override
+	public void jobFinishedByOther() {
+		try {
+			shutdownInternally();
+		}
+		finally {
+			if (toNotify != null) {
+				toNotify.jobFinishedByOther();
+			}
+		}
+	}
+
+	/**
+	 * Job completion notification triggered by JobManager or self
+	 */
+	@Override
+	public void onFatalError(Throwable exception) {
+		// first and in any case, notify our handler, so it can react fast
+		try {
+			if (toNotify != null) {
+				toNotify.onFatalError(exception);
+			}
+		}
+		finally {
+			log.error("JobManager runner encountered a fatal error.", exception);
+			shutdownInternally();
+		}
+	}
+
+	//----------------------------------------------------------------------------------------------
+	// Leadership methods
+	//----------------------------------------------------------------------------------------------
+
+	@Override
+	public void grantLeadership(final UUID leaderSessionID) {
+		synchronized (lock) {
+			if (shutdown) {
+				log.info("JobManagerRunner already shutdown.");
+				return;
+			}
+
+			log.info("JobManager runner for job {} ({}) was granted leadership with session id {} at {}.",
+				jobGraph.getName(), jobGraph.getJobID(), leaderSessionID, getAddress());
+
+			// The operation may be blocking, but since this runner is idle before it been granted the leadership,
+			// it's okay that job manager wait for the operation complete
+			leaderElectionService.confirmLeaderSessionID(leaderSessionID);
+			this.leaderSessionID = leaderSessionID;
+
+			// Double check the leadership after we confirm that, there is a small chance that multiple
+			// job managers schedule the same job after if they try to recover at the same time.
+			// This will eventually be noticed, but can not be ruled out from the beginning.
+			if (leaderElectionService.hasLeadership()) {
+				if (isJobFinishedByOthers()) {
+					log.info("Job {} ({}) already finished by others.", jobGraph.getName(), jobGraph.getJobID());
+					jobFinishedByOther();
+				} else {
+					jobManager.getSelf().startJob();
+				}
+			}
+		}
+	}
+
+	@Override
+	public void revokeLeadership() {
+		synchronized (lock) {
+			if (shutdown) {
+				log.info("JobManagerRunner already shutdown.");
+				return;
+			}
+
+			log.info("JobManager for job {} ({}) was revoked leadership at {}.",
+				jobGraph.getName(), jobGraph.getJobID(), getAddress());
+
+			leaderSessionID = null;
+			jobManager.getSelf().suspendJob(new Exception("JobManager is no longer the leader."));
+		}
+	}
+
+	@Override
+	public String getAddress() {
+		return jobManager.getAddress();
+	}
+
+	@Override
+	public void handleError(Exception exception) {
+		log.error("Leader Election Service encountered a fatal error.", exception);
+		onFatalError(exception);
+	}
+
+	@VisibleForTesting
+	boolean isJobFinishedByOthers() {
+		// TODO
+		return false;
+	}
+
+	@VisibleForTesting
+	boolean isShutdown() {
+		return shutdown;
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerServices.java
new file mode 100644
index 0000000..e6beba6
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerServices.java
@@ -0,0 +1,73 @@
+/*
+ * 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.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore;
+import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
+import org.apache.flink.runtime.executiongraph.restart.RestartStrategyFactory;
+import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
+
+import java.util.concurrent.ExecutorService;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Utility class to hold all auxiliary services used by the {@link JobMaster}.
+ */
+public class JobManagerServices {
+
+	public final ExecutorService executorService;
+
+	public final BlobLibraryCacheManager libraryCacheManager;
+
+	public final RestartStrategyFactory restartStrategyFactory;
+
+	public final SavepointStore savepointStore;
+
+	public final Time timeout;
+
+	public final JobManagerMetricGroup jobManagerMetricGroup;
+
+	public JobManagerServices(
+			ExecutorService executorService,
+			BlobLibraryCacheManager libraryCacheManager,
+			RestartStrategyFactory restartStrategyFactory,
+			SavepointStore savepointStore,
+			Time timeout,
+			JobManagerMetricGroup jobManagerMetricGroup) {
+
+		this.executorService = checkNotNull(executorService);
+		this.libraryCacheManager = checkNotNull(libraryCacheManager);
+		this.restartStrategyFactory = checkNotNull(restartStrategyFactory);
+		this.savepointStore = checkNotNull(savepointStore);
+		this.timeout = checkNotNull(timeout);
+		this.jobManagerMetricGroup = checkNotNull(jobManagerMetricGroup);
+	}
+
+	// ------------------------------------------------------------------------
+	//  Creating the components from a configuration 
+	// ------------------------------------------------------------------------
+	
+	public static JobManagerServices fromConfiguration(Configuration config) throws Exception {
+		// TODO not yet implemented
+		return null;
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 1537396..b52a23c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -18,21 +18,50 @@
 
 package org.apache.flink.runtime.jobmaster;
 
-import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore;
+import org.apache.flink.runtime.checkpoint.stats.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.stats.DisabledCheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.stats.SimpleCheckpointStatsTracker;
+import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.client.JobSubmissionException;
+import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.restart.RestartStrategy;
+import org.apache.flink.runtime.executiongraph.restart.RestartStrategyFactory;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.leaderelection.LeaderContender;
-import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.jsonplan.JsonPlanGenerator;
+import org.apache.flink.runtime.jobgraph.tasks.JobSnapshottingSettings;
+import org.apache.flink.runtime.jobmanager.OnCompletionActions;
+import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
 import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.rpc.RpcMethod;
+import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
 import org.apache.flink.runtime.rpc.RpcEndpoint;
+import org.apache.flink.runtime.rpc.RpcMethod;
 import org.apache.flink.runtime.rpc.RpcService;
 import org.apache.flink.runtime.taskmanager.TaskExecutionState;
-import org.apache.flink.util.Preconditions;
+import scala.concurrent.ExecutionContext$;
+import scala.concurrent.duration.FiniteDuration;
 
-import java.util.UUID;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * JobMaster implementation. The job master is responsible for the execution of a single
@@ -41,7 +70,7 @@ import java.util.UUID;
  * It offers the following methods as part of its rpc interface to interact with the JobMaster
  * remotely:
  * <ul>
- *     <li>{@link #updateTaskExecutionState(TaskExecutionState)} updates the task execution state for
+ * <li>{@link #updateTaskExecutionState(TaskExecutionState)} updates the task execution state for
  * given task</li>
  * </ul>
  */
@@ -52,7 +81,6 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 
 	/** Logical representation of the job */
 	private final JobGraph jobGraph;
-	private final JobID jobID;
 
 	/** Configuration of the job */
 	private final Configuration configuration;
@@ -60,32 +88,67 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 	/** Service to contend for and retrieve the leadership of JM and RM */
 	private final HighAvailabilityServices highAvailabilityServices;
 
-	/** Leader Management */
-	private LeaderElectionService leaderElectionService = null;
-	private UUID leaderSessionID;
+	/** Blob cache manager used across jobs */
+	private final BlobLibraryCacheManager libraryCacheManager;
+
+	/** Factory to create restart strategy for this job */
+	private final RestartStrategyFactory restartStrategyFactory;
+
+	/** Store for save points */
+	private final SavepointStore savepointStore;
+
+	/** The timeout for this job */
+	private final Time timeout;
+
+	/** The scheduler to use for scheduling new tasks as they are needed */
+	private final Scheduler scheduler;
+
+	/** The metrics group used across jobs */
+	private final JobManagerMetricGroup jobManagerMetricGroup;
+
+	/** The execution context which is used to execute futures */
+	private final Executor executionContext;
+
+	private final OnCompletionActions jobCompletionActions;
+
+	/** The execution graph of this job */
+	private volatile ExecutionGraph executionGraph;
+
+	/** The checkpoint recovery factory used by this job */
+	private CheckpointRecoveryFactory checkpointRecoveryFactory;
+
+	private ClassLoader userCodeLoader;
+
+	private RestartStrategy restartStrategy;
+
+	private MetricGroup jobMetrics;
 
-	/**
-	 * The JM's Constructor
-	 *
-	 * @param jobGraph The representation of the job's execution plan
-	 * @param configuration The job's configuration
-	 * @param rpcService The RPC service at which the JM serves
-	 * @param highAvailabilityService The cluster's HA service from the JM can elect and retrieve leaders.
-	 */
 	public JobMaster(
 		JobGraph jobGraph,
 		Configuration configuration,
 		RpcService rpcService,
-		HighAvailabilityServices highAvailabilityService) {
-
+		HighAvailabilityServices highAvailabilityService,
+		BlobLibraryCacheManager libraryCacheManager,
+		RestartStrategyFactory restartStrategyFactory,
+		SavepointStore savepointStore,
+		Time timeout,
+		Scheduler scheduler,
+		JobManagerMetricGroup jobManagerMetricGroup,
+		OnCompletionActions jobCompletionActions)
+	{
 		super(rpcService);
 
-		this.jobGraph = Preconditions.checkNotNull(jobGraph);
-		this.jobID = Preconditions.checkNotNull(jobGraph.getJobID());
-
-		this.configuration = Preconditions.checkNotNull(configuration);
-
-		this.highAvailabilityServices = Preconditions.checkNotNull(highAvailabilityService);
+		this.jobGraph = checkNotNull(jobGraph);
+		this.configuration = checkNotNull(configuration);
+		this.highAvailabilityServices = checkNotNull(highAvailabilityService);
+		this.libraryCacheManager = checkNotNull(libraryCacheManager);
+		this.restartStrategyFactory = checkNotNull(restartStrategyFactory);
+		this.savepointStore = checkNotNull(savepointStore);
+		this.timeout = checkNotNull(timeout);
+		this.scheduler = checkNotNull(scheduler);
+		this.jobManagerMetricGroup = checkNotNull(jobManagerMetricGroup);
+		this.executionContext = checkNotNull(rpcService.getExecutor());
+		this.jobCompletionActions = checkNotNull(jobCompletionActions);
 	}
 
 	public ResourceManagerGateway getResourceManager() {
@@ -93,93 +156,294 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 	}
 
 	//----------------------------------------------------------------------------------------------
-	// Initialization methods
+	// Lifecycle management
 	//----------------------------------------------------------------------------------------------
+
+	/**
+	 * Initializing the job execution environment, should be called before start. Any error occurred during
+	 * initialization will be treated as job submission failure.
+	 *
+	 * @throws JobSubmissionException
+	 */
+	public void init() throws JobSubmissionException {
+		log.info("Initializing job {} ({}).", jobGraph.getName(), jobGraph.getJobID());
+
+		try {
+			// IMPORTANT: We need to make sure that the library registration is the first action,
+			// because this makes sure that the uploaded jar files are removed in case of
+			// unsuccessful
+			try {
+				libraryCacheManager.registerJob(jobGraph.getJobID(), jobGraph.getUserJarBlobKeys(),
+					jobGraph.getClasspaths());
+			} catch (Throwable t) {
+				throw new JobSubmissionException(jobGraph.getJobID(),
+					"Cannot set up the user code libraries: " + t.getMessage(), t);
+			}
+
+			userCodeLoader = libraryCacheManager.getClassLoader(jobGraph.getJobID());
+			if (userCodeLoader == null) {
+				throw new JobSubmissionException(jobGraph.getJobID(),
+					"The user code class loader could not be initialized.");
+			}
+
+			if (jobGraph.getNumberOfVertices() == 0) {
+				throw new JobSubmissionException(jobGraph.getJobID(), "The given job is empty");
+			}
+
+			final RestartStrategies.RestartStrategyConfiguration restartStrategyConfiguration =
+				jobGraph.getSerializedExecutionConfig()
+					.deserializeValue(userCodeLoader)
+					.getRestartStrategy();
+			if (restartStrategyConfiguration != null) {
+				restartStrategy = RestartStrategyFactory.createRestartStrategy(restartStrategyConfiguration);
+			} else {
+				restartStrategy = restartStrategyFactory.createRestartStrategy();
+			}
+
+			log.info("Using restart strategy {} for {} ({}).", restartStrategy, jobGraph.getName(), jobGraph.getJobID());
+
+			if (jobManagerMetricGroup != null) {
+				jobMetrics = jobManagerMetricGroup.addJob(jobGraph);
+			}
+			if (jobMetrics == null) {
+				jobMetrics = new UnregisteredMetricsGroup();
+			}
+
+			try {
+				checkpointRecoveryFactory = highAvailabilityServices.getCheckpointRecoveryFactory();
+			} catch (Exception e) {
+				log.error("Could not get the checkpoint recovery factory.", e);
+				throw new JobSubmissionException(jobGraph.getJobID(), "Could not get the checkpoint recovery factory.", e);
+			}
+
+		} catch (Throwable t) {
+			log.error("Failed to initializing job {} ({})", jobGraph.getName(), jobGraph.getJobID(), t);
+
+			libraryCacheManager.unregisterJob(jobGraph.getJobID());
+
+			if (t instanceof JobSubmissionException) {
+				throw (JobSubmissionException) t;
+			} else {
+				throw new JobSubmissionException(jobGraph.getJobID(), "Failed to initialize job " +
+					jobGraph.getName() + " (" + jobGraph.getJobID() + ")", t);
+			}
+		}
+	}
+
+	@Override
 	public void start() {
 		super.start();
-
-		// register at the election once the JM starts
-		registerAtElectionService();
 	}
 
+	@Override
+	public void shutDown() {
+		super.shutDown();
+
+		suspendJob(new Exception("JobManager is shutting down."));
+	}
 
 	//----------------------------------------------------------------------------------------------
-	// JobMaster Leadership methods
+	// RPC methods
 	//----------------------------------------------------------------------------------------------
 
 	/**
-	 * Retrieves the election service and contend for the leadership.
+	 * Start to run the job, runtime data structures like ExecutionGraph will be constructed now and checkpoint
+	 * being recovered. After this, we will begin to schedule the job.
 	 */
-	private void registerAtElectionService() {
-		try {
-			leaderElectionService = highAvailabilityServices.getJobMasterLeaderElectionService(jobID);
-			leaderElectionService.start(new JobMasterLeaderContender());
-		} catch (Exception e) {
-			throw new RuntimeException("Fail to register at the election of JobMaster", e);
+	@RpcMethod
+	public void startJob() {
+		log.info("Starting job {} ({}).", jobGraph.getName(), jobGraph.getJobID());
+
+		if (executionGraph != null) {
+			executionGraph = new ExecutionGraph(
+				ExecutionContext$.MODULE$.fromExecutor(executionContext),
+				jobGraph.getJobID(),
+				jobGraph.getName(),
+				jobGraph.getJobConfiguration(),
+				jobGraph.getSerializedExecutionConfig(),
+				new FiniteDuration(timeout.getSize(), timeout.getUnit()),
+				restartStrategy,
+				jobGraph.getUserJarBlobKeys(),
+				jobGraph.getClasspaths(),
+				userCodeLoader,
+				jobMetrics);
+		} else {
+			// TODO: update last active time in JobInfo
 		}
-	}
 
-	/**
-	 * Start the execution when the leadership is granted.
-	 *
-	 * @param newLeaderSessionID The identifier of the new leadership session
-	 */
-	public void grantJobMasterLeadership(final UUID newLeaderSessionID) {
-		runAsync(new Runnable() {
-			@Override
-			public void run() {
-				log.info("JobManager {} grants leadership with session id {}.", getAddress(), newLeaderSessionID);
+		try {
+			executionGraph.setScheduleMode(jobGraph.getScheduleMode());
+			executionGraph.setQueuedSchedulingAllowed(jobGraph.getAllowQueuedScheduling());
+
+			try {
+				executionGraph.setJsonPlan(JsonPlanGenerator.generatePlan(jobGraph));
+			} catch (Exception e) {
+				log.warn("Cannot create JSON plan for job {} ({})", jobGraph.getJobID(), jobGraph.getName(), e);
+				executionGraph.setJsonPlan("{}");
+			}
 
-				// The operation may be blocking, but since JM is idle before it grants the leadership, it's okay that
-				// JM waits here for the operation's completeness.
-				leaderSessionID = newLeaderSessionID;
-				leaderElectionService.confirmLeaderSessionID(newLeaderSessionID);
+			// initialize the vertices that have a master initialization hook
+			// file output formats create directories here, input formats create splits
+			if (log.isDebugEnabled()) {
+				log.debug("Running initialization on master for job {} ({}).", jobGraph.getJobID(), jobGraph.getName());
+			}
+			for (JobVertex vertex : jobGraph.getVertices()) {
+				final String executableClass = vertex.getInvokableClassName();
+				if (executableClass == null || executableClass.length() == 0) {
+					throw new JobExecutionException(jobGraph.getJobID(),
+						"The vertex " + vertex.getID() + " (" + vertex.getName() + ") has no invokable class.");
+				}
+				if (vertex.getParallelism() == ExecutionConfig.PARALLELISM_AUTO_MAX) {
+					vertex.setParallelism(scheduler.getTotalNumberOfSlots());
+				}
+
+				try {
+					vertex.initializeOnMaster(userCodeLoader);
+				} catch (Throwable t) {
+					throw new JobExecutionException(jobGraph.getJobID(),
+						"Cannot initialize task '" + vertex.getName() + "': " + t.getMessage(), t);
+				}
+			}
 
-				// TODO:: execute the job when the leadership is granted.
+			// topologically sort the job vertices and attach the graph to the existing one
+			final List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
+			if (log.isDebugEnabled()) {
+				log.debug("Adding {} vertices from job graph {} ({}).", sortedTopology.size(),
+					jobGraph.getJobID(), jobGraph.getName());
 			}
-		});
-	}
+			executionGraph.attachJobGraph(sortedTopology);
 
-	/**
-	 * Stop the execution when the leadership is revoked.
-	 */
-	public void revokeJobMasterLeadership() {
-		runAsync(new Runnable() {
-			@Override
-			public void run() {
-				log.info("JobManager {} was revoked leadership.", getAddress());
+			if (log.isDebugEnabled()) {
+				log.debug("Successfully created execution graph from job graph {} ({}).",
+					jobGraph.getJobID(), jobGraph.getName());
+			}
 
-				// TODO:: cancel the job's execution and notify all listeners
-				cancelAndClearEverything(new Exception("JobManager is no longer the leader."));
+			final JobSnapshottingSettings snapshotSettings = jobGraph.getSnapshotSettings();
+			if (snapshotSettings != null) {
+				List<ExecutionJobVertex> triggerVertices = getExecutionJobVertexWithId(
+					executionGraph, snapshotSettings.getVerticesToTrigger());
+
+				List<ExecutionJobVertex> ackVertices = getExecutionJobVertexWithId(
+					executionGraph, snapshotSettings.getVerticesToAcknowledge());
+
+				List<ExecutionJobVertex> confirmVertices = getExecutionJobVertexWithId(
+					executionGraph, snapshotSettings.getVerticesToConfirm());
+
+				CompletedCheckpointStore completedCheckpoints = checkpointRecoveryFactory.createCheckpointStore(
+					jobGraph.getJobID(), userCodeLoader);
+
+				CheckpointIDCounter checkpointIdCounter = checkpointRecoveryFactory.createCheckpointIDCounter(
+					jobGraph.getJobID());
+
+				// Checkpoint stats tracker
+				boolean isStatsDisabled = configuration.getBoolean(
+					ConfigConstants.JOB_MANAGER_WEB_CHECKPOINTS_DISABLE,
+					ConfigConstants.DEFAULT_JOB_MANAGER_WEB_CHECKPOINTS_DISABLE);
+
+				final CheckpointStatsTracker checkpointStatsTracker;
+				if (isStatsDisabled) {
+					checkpointStatsTracker = new DisabledCheckpointStatsTracker();
+				} else {
+					int historySize = configuration.getInteger(
+						ConfigConstants.JOB_MANAGER_WEB_CHECKPOINTS_HISTORY_SIZE,
+						ConfigConstants.DEFAULT_JOB_MANAGER_WEB_CHECKPOINTS_HISTORY_SIZE);
+					checkpointStatsTracker = new SimpleCheckpointStatsTracker(historySize, ackVertices, jobMetrics);
+				}
+
+				executionGraph.enableSnapshotCheckpointing(
+					snapshotSettings.getCheckpointInterval(),
+					snapshotSettings.getCheckpointTimeout(),
+					snapshotSettings.getMinPauseBetweenCheckpoints(),
+					snapshotSettings.getMaxConcurrentCheckpoints(),
+					triggerVertices,
+					ackVertices,
+					confirmVertices,
+					checkpointIdCounter,
+					completedCheckpoints,
+					savepointStore,
+					checkpointStatsTracker);
+			}
+
+			// TODO: register this class to execution graph as job status change listeners
+
+			// TODO: register client as job / execution status change listeners if they are interested
+
+			/*
+			TODO: decide whether we should take the savepoint before recovery
+
+			if (isRecovery) {
+				// this is a recovery of a master failure (this master takes over)
+				executionGraph.restoreLatestCheckpointedState();
+			} else {
+				if (snapshotSettings != null) {
+					String savepointPath = snapshotSettings.getSavepointPath();
+					if (savepointPath != null) {
+						// got a savepoint
+						log.info("Starting job from savepoint {}.", savepointPath);
+
+						// load the savepoint as a checkpoint into the system
+						final CompletedCheckpoint savepoint = SavepointLoader.loadAndValidateSavepoint(
+							jobGraph.getJobID(), executionGraph.getAllVertices(), savepointStore, savepointPath);
+						executionGraph.getCheckpointCoordinator().getCheckpointStore().addCheckpoint(savepoint);
+
+						// Reset the checkpoint ID counter
+						long nextCheckpointId = savepoint.getCheckpointID() + 1;
+						log.info("Reset the checkpoint ID to " + nextCheckpointId);
+						executionGraph.getCheckpointCoordinator().getCheckpointIdCounter().setCount(nextCheckpointId);
 
-				leaderSessionID = null;
+						executionGraph.restoreLatestCheckpointedState();
+					}
+				}
 			}
-		});
-	}
+			*/
 
-	/**
-	 * Handles error occurring in the leader election service
-	 *
-	 * @param exception Exception thrown in the leader election service
-	 */
-	public void onJobMasterElectionError(final Exception exception) {
-		runAsync(new Runnable() {
-			@Override
-			public void run() {
-				log.error("Received an error from the LeaderElectionService.", exception);
+		} catch (Throwable t) {
+			log.error("Failed to start job {} ({})", jobGraph.getName(), jobGraph.getJobID(), t);
 
-				// TODO:: cancel the job's execution and shutdown the JM
-				cancelAndClearEverything(exception);
+			executionGraph.fail(t);
+			executionGraph = null;
 
-				leaderSessionID = null;
+			final Throwable rt;
+			if (t instanceof JobExecutionException) {
+				rt = (JobExecutionException) t;
+			} else {
+				rt = new JobExecutionException(jobGraph.getJobID(),
+					"Failed to start job " + jobGraph.getJobID() + " (" + jobGraph.getName() + ")", t);
 			}
-		});
 
+			// TODO: notify client about this failure
+
+			jobCompletionActions.jobFailed(rt);
+			return;
+		}
+
+		// start scheduling job in another thread
+		executionContext.execute(new Runnable() {
+			@Override
+			public void run() {
+				if (executionGraph != null) {
+					try {
+						executionGraph.scheduleForExecution(scheduler);
+					} catch (Throwable t) {
+						executionGraph.fail(t);
+					}
+				}
+			}
+		});
 	}
 
-	//----------------------------------------------------------------------------------------------
-	// RPC methods
-	//----------------------------------------------------------------------------------------------
+	/**
+	 * Suspending job, all the running tasks will be cancelled, and runtime data will be cleared.
+	 *
+	 * @param cause The reason of why this job been suspended.
+	 */
+	@RpcMethod
+	public void suspendJob(final Throwable cause) {
+		if (executionGraph != null) {
+			executionGraph.suspend(cause);
+			executionGraph = null;
+		}
+	}
 
 	/**
 	 * Updates the task execution state for a given task.
@@ -208,37 +472,26 @@ public class JobMaster extends RpcEndpoint<JobMasterGateway> {
 	//----------------------------------------------------------------------------------------------
 
 	/**
-	 * Cancel the current job and notify all listeners the job's cancellation.
+	 * Converts JobVertexIDs to corresponding ExecutionJobVertexes
 	 *
-	 * @param cause Cause for the cancelling.
+	 * @param executionGraph The execution graph that holds the relationship
+	 * @param vertexIDs      The vertexIDs need to be converted
+	 * @return The corresponding ExecutionJobVertexes
+	 * @throws JobExecutionException
 	 */
-	private void cancelAndClearEverything(Throwable cause) {
-		// currently, nothing to do here
-	}
-
-	// ------------------------------------------------------------------------
-	//  Utility classes
-	// ------------------------------------------------------------------------
-	private class JobMasterLeaderContender implements LeaderContender {
-
-		@Override
-		public void grantLeadership(UUID leaderSessionID) {
-			JobMaster.this.grantJobMasterLeadership(leaderSessionID);
-		}
-
-		@Override
-		public void revokeLeadership() {
-			JobMaster.this.revokeJobMasterLeadership();
-		}
-
-		@Override
-		public String getAddress() {
-			return JobMaster.this.getAddress();
-		}
-
-		@Override
-		public void handleError(Exception exception) {
-			onJobMasterElectionError(exception);
+	private static List<ExecutionJobVertex> getExecutionJobVertexWithId(
+		final ExecutionGraph executionGraph, final List<JobVertexID> vertexIDs)
+		throws JobExecutionException
+	{
+		final List<ExecutionJobVertex> ret = new ArrayList<>(vertexIDs.size());
+		for (JobVertexID vertexID : vertexIDs) {
+			final ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(vertexID);
+			if (executionJobVertex == null) {
+				throw new JobExecutionException(executionGraph.getJobID(),
+					"The snapshot checkpointing settings refer to non-existent vertex " + vertexID);
+			}
+			ret.add(executionJobVertex);
 		}
+		return ret;
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
index 86bf17c..b281ea8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
@@ -29,6 +29,19 @@ import org.apache.flink.runtime.taskmanager.TaskExecutionState;
 public interface JobMasterGateway extends RpcGateway {
 
 	/**
+	 * Making this job begins to run.
+	 */
+	void startJob();
+
+	/**
+	 * Suspending job, all the running tasks will be cancelled, and runtime status will be cleared. Should re-submit
+	 * the job before restarting it.
+	 *
+	 * @param cause The reason of why this job been suspended.
+	 */
+	void suspendJob(final Throwable cause);
+
+	/**
 	 * Updates the task execution state for a given task.
 	 *
 	 * @param taskExecutionState New task execution state for a given task

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniClusterJobDispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniClusterJobDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniClusterJobDispatcher.java
new file mode 100644
index 0000000..792bfd5
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/MiniClusterJobDispatcher.java
@@ -0,0 +1,385 @@
+/*
+ * 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.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.client.JobExecutionException;
+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.rpc.RpcService;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * The dispatcher that runs in the mini cluster, waits for jobs, and starts job masters
+ * upon receiving jobs.
+ */
+public class MiniClusterJobDispatcher {
+
+	private static final Logger LOG = LoggerFactory.getLogger(MiniClusterJobDispatcher.class);
+
+	// ------------------------------------------------------------------------
+
+	/** lock to ensure that this dispatcher executes only one job at a time */
+	private final Object lock = new Object();
+
+	/** the configuration with which the mini cluster was started */
+	private final Configuration configuration;
+
+	/** the RPC service to use by the job managers */
+	private final RpcService rpcService;
+
+	/** services for discovery, leader election, and recovery */
+	private final HighAvailabilityServices haServices;
+
+	/** al the services that the JobManager needs, such as BLOB service, factories, etc */
+	private final JobManagerServices jobManagerServices;
+
+	/** The number of JobManagers to launch (more than one simulates a high-availability setup) */
+	private final int numJobManagers;
+
+	/** The runner for the job and master. non-null if a job is currently running */
+	private volatile JobManagerRunner[] runners;
+
+	/** flag marking the dispatcher as hut down */
+	private volatile boolean shutdown;
+
+
+	/**
+	 * Starts a mini cluster job dispatcher.
+	 * 
+	 * <p>The dispatcher kicks off one JobManager per job, a behavior similar to a
+	 * non-highly-available setup.
+	 * 
+	 * @param config The configuration of the mini cluster
+	 * @param haServices Access to the discovery, leader election, and recovery services
+	 * 
+	 * @throws Exception Thrown, if the services for the JobMaster could not be started.
+	 */
+	public MiniClusterJobDispatcher(
+			Configuration config,
+			RpcService rpcService,
+			HighAvailabilityServices haServices) throws Exception {
+		this(config, rpcService, haServices, 1);
+	}
+
+	/**
+	 * Starts a mini cluster job dispatcher.
+	 *
+	 * <p>The dispatcher may kick off more than one JobManager per job, thus simulating
+	 * a highly-available setup.
+	 * 
+	 * @param config The configuration of the mini cluster
+	 * @param haServices Access to the discovery, leader election, and recovery services
+	 * @param numJobManagers The number of JobMasters to start for each job.
+	 * 
+	 * @throws Exception Thrown, if the services for the JobMaster could not be started.
+	 */
+	public MiniClusterJobDispatcher(
+			Configuration config,
+			RpcService rpcService,
+			HighAvailabilityServices haServices,
+			int numJobManagers) throws Exception {
+
+		checkArgument(numJobManagers >= 1);
+		this.configuration = checkNotNull(config);
+		this.rpcService = checkNotNull(rpcService);
+		this.haServices = checkNotNull(haServices);
+		this.numJobManagers = numJobManagers;
+
+		LOG.info("Creating JobMaster services");
+		this.jobManagerServices = JobManagerServices.fromConfiguration(config);
+	}
+
+	// ------------------------------------------------------------------------
+	//  life cycle
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Shuts down the mini cluster dispatcher. If a job is currently running, that job will be
+	 * terminally failed.
+	 */
+	public void shutdown() {
+		synchronized (lock) {
+			if (!shutdown) {
+				shutdown = true;
+
+				LOG.info("Shutting down the dispatcher");
+
+				// in this shutdown code we copy the references to the stack first,
+				// to avoid concurrent modification
+
+				JobManagerRunner[] runners = this.runners;
+				if (runners != null) {
+					this.runners = null;
+
+					Exception shutdownException = new Exception("The MiniCluster is shutting down");
+					for (JobManagerRunner runner : runners) {
+						runner.shutdown(shutdownException);
+					}
+				}
+			}
+		}
+	}
+
+	// ------------------------------------------------------------------------
+	//  submitting jobs
+	// ------------------------------------------------------------------------
+
+	/**
+	 * This method executes a job in detached mode. The method returns immediately after the job
+	 * has been added to the
+	 *
+	 * @param job  The Flink job to execute
+	 *
+	 * @throws JobExecutionException Thrown if anything went amiss during initial job launch,
+	 *         or if the job terminally failed.
+	 */
+	public void runDetached(JobGraph job) throws JobExecutionException {
+		checkNotNull(job);
+
+		LOG.info("Received job for detached execution {} ({})", job.getName(), job.getJobID());
+
+		synchronized (lock) {
+			checkState(!shutdown, "mini cluster is shut down");
+			checkState(runners == null, "mini cluster can only execute one job at a time");
+
+			OnCompletionActions onJobCompletion = new DetachedFinalizer(numJobManagers);
+
+			this.runners = startJobRunners(job, onJobCompletion);
+		}
+	}
+
+	/**
+	 * This method runs a job in blocking mode. The method returns only after the job
+	 * completed successfully, or after it failed terminally.
+	 *
+	 * @param job  The Flink job to execute 
+	 * @return The result of the job execution
+	 *
+	 * @throws JobExecutionException Thrown if anything went amiss during initial job lauch,
+	 *         or if the job terminally failed.
+	 */
+	public JobExecutionResult runJobBlocking(JobGraph job) throws JobExecutionException, InterruptedException {
+		checkNotNull(job);
+		
+		LOG.info("Received job for blocking execution {} ({})", job.getName(), job.getJobID());
+		final BlockingJobSync onJobCompletion = new BlockingJobSync(job.getJobID(), numJobManagers);
+
+		synchronized (lock) {
+			checkState(!shutdown, "mini cluster is shut down");
+			checkState(runners == null, "mini cluster can only execute one job at a time");
+
+			this.runners = startJobRunners(job, onJobCompletion);
+		}
+
+		try {
+			return onJobCompletion.getResult();
+		}
+		finally {
+			// always clear the status for the next job
+			runners = null;
+		}
+	}
+
+	private JobManagerRunner[] startJobRunners(JobGraph job, OnCompletionActions onCompletion) throws JobExecutionException {
+		LOG.info("Starting {} JobMaster(s) for job {} ({})", numJobManagers, job.getName(), job.getJobID());
+
+		JobManagerRunner[] runners = new JobManagerRunner[numJobManagers];
+		for (int i = 0; i < numJobManagers; i++) {
+			try {
+				runners[i] = new JobManagerRunner(job, configuration,
+						rpcService, haServices, jobManagerServices, onCompletion);
+				runners[i].start();
+			}
+			catch (Throwable t) {
+				// shut down all the ones so far
+				Exception shutdownCause = new Exception("Failed to properly start all mini cluster JobManagers", t);
+
+				for (int k = 0; k <= i; k++) {
+					try {
+						if (runners[i] != null) {
+							runners[i].shutdown(shutdownCause);
+						}
+					} catch (Throwable ignored) {
+						// silent shutdown
+					}
+				}
+
+				throw new JobExecutionException(job.getJobID(), "Could not start the JobManager(s) for the job", t);
+			}
+		}
+
+		return runners;
+	}
+
+	// ------------------------------------------------------------------------
+	//  test methods to simulate job master failures
+	// ------------------------------------------------------------------------
+
+	public void killJobMaster(int which) {
+		checkArgument(which >= 0 && which < numJobManagers, "no such job master");
+		checkState(!shutdown, "mini cluster is shut down");
+
+		JobManagerRunner[] runners = this.runners;
+		checkState(runners != null, "mini cluster it not executing a job right now");
+
+		runners[which].shutdown(new Throwable("kill JobManager"));
+	}
+
+	// ------------------------------------------------------------------------
+	//  utility classes
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Simple class that waits for all runners to have reported that they are done.
+	 * In the case of a high-availability test setup, there may be multiple runners.
+	 * After that, it marks the mini cluster as ready to receive new jobs.
+	 */
+	private class DetachedFinalizer implements OnCompletionActions {
+
+		private final AtomicInteger numJobManagersToWaitFor;
+
+		private DetachedFinalizer(int numJobManagersToWaitFor) {
+			this.numJobManagersToWaitFor = new AtomicInteger(numJobManagersToWaitFor);
+		}
+
+		@Override
+		public void jobFinished(JobExecutionResult result) {
+			decrementCheckAndCleanup();
+		}
+
+		@Override
+		public void jobFailed(Throwable cause) {
+			decrementCheckAndCleanup();
+		}
+
+		@Override
+		public void jobFinishedByOther() {
+			decrementCheckAndCleanup();
+		}
+
+		@Override
+		public void onFatalError(Throwable exception) {
+			decrementCheckAndCleanup();
+		}
+
+		private void decrementCheckAndCleanup() {
+			if (numJobManagersToWaitFor.decrementAndGet() == 0) {
+				MiniClusterJobDispatcher.this.runners = null;
+			}
+		}
+	}
+
+	// ------------------------------------------------------------------------
+
+	/**
+	 * This class is used to sync on blocking jobs across multiple runners.
+	 * Only after all runners reported back that they are finished, the
+	 * result will be released.
+	 * 
+	 * That way it is guaranteed that after the blocking job submit call returns,
+	 * the dispatcher is immediately free to accept another job.
+	 */
+	private static class BlockingJobSync implements OnCompletionActions {
+
+		private final JobID jobId;
+
+		private final CountDownLatch jobMastersToWaitFor;
+
+		private volatile Throwable jobException;
+
+		private volatile Throwable runnerException;
+
+		private volatile JobExecutionResult result;
+		
+		BlockingJobSync(JobID jobId, int numJobMastersToWaitFor) {
+			this.jobId = jobId;
+			this.jobMastersToWaitFor = new CountDownLatch(numJobMastersToWaitFor);
+		}
+
+		@Override
+		public void jobFinished(JobExecutionResult jobResult) {
+			this.result = jobResult;
+			jobMastersToWaitFor.countDown();
+		}
+
+		@Override
+		public void jobFailed(Throwable cause) {
+			jobException = cause;
+			jobMastersToWaitFor.countDown();
+		}
+
+		@Override
+		public void jobFinishedByOther() {
+			this.jobMastersToWaitFor.countDown();
+		}
+
+		@Override
+		public void onFatalError(Throwable exception) {
+			if (runnerException == null) {
+				runnerException = exception;
+			}
+		}
+
+		public JobExecutionResult getResult() throws JobExecutionException, InterruptedException {
+			jobMastersToWaitFor.await();
+
+			final Throwable jobFailureCause = this.jobException;
+			final Throwable runnerException = this.runnerException;
+			final JobExecutionResult result = this.result;
+
+			// (1) we check if teh job terminated with an exception
+			// (2) we check whether the job completed successfully
+			// (3) we check if we have exceptions from the JobManagers. the job may still have
+			//     completed successfully in that case, if multiple JobMasters were running
+			//     and other took over. only if all encounter a fatal error, the job cannot finish
+
+			if (jobFailureCause != null) {
+				if (jobFailureCause instanceof JobExecutionException) {
+					throw (JobExecutionException) jobFailureCause;
+				}
+				else {
+					throw new JobExecutionException(jobId, "The job execution failed", jobFailureCause);
+				}
+			}
+			else if (result != null) {
+				return result;
+			}
+			else if (runnerException != null) {
+				throw new JobExecutionException(jobId,
+						"The job execution failed because all JobManagers encountered fatal errors", runnerException);
+			}
+			else {
+				throw new IllegalStateException("Bug: Job finished with neither error nor result.");
+			}
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/FatalErrorHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/FatalErrorHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/FatalErrorHandler.java
new file mode 100644
index 0000000..7721117
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/FatalErrorHandler.java
@@ -0,0 +1,24 @@
+/*
+ * 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.rpc;
+
+public interface FatalErrorHandler {
+
+	void onFatalError(Throwable exception);
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index cf709c8..9e3c3b9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.taskexecutor;
 import akka.actor.ActorSystem;
 import com.typesafe.config.Config;
 import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.io.network.ConnectionManager;
 import org.apache.flink.runtime.io.network.LocalConnectionManager;
@@ -28,6 +29,7 @@ import org.apache.flink.runtime.io.network.TaskEventDispatcher;
 import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
 import org.apache.flink.runtime.io.network.netty.NettyConnectionManager;
 import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
+import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
 import org.apache.flink.runtime.query.KvStateRegistry;
 import org.apache.flink.runtime.query.netty.DisabledKvStateRequestStats;
 import org.apache.flink.runtime.query.netty.KvStateServer;
@@ -340,6 +342,16 @@ public class TaskExecutor extends RpcEndpoint<TaskExecutorGateway> {
 			public LeaderElectionService getJobMasterLeaderElectionService(JobID jobID) throws Exception {
 				return null;
 			}
+
+			@Override
+			public CheckpointRecoveryFactory getCheckpointRecoveryFactory() throws Exception {
+				return null;
+			}
+
+			@Override
+			public SubmittedJobGraphStore getSubmittedJobGraphStore() throws Exception {
+				return null;
+			}
 		};
 
 		// start all the TaskManager services (network stack,  library cache, ...)

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java
index 2ac43be..1a5450d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java
@@ -19,10 +19,13 @@
 package org.apache.flink.runtime.highavailability;
 
 import org.apache.flink.api.common.JobID;
-import org.apache.flink.hadoop.shaded.org.jboss.netty.util.internal.ConcurrentHashMap;
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
+import org.apache.flink.runtime.jobmanager.SubmittedJobGraphStore;
 import org.apache.flink.runtime.leaderelection.LeaderElectionService;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
 
+import java.util.concurrent.ConcurrentHashMap;
+
 /**
  * A variant of the HighAvailabilityServices for testing. Each individual service can be set
  * to an arbitrary implementation, such as a mock or default service.
@@ -37,6 +40,9 @@ public class TestingHighAvailabilityServices implements HighAvailabilityServices
 
 	private volatile LeaderElectionService resourceManagerLeaderElectionService;
 
+	private volatile CheckpointRecoveryFactory checkpointRecoveryFactory;
+
+	private volatile SubmittedJobGraphStore submittedJobGraphStore;
 
 	// ------------------------------------------------------------------------
 	//  Setters for mock / testing implementations
@@ -58,6 +64,14 @@ public class TestingHighAvailabilityServices implements HighAvailabilityServices
 		this.resourceManagerLeaderElectionService = leaderElectionService;
 	}
 
+	public void setCheckpointRecoveryFactory(CheckpointRecoveryFactory checkpointRecoveryFactory) {
+		this.checkpointRecoveryFactory = checkpointRecoveryFactory;
+	}
+
+	public void setSubmittedJobGraphStore(SubmittedJobGraphStore submittedJobGraphStore) {
+		this.submittedJobGraphStore = submittedJobGraphStore;
+	}
+
 	// ------------------------------------------------------------------------
 	//  HA Services Methods
 	// ------------------------------------------------------------------------
@@ -103,4 +117,27 @@ public class TestingHighAvailabilityServices implements HighAvailabilityServices
 			throw new IllegalStateException("ResourceManagerLeaderElectionService has not been set");
 		}
 	}
+
+	@Override
+	public CheckpointRecoveryFactory getCheckpointRecoveryFactory() throws Exception {
+		CheckpointRecoveryFactory factory = checkpointRecoveryFactory;
+
+		if (factory != null) {
+			return factory;
+		} else {
+			throw new IllegalStateException("CheckpointRecoveryFactory has not been set");
+		}
+	}
+
+	@Override
+	public SubmittedJobGraphStore getSubmittedJobGraphStore() throws Exception {
+		SubmittedJobGraphStore store = submittedJobGraphStore;
+
+		if (store != null) {
+			return store;
+		} else {
+			throw new IllegalStateException("SubmittedJobGraphStore has not been set");
+
+		}
+	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerMockTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerMockTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerMockTest.java
new file mode 100644
index 0000000..dc3b5fd
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerMockTest.java
@@ -0,0 +1,254 @@
+/*
+ * 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.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.OnCompletionActions;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.rpc.RpcService;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.UUID;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(JobManagerRunner.class)
+public class JobManagerRunnerMockTest {
+
+	private JobManagerRunner runner;
+
+	private JobMaster jobManager;
+
+	private JobMasterGateway jobManagerGateway;
+
+	private LeaderElectionService leaderElectionService;
+
+	private TestingOnCompletionActions jobCompletion;
+
+	@Before
+	public void setUp() throws Exception {
+		jobManager = mock(JobMaster.class);
+		jobManagerGateway = mock(JobMasterGateway.class);
+		when(jobManager.getSelf()).thenReturn(jobManagerGateway);
+
+		PowerMockito.whenNew(JobMaster.class).withAnyArguments().thenReturn(jobManager);
+
+		jobCompletion = new TestingOnCompletionActions();
+
+		leaderElectionService = mock(LeaderElectionService.class);
+		when(leaderElectionService.hasLeadership()).thenReturn(true);
+
+		HighAvailabilityServices haServices = mock(HighAvailabilityServices.class);
+		when(haServices.getJobMasterLeaderElectionService(any(JobID.class))).thenReturn(leaderElectionService);
+
+		runner = PowerMockito.spy(new JobManagerRunner(
+			new JobGraph("test"),
+			mock(Configuration.class),
+			mock(RpcService.class),
+			haServices,
+			mock(JobManagerServices.class),
+			jobCompletion));
+	}
+
+	@After
+	public void tearDown() throws Exception {
+	}
+
+	@Test
+	public void testStartAndShutdown() throws Exception {
+		runner.start();
+		verify(jobManager).init();
+		verify(jobManager).start();
+		verify(leaderElectionService).start(runner);
+
+		assertTrue(!jobCompletion.isJobFinished());
+		assertTrue(!jobCompletion.isJobFailed());
+
+		runner.shutdown();
+		verify(leaderElectionService).stop();
+		verify(jobManager).shutDown();
+	}
+
+	@Test
+	public void testShutdownBeforeGrantLeadership() throws Exception {
+		runner.start();
+		verify(jobManager).init();
+		verify(jobManager).start();
+		verify(leaderElectionService).start(runner);
+
+		runner.shutdown();
+		verify(leaderElectionService).stop();
+		verify(jobManager).shutDown();
+
+		assertTrue(!jobCompletion.isJobFinished());
+		assertTrue(!jobCompletion.isJobFailed());
+
+		runner.grantLeadership(UUID.randomUUID());
+		assertTrue(!jobCompletion.isJobFinished());
+		assertTrue(!jobCompletion.isJobFailed());
+
+	}
+
+	@Test
+	public void testJobFinishedByOtherBeforeGrantLeadership() throws Exception {
+		runner.start();
+
+		when(runner.isJobFinishedByOthers()).thenReturn(true);
+		runner.grantLeadership(UUID.randomUUID());
+
+		// runner should shutdown automatic and informed the job completion
+		verify(leaderElectionService).stop();
+		verify(jobManager).shutDown();
+
+		assertTrue(jobCompletion.isJobFinished());
+		assertTrue(jobCompletion.isJobFinishedByOther());
+	}
+
+	@Test
+	public void testJobFinished() throws Exception {
+		runner.start();
+
+		runner.grantLeadership(UUID.randomUUID());
+		verify(jobManagerGateway).startJob();
+		assertTrue(!jobCompletion.isJobFinished());
+
+		// runner been told by JobManager that job is finished
+		runner.jobFinished(mock(JobExecutionResult.class));
+
+		assertTrue(jobCompletion.isJobFinished());
+		assertFalse(jobCompletion.isJobFinishedByOther());
+		verify(leaderElectionService).stop();
+		verify(jobManager).shutDown();
+		assertTrue(runner.isShutdown());
+	}
+
+	@Test
+	public void testJobFailed() throws Exception {
+		runner.start();
+
+		runner.grantLeadership(UUID.randomUUID());
+		verify(jobManagerGateway).startJob();
+		assertTrue(!jobCompletion.isJobFinished());
+
+		// runner been told by JobManager that job is failed
+		runner.jobFailed(new Exception("failed manually"));
+
+		assertTrue(jobCompletion.isJobFailed());
+		verify(leaderElectionService).stop();
+		verify(jobManager).shutDown();
+		assertTrue(runner.isShutdown());
+	}
+
+	@Test
+	public void testLeadershipRevoked() throws Exception {
+		runner.start();
+
+		runner.grantLeadership(UUID.randomUUID());
+		verify(jobManagerGateway).startJob();
+		assertTrue(!jobCompletion.isJobFinished());
+
+		runner.revokeLeadership();
+		verify(jobManagerGateway).suspendJob(any(Throwable.class));
+		assertFalse(runner.isShutdown());
+	}
+
+	@Test
+	public void testRegainLeadership() throws Exception {
+		runner.start();
+
+		runner.grantLeadership(UUID.randomUUID());
+		verify(jobManagerGateway).startJob();
+		assertTrue(!jobCompletion.isJobFinished());
+
+		runner.revokeLeadership();
+		verify(jobManagerGateway).suspendJob(any(Throwable.class));
+		assertFalse(runner.isShutdown());
+
+		runner.grantLeadership(UUID.randomUUID());
+		verify(jobManagerGateway, times(2)).startJob();
+	}
+
+	private static class TestingOnCompletionActions implements OnCompletionActions {
+
+		private volatile JobExecutionResult result;
+
+		private volatile Throwable failedCause;
+
+		private volatile boolean finishedByOther;
+
+		@Override
+		public void jobFinished(JobExecutionResult result) {
+			checkArgument(!isJobFinished(), "job finished already");
+			checkArgument(!isJobFailed(), "job failed already");
+
+			this.result = result;
+		}
+
+		@Override
+		public void jobFailed(Throwable cause) {
+			checkArgument(!isJobFinished(), "job finished already");
+			checkArgument(!isJobFailed(), "job failed already");
+
+			this.failedCause = cause;
+		}
+
+		@Override
+		public void jobFinishedByOther() {
+			checkArgument(!isJobFinished(), "job finished already");
+			checkArgument(!isJobFailed(), "job failed already");
+
+			this.finishedByOther = true;
+		}
+
+		@Override
+		public void onFatalError(Throwable exception) {
+			jobFailed(exception);
+		}
+
+		boolean isJobFinished() {
+			return result != null || finishedByOther;
+		}
+
+		boolean isJobFinishedByOther() {
+			return finishedByOther;
+		}
+
+		boolean isJobFailed() {
+			return failedCause != null;
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/474ace75/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java
index 6363662..e05c8d8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcConnectionTest.java
@@ -19,23 +19,21 @@
 package org.apache.flink.runtime.rpc;
 
 import akka.actor.ActorSystem;
-import akka.util.Timeout;
 
+import org.apache.flink.api.common.time.Time;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.concurrent.Future;
 import org.apache.flink.runtime.rpc.akka.AkkaRpcService;
 import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException;
 import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
 
-import org.junit.AfterClass;
 import org.junit.Test;
 
 import scala.Option;
 import scala.Tuple2;
-import scala.concurrent.Await;
-import scala.concurrent.Future;
-import scala.concurrent.duration.FiniteDuration;
 
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -57,19 +55,20 @@ public class RpcConnectionTest {
 
 			// we start the RPC service with a very long timeout to ensure that the test
 			// can only pass if the connection problem is not recognized merely via a timeout
-			rpcService = new AkkaRpcService(actorSystem, new Timeout(10000000, TimeUnit.SECONDS));
+			rpcService = new AkkaRpcService(actorSystem, Time.of(10000000, TimeUnit.SECONDS));
 
 			Future<TaskExecutorGateway> future = rpcService.connect("foo.bar.com.test.invalid", TaskExecutorGateway.class);
 
-			Await.result(future, new FiniteDuration(10000000, TimeUnit.SECONDS));
+			future.get(10000000, TimeUnit.SECONDS);
 			fail("should never complete normally");
 		}
 		catch (TimeoutException e) {
 			fail("should not fail with a generic timeout exception");
 		}
-		catch (RpcConnectionException e) {
+		catch (ExecutionException e) {
 			// that is what we want
-			assertTrue("wrong error message", e.getMessage().contains("foo.bar.com.test.invalid"));
+			assertTrue(e.getCause() instanceof RpcConnectionException);
+			assertTrue("wrong error message", e.getCause().getMessage().contains("foo.bar.com.test.invalid"));
 		}
 		catch (Throwable t) {
 			fail("wrong exception: " + t);


Mime
View raw message