[FLINK-9028][flip6] perform parameters checking before starting cluster

flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java

@@ -20,6 +20,7 @@
 
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ResourceManagerOptions;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServices;
 import org.apache.flink.runtime.taskexecutor.TaskManagerServices;
 
 import java.util.HashMap;
@@ -101,7 +102,7 @@ public String toString() {
 	// ------------------------------------------------------------------------
 	//  Factory
 	// ------------------------------------------------------------------------
-	
+
 	/**
 	 * Computes the parameters to be used to start a TaskManager Java process.
 	 *
@@ -112,35 +113,11 @@ public String toString() {
 	public static ContaineredTaskManagerParameters create(
 		Configuration config, long containerMemoryMB, int numSlots)
 	{
-		// (1) compute how much memory we subtract from the total memory, to get the Java memory
-
-		final float memoryCutoffRatio = config.getFloat(
-			ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO);
-
-		final int minCutoff = config.getInteger(
-			ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN);
-
-		if (memoryCutoffRatio >= 1 || memoryCutoffRatio <= 0) {
-			throw new IllegalArgumentException("The configuration value '"
-				+ ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key() + "' must be between 0 and 1. Value given="
-				+ memoryCutoffRatio);
-		}
-
-		if (minCutoff >= containerMemoryMB) {
-			throw new IllegalArgumentException("The configuration value '"
-				+ ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN.key() + "'='" + minCutoff
-				+ "' is larger than the total container memory " + containerMemoryMB);
-		}
-
-		long cutoff = (long) (containerMemoryMB * memoryCutoffRatio);
-		if (cutoff < minCutoff) {
-			cutoff = minCutoff;
-		}
-
-		final long javaMemorySizeMB = containerMemoryMB - cutoff;
+		// (1) try to compute how much memory used by container
+		final long cutoffMB = ResourceManagerRuntimeServices.calculateCutoffMB(config, containerMemoryMB);
 
 		// (2) split the remaining Java memory between heap and off-heap
-		final long heapSizeMB = TaskManagerServices.calculateHeapSizeMB(javaMemorySizeMB, config);
+		final long heapSizeMB = TaskManagerServices.calculateHeapSizeMB(containerMemoryMB - cutoffMB, config);
 		// use the cut-off memory for off-heap (that was its intention)
 		final long offHeapSizeMB = containerMemoryMB - heapSizeMB;
 

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

@@ -313,7 +313,7 @@ public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout)
 	@Override
 	public CompletableFuture<Collection<JobID>> listJobs(Time timeout) {
 		if (jobManagerRunners.isEmpty()) {
-			System.out.println("empty");
+			log.info("empty");
 		}
 		return CompletableFuture.completedFuture(
 			Collections.unmodifiableSet(new HashSet<>(jobManagerRunners.keySet())));

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRuntimeServices.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.runtime.resourcemanager;
 
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ResourceManagerOptions;
 import org.apache.flink.runtime.concurrent.ScheduledExecutor;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
@@ -53,6 +55,46 @@ public void shutDown() throws Exception {
 
 	// -------------------- Static methods --------------------------------------
 
+	/**
+	 * Check whether the config is valid, it will throw an exception if the config
+	 * is invalid or return the cut off value.
+	 *
+	 * @param config The Flink configuration.
+	 * @param containerMemoryMB The size of the complete container, in megabytes.
+	 *
+	 * @return cut off size used by container.
+	 */
+	public static long calculateCutoffMB(Configuration config, long containerMemoryMB) {
+		Preconditions.checkArgument(containerMemoryMB > 0);
+
+		// (1) check cutoff ratio
+		final float memoryCutoffRatio = config.getFloat(
+			ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO);
+
+		if (memoryCutoffRatio >= 1 || memoryCutoffRatio <= 0) {
+			throw new IllegalArgumentException("The configuration value '"
+				+ ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key() + "' must be between 0 and 1. Value given="
+				+ memoryCutoffRatio);
+		}
+
+		// (2) check min cutoff value
+		final int minCutoff = config.getInteger(
+			ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN);
+
+		if (minCutoff >= containerMemoryMB) {
+			throw new IllegalArgumentException("The configuration value '"
+				+ ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN.key() + "'='" + minCutoff
+				+ "' is larger than the total container memory " + containerMemoryMB);
+		}
+
+		// (3) check between heap and off-heap
+		long cutoff = (long) (containerMemoryMB * memoryCutoffRatio);
+		if (cutoff < minCutoff) {
+			cutoff = minCutoff;
+		}
+		return cutoff;
+	}
+
 	public static ResourceManagerRuntimeServices fromConfiguration(
 			ResourceManagerRuntimeServicesConfiguration configuration,
 			HighAvailabilityServices highAvailabilityServices,

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java

@@ -20,6 +20,7 @@
 
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ResourceManagerOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.core.memory.MemoryType;
 import org.apache.flink.queryablestate.network.stats.DisabledKvStateRequestStats;

flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRuntimeServicesTest.java

@@ -0,0 +1,58 @@
+/*
+ * 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.resourcemanager;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ResourceManagerOptions;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Tests to guard {@link ResourceManagerRuntimeServices}.
+ */
+public class ResourceManagerRuntimeServicesTest {
+
+	/**
+	 * Test to guard {@link ResourceManagerRuntimeServices#calculateCutoffMB(Configuration, long)}.
+	 */
+	@Test
+	public void calculateCutoffMB() throws Exception {
+
+		Configuration config = new Configuration();
+		long containerMemoryMB = 1000;
+
+		config.setFloat(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO, 0.1f);
+		config.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 128);
+
+		Assert.assertEquals(128, ResourceManagerRuntimeServices.calculateCutoffMB(config, containerMemoryMB));
+
+		config.setFloat(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO, 0.2f);
+		Assert.assertEquals(200, ResourceManagerRuntimeServices.calculateCutoffMB(config, containerMemoryMB));
+
+		config.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 1000);
+
+		try {
+			ResourceManagerRuntimeServices.calculateCutoffMB(config, containerMemoryMB);
+		} catch (Exception expected) {
+			// we expected it.
+			return;
+		}
+		Assert.fail();
+	}
+}

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java

