You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/03/21 09:57:18 UTC
[1/3] flink git commit: [FLINK-9028] [yarn] Improve failure message
if cluster cannot be started
Repository: flink
Updated Branches:
refs/heads/master 328f72d14 -> f9df13c50
[FLINK-9028] [yarn] Improve failure message if cluster cannot be started
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7c952dd3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7c952dd3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7c952dd3
Branch: refs/heads/master
Commit: 7c952dd3a75bc64d10bf9be12e405bbc349422b1
Parents: 38aa863
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Mar 20 15:38:02 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 21 10:56:54 2018 +0100
----------------------------------------------------------------------
.../ContaineredTaskManagerParameters.java | 28 ++++++++++----------
.../ContaineredTaskManagerParametersTest.java | 9 +++----
.../taskexecutor/TaskManagerServicesTest.java | 2 +-
.../yarn/AbstractYarnClusterDescriptor.java | 26 +++++++++---------
.../flink/yarn/cli/FlinkYarnSessionCli.java | 1 -
.../flink/yarn/FlinkYarnSessionCliTest.java | 9 ++++---
6 files changed, 37 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7c952dd3/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java
index fa7fdf4..a4e7d25 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java
@@ -32,23 +32,22 @@ import java.util.Map;
public class ContaineredTaskManagerParameters implements java.io.Serializable {
private static final long serialVersionUID = -3096987654278064670L;
-
- /** Total container memory, in bytes */
+
+ /** Total container memory, in bytes. */
private final long totalContainerMemoryMB;
- /** Heap size to be used for the Java process */
+ /** Heap size to be used for the Java process. */
private final long taskManagerHeapSizeMB;
- /** Direct memory limit for the Java process */
+ /** Direct memory limit for the Java process. */
private final long taskManagerDirectMemoryLimitMB;
- /** The number of slots per TaskManager */
+ /** The number of slots per TaskManager. */
private final int numSlots;
-
- /** Environment variables to add to the Java process */
+
+ /** Environment variables to add to the Java process. */
private final HashMap<String, String> taskManagerEnv;
-
public ContaineredTaskManagerParameters(
long totalContainerMemoryMB,
long taskManagerHeapSizeMB,
@@ -62,7 +61,7 @@ public class ContaineredTaskManagerParameters implements java.io.Serializable {
this.numSlots = numSlots;
this.taskManagerEnv = taskManagerEnv;
}
-
+
// ------------------------------------------------------------------------
public long taskManagerTotalMemoryMB() {
@@ -87,7 +86,7 @@ public class ContaineredTaskManagerParameters implements java.io.Serializable {
// ------------------------------------------------------------------------
-
+
@Override
public String toString() {
return "TaskManagerParameters {" +
@@ -104,7 +103,7 @@ public class ContaineredTaskManagerParameters implements java.io.Serializable {
// ------------------------------------------------------------------------
/**
- * calcuate cutoff memory size used by container, it will throw an {@link IllegalArgumentException}
+ * 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.
@@ -151,8 +150,9 @@ public class ContaineredTaskManagerParameters implements java.io.Serializable {
* @return The parameters to start the TaskManager processes with.
*/
public static ContaineredTaskManagerParameters create(
- Configuration config, long containerMemoryMB, int numSlots)
- {
+ Configuration config,
+ long containerMemoryMB,
+ int numSlots) {
// (1) try to compute how much memory used by container
final long cutoffMB = calculateCutoffMB(config, containerMemoryMB);
@@ -164,7 +164,7 @@ public class ContaineredTaskManagerParameters implements java.io.Serializable {
// (3) obtain the additional environment variables from the configuration
final HashMap<String, String> envVars = new HashMap<>();
final String prefix = ResourceManagerOptions.CONTAINERIZED_TASK_MANAGER_ENV_PREFIX;
-
+
for (String key : config.keySet()) {
if (key.startsWith(prefix) && key.length() > prefix.length()) {
// remove prefix
http://git-wip-us.apache.org/repos/asf/flink/blob/7c952dd3/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
index 230a934..8537d17 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
@@ -98,10 +98,10 @@ public class ContaineredTaskManagerParametersTest extends TestLogger {
* Test to guard {@link ContaineredTaskManagerParameters#calculateCutoffMB(Configuration, long)}.
*/
@Test
- public void testCalculateCutoffMB() throws Exception {
+ public void testCalculateCutoffMB() {
Configuration config = new Configuration();
- long containerMemoryMB = 1000;
+ long containerMemoryMB = 1000L;
config.setFloat(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO, 0.1f);
config.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 128);
@@ -117,10 +117,9 @@ public class ContaineredTaskManagerParametersTest extends TestLogger {
try {
ContaineredTaskManagerParameters.calculateCutoffMB(config, containerMemoryMB);
- } catch (IllegalArgumentException expected) {
+ fail("Expected to fail with an invalid argument exception.");
+ } catch (IllegalArgumentException ignored) {
// we expected it.
- return;
}
- fail();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7c952dd3/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java
index d3d5444..f6e7b07 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java
@@ -44,7 +44,7 @@ public class TaskManagerServicesTest extends TestLogger {
*/
@SuppressWarnings("deprecation")
@Test
- public void calculateNetworkBufOld() throws Exception {
+ public void calculateNetworkBufOld() {
Configuration config = new Configuration();
config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, 1);
http://git-wip-us.apache.org/repos/asf/flink/blob/7c952dd3/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index eab5e39..caf7a76 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -413,23 +413,23 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
/**
* Method to validate cluster specification before deploy it, it will throw
- * an {@link IllegalConfigurationException} if the {@link ClusterSpecification} is invalid.
+ * an {@link FlinkException} if the {@link ClusterSpecification} is invalid.
+ *
+ * @param clusterSpecification cluster specification to check against the configuration of the
+ * AbstractYarnClusterDescriptor
+ * @throws FlinkException if the cluster cannot be started with the provided {@link ClusterSpecification}
*/
- private void validateClusterSpecification(ClusterSpecification clusterSpecification) {
- long taskManagerMemorySize = clusterSpecification.getTaskManagerMemoryMB();
- 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);
- }
-
+ private void validateClusterSpecification(ClusterSpecification clusterSpecification) throws FlinkException {
try {
+ final long taskManagerMemorySize = clusterSpecification.getTaskManagerMemoryMB();
// We do the validation by calling the calculation methods here
+ // Internally these methods will check whether the cluster can be started with the provided
+ // ClusterSpecification and the configured memory requirements
+ final long cutoff = ContaineredTaskManagerParameters.calculateCutoffMB(flinkConfiguration, taskManagerMemorySize);
TaskManagerServices.calculateHeapSizeMB(taskManagerMemorySize - cutoff, flinkConfiguration);
- } catch (IllegalArgumentException heapSizeConfigurationInvalidEx) {
- throw new IllegalConfigurationException("Configurations related to heap size checked failed.", heapSizeConfigurationInvalidEx);
+ } catch (IllegalArgumentException iae) {
+ throw new FlinkException("Cannot fulfill the minimum memory requirements with the provided " +
+ "cluster specification. Please increase the memory of the cluster.", iae);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7c952dd3/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
index 1443f99..2311e87 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -31,7 +31,6 @@ import org.apache.flink.configuration.GlobalConfiguration;
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;
http://git-wip-us.apache.org/repos/asf/flink/blob/7c952dd3/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
index 5b0d422..62110ed 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
@@ -260,13 +260,14 @@ public class FlinkYarnSessionCliTest extends TestLogger {
@Test
public void testCommandLineClusterSpecification() throws Exception {
final Configuration configuration = new Configuration();
- configuration.setInteger(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, 1337);
- configuration.setInteger(TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY, 7331);
- configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 2);
-
final int jobManagerMemory = 1337;
final int taskManagerMemory = 7331;
final int slotsPerTaskManager = 30;
+
+ configuration.setInteger(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, jobManagerMemory);
+ configuration.setInteger(TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY, taskManagerMemory);
+ configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, slotsPerTaskManager);
+
final String[] args = {"-yjm", String.valueOf(jobManagerMemory), "-ytm", String.valueOf(taskManagerMemory), "-ys", String.valueOf(slotsPerTaskManager)};
final FlinkYarnSessionCli flinkYarnSessionCli = new FlinkYarnSessionCli(
configuration,
[2/3] flink git commit: [FLINK-9028] [yarn] Perform parameters
checking before Yarn starting cluster
Posted by tr...@apache.org.
[FLINK-9028] [yarn] Perform parameters checking before Yarn starting cluster
This closes #5726.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/38aa863d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/38aa863d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/38aa863d
Branch: refs/heads/master
Commit: 38aa863d5a710b283b5c9b2eb9225d6fb9cc0c70
Parents: 328f72d
Author: sihuazhou <su...@163.com>
Authored: Tue Mar 20 15:59:33 2018 +0800
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 21 10:56:54 2018 +0100
----------------------------------------------------------------------
.../ContaineredTaskManagerParameters.java | 41 ++++++++++++++------
.../flink/runtime/dispatcher/Dispatcher.java | 2 +-
.../ContaineredTaskManagerParametersTest.java | 32 +++++++++++++++
.../taskexecutor/TaskManagerServicesTest.java | 1 -
.../yarn/AbstractYarnClusterDescriptor.java | 27 +++++++++++++
.../flink/yarn/cli/FlinkYarnSessionCli.java | 3 +-
.../flink/yarn/FlinkYarnSessionCliTest.java | 4 +-
.../flink/yarn/YarnClusterDescriptorTest.java | 15 ++++---
8 files changed, 103 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/38aa863d/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java
index c4dd486..fa7fdf4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.clusterframework;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.runtime.taskexecutor.TaskManagerServices;
+import org.apache.flink.util.Preconditions;
import java.util.HashMap;
import java.util.Map;
@@ -101,46 +102,62 @@ public class ContaineredTaskManagerParameters implements java.io.Serializable {
// ------------------------------------------------------------------------
// Factory
// ------------------------------------------------------------------------
-
+
/**
- * Computes the parameters to be used to start a TaskManager Java process.
+ * 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 The parameters to start the TaskManager processes with.
+ *
+ * @return cutoff memory size used by container.
*/
- 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
+ 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);
- 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);
}
+ // (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;
+ }
- final long javaMemorySizeMB = containerMemoryMB - cutoff;
+ /**
+ * Computes the parameters to be used to start a TaskManager Java process.
+ *
+ * @param config The Flink configuration.
+ * @param containerMemoryMB The size of the complete container, in megabytes.
+ * @return The parameters to start the TaskManager processes with.
+ */
+ public static ContaineredTaskManagerParameters create(
+ Configuration config, long containerMemoryMB, int numSlots)
+ {
+ // (1) try to compute how much memory used by container
+ final long cutoffMB = 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;
http://git-wip-us.apache.org/repos/asf/flink/blob/38aa863d/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index 91a4f73..68b4046 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -313,7 +313,7 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
@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())));
http://git-wip-us.apache.org/repos/asf/flink/blob/38aa863d/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
index 8d9ea88..230a934 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParametersTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.clusterframework;
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;
@@ -27,6 +28,7 @@ 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 class ContaineredTaskManagerParametersTest extends TestLogger {
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();
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/38aa863d/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java
index b0c6c60..d3d5444 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesTest.java
@@ -214,5 +214,4 @@ public class TaskManagerServicesTest extends TestLogger {
config.setFloat(TaskManagerOptions.MANAGED_MEMORY_FRACTION, 0.1f); // 10%
assertEquals(810, TaskManagerServices.calculateHeapSizeMB(1000, config));
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/38aa863d/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index bdb471a..eab5e39 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -36,9 +36,11 @@ import org.apache.flink.configuration.SecurityOptions;
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.taskexecutor.TaskManagerServices;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.ShutdownHookUtil;
@@ -410,6 +412,28 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
}
/**
+ * 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;
+ 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);
+ }
+ }
+
+ /**
* This method will block until the ApplicationMaster/JobManager have been deployed on YARN.
*
* @param clusterSpecification Initial cluster specification for the Flink cluster to be deployed
@@ -423,6 +447,9 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
@Nullable JobGraph jobGraph,
boolean detached) throws Exception {
+ // ------------------ Check if configuration is valid --------------------
+ validateClusterSpecification(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),
http://git-wip-us.apache.org/repos/asf/flink/blob/38aa863d/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
index 2cdc19d..1443f99 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -31,6 +31,7 @@ import org.apache.flink.configuration.GlobalConfiguration;
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 class FlinkYarnSessionCli extends AbstractCustomCommandLine<ApplicationId
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();
http://git-wip-us.apache.org/repos/asf/flink/blob/38aa863d/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
index 20ce314..5b0d422 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
@@ -264,8 +264,8 @@ public class FlinkYarnSessionCliTest extends TestLogger {
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(
http://git-wip-us.apache.org/repos/asf/flink/blob/38aa863d/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
index dd8b625..52bf8bb 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.util.TestLogger;
import org.apache.flink.yarn.cli.FlinkYarnSessionCli;
@@ -91,8 +92,11 @@ public class YarnClusterDescriptorTest extends TestLogger {
@Test
public void testFailIfTaskSlotsHigherThanMaxVcores() throws ClusterDeploymentException {
+ final Configuration flinkConfiguration = new Configuration();
+ flinkConfiguration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 0);
+
YarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(
- new Configuration(),
+ flinkConfiguration,
yarnConfiguration,
temporaryFolder.getRoot().getAbsolutePath(),
yarnClient,
@@ -101,8 +105,8 @@ public class YarnClusterDescriptorTest extends TestLogger {
clusterDescriptor.setLocalJarPath(new Path(flinkJar.getPath()));
ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
- .setMasterMemoryMB(-1)
- .setTaskManagerMemoryMB(-1)
+ .setMasterMemoryMB(1)
+ .setTaskManagerMemoryMB(1)
.setNumberTaskManagers(1)
.setSlotsPerTaskManager(Integer.MAX_VALUE)
.createClusterSpecification();
@@ -126,6 +130,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
Configuration configuration = new Configuration();
// overwrite vcores in config
configuration.setInteger(YarnConfigOptions.VCORES, Integer.MAX_VALUE);
+ configuration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 0);
YarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(
configuration,
@@ -138,8 +143,8 @@ public class YarnClusterDescriptorTest extends TestLogger {
// configure slots
ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
- .setMasterMemoryMB(-1)
- .setTaskManagerMemoryMB(-1)
+ .setMasterMemoryMB(1)
+ .setTaskManagerMemoryMB(1)
.setNumberTaskManagers(1)
.setSlotsPerTaskManager(1)
.createClusterSpecification();
[3/3] flink git commit: [FLINK-9022][state] Fix resource release in
StreamTaskStateInitializerImpl.streamOperatorStateContext()
Posted by tr...@apache.org.
[FLINK-9022][state] Fix resource release in StreamTaskStateInitializerImpl.streamOperatorStateContext()
This closes #5716.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f9df13c5
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f9df13c5
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f9df13c5
Branch: refs/heads/master
Commit: f9df13c5058f194a5c686b9b753345d9226fc87a
Parents: 7c952dd
Author: sihuazhou <su...@163.com>
Authored: Mon Mar 19 19:48:32 2018 +0800
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Mar 21 10:56:55 2018 +0100
----------------------------------------------------------------------
.../contrib/streaming/state/RocksDBKeyedStateBackend.java | 3 ++-
.../streaming/api/operators/AbstractStreamOperator.java | 4 ++--
.../api/operators/StreamTaskStateInitializerImpl.java | 9 +++------
3 files changed, 7 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f9df13c5/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 6a66121..41b7bd0 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -1620,7 +1620,6 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
CheckpointOptions checkpointOptions) throws Exception {
long startTime = System.currentTimeMillis();
- final CloseableRegistry snapshotCloseableRegistry = new CloseableRegistry();
if (kvStateInformation.isEmpty()) {
if (LOG.isDebugEnabled()) {
@@ -1647,6 +1646,8 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
CheckpointedStateScope.EXCLUSIVE,
primaryStreamFactory);
+ final CloseableRegistry snapshotCloseableRegistry = new CloseableRegistry();
+
final RocksDBFullSnapshotOperation<K> snapshotOperation =
new RocksDBFullSnapshotOperation<>(
RocksDBKeyedStateBackend.this,
http://git-wip-us.apache.org/repos/asf/flink/blob/f9df13c5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
index 4d3f9f5..e447cbe 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
@@ -248,8 +248,8 @@ public abstract class AbstractStreamOperator<OUT>
context.isRestored(), // information whether we restore or start for the first time
operatorStateBackend, // access to operator state backend
keyedStateStore, // access to keyed state backend
- keyedStateInputs, // access to operator state stream
- operatorStateInputs); // access to keyed state stream
+ keyedStateInputs, // access to keyed state stream
+ operatorStateInputs); // access to operator state stream
initializeState(initializationContext);
} finally {
http://git-wip-us.apache.org/repos/asf/flink/blob/f9df13c5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
index 7e91554..d9bd089 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamTaskStateInitializerImpl.java
@@ -166,11 +166,12 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
// cleanup if something went wrong before results got published.
if (streamTaskCloseableRegistry.unregisterCloseable(keyedStatedBackend)) {
- IOUtils.closeQuietly(keyedStatedBackend);
+ // release resource (e.g native resource)
+ keyedStatedBackend.dispose();
}
if (streamTaskCloseableRegistry.unregisterCloseable(operatorStateBackend)) {
- IOUtils.closeQuietly(keyedStatedBackend);
+ operatorStateBackend.dispose();
}
if (streamTaskCloseableRegistry.unregisterCloseable(rawKeyedStateInputs)) {
@@ -181,10 +182,6 @@ public class StreamTaskStateInitializerImpl implements StreamTaskStateInitialize
IOUtils.closeQuietly(rawOperatorStateInputs);
}
- if (streamTaskCloseableRegistry.unregisterCloseable(rawOperatorStateInputs)) {
- IOUtils.closeQuietly(rawOperatorStateInputs);
- }
-
throw new Exception("Exception while creating StreamOperatorStateContext.", ex);
}
}