Till's comments.

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

@@ -20,8 +20,8 @@
 
 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 org.apache.flink.util.Preconditions;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -103,6 +103,46 @@ public String toString() {
 	//  Factory
 	// ------------------------------------------------------------------------
 
+	/**
+	 * calcuate cutoff memory size used by container, it will throw an {@link IllegalArgumentException}
+	 * if the config is invalid or return the cutoff value if valid.
+	 *
+	 * @param config The Flink configuration.
+	 * @param containerMemoryMB The size of the complete container, in megabytes.
+	 *
+	 * @return cutoff memory 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;
+	}
+
 	/**
 	 * Computes the parameters to be used to start a TaskManager Java process.
 	 *
@@ -114,7 +154,7 @@ public static ContaineredTaskManagerParameters create(
 		Configuration config, long containerMemoryMB, int numSlots)
 	{
 		// (1) try to compute how much memory used by container
-		final long cutoffMB = ResourceManagerRuntimeServices.calculateCutoffMB(config, containerMemoryMB);
+		final long cutoffMB = calculateCutoffMB(config, containerMemoryMB);
 
 		// (2) split the remaining Java memory between heap and off-heap
 		final long heapSizeMB = TaskManagerServices.calculateHeapSizeMB(containerMemoryMB - cutoffMB, config);

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

@@ -18,8 +18,6 @@
 
 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;
@@ -55,46 +53,6 @@ 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,7 +20,6 @@
 
 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/clusterframework/ContaineredTaskManagerParametersTest.java

@@ -20,13 +20,15 @@
 
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ResourceManagerOptions;
 import org.apache.flink.util.TestLogger;
 import org.junit.Test;
 
 import static org.apache.flink.configuration.TaskManagerOptions.MEMORY_OFF_HEAP;
 import static org.apache.flink.runtime.taskexecutor.TaskManagerServices.calculateNetworkBufferMemory;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 public class ContaineredTaskManagerParametersTest extends TestLogger {
 	private static final long CONTAINER_MEMORY = 8192L;
@@ -91,4 +93,34 @@ public void testTotalMemoryDoesNotExceedContainerMemoryOffHeap() {
 		assertTrue(params.taskManagerHeapSizeMB() +
 			params.taskManagerDirectMemoryLimitMB() <= CONTAINER_MEMORY);
 	}
+
+	/**
+	 * Test to guard {@link ContaineredTaskManagerParameters#calculateCutoffMB(Configuration, long)}.
+	 */
+	@Test
+	public void testCalculateCutoffMB() 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);
+
+		assertEquals(128,
+			ContaineredTaskManagerParameters.calculateCutoffMB(config, containerMemoryMB));
+
+		config.setFloat(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO, 0.2f);
+		assertEquals(200,
+			ContaineredTaskManagerParameters.calculateCutoffMB(config, containerMemoryMB));
+
+		config.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 1000);
+
+		try {
+			ContaineredTaskManagerParameters.calculateCutoffMB(config, containerMemoryMB);
+		} catch (IllegalArgumentException expected) {
+			// we expected it.
+			return;
+		}
+		fail();
+	}
 }

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

@@ -36,10 +36,10 @@
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.runtime.akka.AkkaUtils;
 import org.apache.flink.runtime.clusterframework.BootstrapTools;
+import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters;
 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;
@@ -411,10 +411,26 @@ public void terminateCluster(ApplicationId applicationId) throws FlinkException
 		}
 	}
 
-	private void checkConfig(ClusterSpecification clusterSpecification) {
+	/**
+	 * Method to validate cluster specification before deploy it, it will throw
+	 * an {@link IllegalConfigurationException} if the {@link ClusterSpecification} is invalid.
+	 */
+	private void validateClusterSpecification(ClusterSpecification clusterSpecification) {
 		long taskManagerMemorySize = clusterSpecification.getTaskManagerMemoryMB();
-		long cutoff = ResourceManagerRuntimeServices.calculateCutoffMB(flinkConfiguration, taskManagerMemorySize);
-		TaskManagerServices.calculateHeapSizeMB(taskManagerMemorySize - cutoff, flinkConfiguration);
+		long cutoff;
+		try {
+			// We do the validation by calling the calculation methods here
+			cutoff = ContaineredTaskManagerParameters.calculateCutoffMB(flinkConfiguration, taskManagerMemorySize);
+		} catch (IllegalArgumentException cutoffConfigurationInvalidEx) {
+			throw new IllegalConfigurationException("Configurations related to cutoff checked failed.", cutoffConfigurationInvalidEx);
+		}
+
+		try {
+			// We do the validation by calling the calculation methods here
+			TaskManagerServices.calculateHeapSizeMB(taskManagerMemorySize - cutoff, flinkConfiguration);
+		} catch (IllegalArgumentException heapSizeConfigurationInvalidEx) {
+			throw new IllegalConfigurationException("Configurations related to heap size checked failed.", heapSizeConfigurationInvalidEx);
+		}
 	}
 
 	/**
@@ -432,7 +448,7 @@ protected ClusterClient<ApplicationId> deployInternal(
 			boolean detached) throws Exception {
 
 		// ------------------ Check if configuration is valid --------------------
-		checkConfig(clusterSpecification);
+		validateClusterSpecification(clusterSpecification);
 
 		if (UserGroupInformation.isSecurityEnabled()) {
 			// note: UGI::hasKerberosCredentials inaccurately reports false

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

@@ -31,7 +31,6 @@
 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;