@@ -214,5 +214,4 @@ public void calculateHeapSizeMB() throws Exception {
 		config.setFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION, 0.1f); // 10%
 		assertEquals(810, TaskManagerServices.calculateHeapSizeMB(1000, config));
 	}
-
 }

flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java

@@ -39,6 +39,8 @@
 import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServices;
+import org.apache.flink.runtime.taskexecutor.TaskManagerServices;
 import org.apache.flink.util.FlinkException;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.ShutdownHookUtil;
@@ -409,6 +411,12 @@ public void terminateCluster(ApplicationId applicationId) throws FlinkException
 		}
 	}
 
+	private void checkConfig(ClusterSpecification clusterSpecification) {
+		long taskManagerMemorySize = clusterSpecification.getTaskManagerMemoryMB();
+		long cutoff = ResourceManagerRuntimeServices.calculateCutoffMB(flinkConfiguration, taskManagerMemorySize);
+		TaskManagerServices.calculateHeapSizeMB(taskManagerMemorySize - cutoff, flinkConfiguration);
+	}
+
 	/**
 	 * This method will block until the ApplicationMaster/JobManager have been deployed on YARN.
 	 *
@@ -423,6 +431,9 @@ protected ClusterClient<ApplicationId> deployInternal(
 			@Nullable JobGraph jobGraph,
 			boolean detached) throws Exception {
 
+		// ------------------ Check if configuration is valid --------------------
+		checkConfig(clusterSpecification);
+
 		if (UserGroupInformation.isSecurityEnabled()) {
 			// note: UGI::hasKerberosCredentials inaccurately reports false
 			// for logins based on a keytab (fixed in Hadoop 2.6.1, see HADOOP-10786),

flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java

@@ -31,6 +31,7 @@
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters;
 import org.apache.flink.runtime.clusterframework.messages.GetClusterStatusResponse;
 import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter;
 import org.apache.flink.runtime.security.SecurityConfiguration;
@@ -631,7 +632,7 @@ public int run(String[] args) throws CliArgsException, FlinkException {
 				if (detachedMode) {
 					LOG.info("The Flink YARN client has been started in detached mode. In order to stop " +
 						"Flink on YARN, use the following command or a YARN web interface to stop it:\n" +
-						"yarn application -kill " + applicationId.getOpt());
+						"yarn application -kill " + yarnApplicationId);
 				} else {
 					ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
 

flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java

@@ -264,8 +264,8 @@ public void testCommandLineClusterSpecification() throws Exception {
 		configuration.setInteger(TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY, 7331);
 		configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 2);
 
-		final int jobManagerMemory = 42;
-		final int taskManagerMemory = 41;
+		final int jobManagerMemory = 1337;
+		final int taskManagerMemory = 7331;
 		final int slotsPerTaskManager = 30;
 		final String[] args = {"-yjm", String.valueOf(jobManagerMemory), "-ytm", String.valueOf(taskManagerMemory), "-ys", String.valueOf(slotsPerTaskManager)};
 		final FlinkYarnSessionCli flinkYarnSessionCli = new FlinkYarnSessionCli(