Improve shutdown of JobManager (graceful exit of runing jobs and RPC service)

flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManager.java

@@ -40,7 +40,6 @@ import org.apache.commons.cli.Option;
 import org.apache.commons.cli.OptionBuilder;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
-
 import org.apache.flink.api.common.accumulators.Accumulator;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
@@ -61,6 +60,7 @@ import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
 import org.apache.flink.runtime.executiongraph.ExecutionGraph;
 import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
 import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.instance.Hardware;
 import org.apache.flink.runtime.instance.HardwareDescription;
 import org.apache.flink.runtime.instance.Instance;
 import org.apache.flink.runtime.instance.InstanceConnectionInfo;
@@ -92,7 +92,6 @@ import org.apache.flink.runtime.util.EnvironmentInformation;
 import org.apache.flink.runtime.util.ExecutorThreadFactory;
 import org.apache.flink.runtime.util.SerializableArrayList;
 import org.apache.flink.util.StringUtils;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -112,7 +111,7 @@ public class JobManager implements ExtendedManagementProtocol, InputSplitProvide
 	
 	
 	/** Executor service for asynchronous commands (to relieve the RPC threads of work) */
-	private final ExecutorService executorService = Executors.newCachedThreadPool(ExecutorThreadFactory.INSTANCE);
+	private final ExecutorService executorService = Executors.newFixedThreadPool(2 * Hardware.getNumberCPUCores(), ExecutorThreadFactory.INSTANCE);
 	
 
 	/** The RPC end point through which the JobManager gets its calls */
@@ -231,17 +230,12 @@ public class JobManager implements ExtendedManagementProtocol, InputSplitProvide
 			return;
 		}
 		
-		// Stop instance manager
-		if (this.instanceManager != null) {
-			this.instanceManager.shutdown();
-		}
-
-		// Stop RPC server
-		if (this.jobManagerServer != null) {
-			this.jobManagerServer.stop();
+		for (ExecutionGraph e : this.currentJobs.values()) {
+			e.fail(new Exception("The JobManager is shutting down."));
 		}
 		
 		// Stop the executor service
+		// this waits for any pending calls to be done
 		if (this.executorService != null) {
 			this.executorService.shutdown();
 			try {
@@ -251,6 +245,16 @@ public class JobManager implements ExtendedManagementProtocol, InputSplitProvide
 			}
 		}
 
+		// Stop instance manager
+		if (this.instanceManager != null) {
+			this.instanceManager.shutdown();
+		}
+
+		// Stop RPC server
+		if (this.jobManagerServer != null) {
+			this.jobManagerServer.stop();
+		}
+
 		// Stop and clean up the job progress collector
 		if (this.eventCollector != null) {
 			this.eventCollector.shutdown();