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 2020/01/16 15:19:30 UTC
[flink] 06/06: [FLINK-15598][yarn] Do not overwrite configuration
with cluster specification in YarnClusterDescriptor#startAppMaster.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 91d96abe5f42bd088a326870b4885d79611fccb5
Author: Xintong Song <to...@gmail.com>
AuthorDate: Fri Jan 10 13:56:11 2020 +0800
[FLINK-15598][yarn] Do not overwrite configuration with cluster specification in YarnClusterDescriptor#startAppMaster.
ClusterSpecification is generated from configuration at the first place, in AbstractContainerizedClusterClientFactory. It does not make sense to write the values back.
We are aware that ClusterSpecification#taskManagerMemoryMB might be updated in YarnClusterDescriptor#validateClusterResources, which is before reaching YarnClusterDescriptor#startAppMaster. However, this will only happen if the TM process memory is configured smaller than Yarn minimum allocation. While using the original process memory size may lead to waste, it should not cause any serious problems or failures. On the other hand, overwriting the process memory size may lead to failur [...]
This closes #10863.
---
.../apache/flink/yarn/YARNHighAvailabilityITCase.java | 3 +++
.../src/test/java/org/apache/flink/yarn/YARNITCase.java | 3 +++
.../org/apache/flink/yarn/YarnConfigurationITCase.java | 5 ++++-
.../org/apache/flink/yarn/YarnClusterDescriptor.java | 16 +---------------
4 files changed, 11 insertions(+), 16 deletions(-)
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
index 5ba1671..6420078 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
@@ -26,8 +26,10 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmaster.JobResult;
@@ -271,6 +273,7 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
@Nonnull
private YarnClusterDescriptor setupYarnClusterDescriptor() {
final Configuration flinkConfiguration = new Configuration();
+ flinkConfiguration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
flinkConfiguration.setString(YarnConfigOptions.APPLICATION_ATTEMPTS, "10");
flinkConfiguration.setString(HighAvailabilityOptions.HA_MODE, "zookeeper");
flinkConfiguration.setString(HighAvailabilityOptions.HA_STORAGE_PATH, storageDir);
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
index 1252d0b..89d17f8 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
@@ -22,6 +22,8 @@ import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -85,6 +87,7 @@ public class YARNITCase extends YarnTestBase {
private void deployPerjob(YarnConfigOptions.UserJarInclusion userJarInclusion, JobGraph jobGraph) throws Exception {
Configuration configuration = new Configuration();
+ configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
configuration.setString(AkkaOptions.ASK_TIMEOUT, "30 s");
configuration.setString(CLASSPATH_INCLUDE_USER_JAR, userJarInclusion.toString());
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
index 81fef14..ba4fbb4 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
@@ -26,6 +26,7 @@ import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ResourceManagerOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.TaskExecutorResourceSpec;
import org.apache.flink.runtime.clusterframework.TaskExecutorResourceUtils;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -85,10 +86,12 @@ public class YarnConfigurationITCase extends YarnTestBase {
// disable heap cutoff min
configuration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 0);
+ final int slotsPerTaskManager = 3;
+ configuration.set(TaskManagerOptions.NUM_TASK_SLOTS, slotsPerTaskManager);
+
final TaskExecutorResourceSpec tmResourceSpec = TaskExecutorResourceUtils.resourceSpecFromConfig(configuration);
final int masterMemory = 64;
final int taskManagerMemory = tmResourceSpec.getTotalProcessMemorySize().getMebiBytes();
- final int slotsPerTaskManager = 3;
final YarnConfiguration yarnConfiguration = getYarnConfiguration();
final YarnClusterDescriptor clusterDescriptor = YarnTestUtils.createClusterDescriptorWithLogging(
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
index ab35ce2..131685a 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
@@ -34,11 +34,9 @@ import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.JobManagerOptions;
-import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.configuration.SecurityOptions;
-import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.plugin.PluginConfig;
import org.apache.flink.core.plugin.PluginUtils;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
@@ -522,7 +520,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
ClusterResourceDescription freeClusterResources) throws YarnDeploymentException {
int jobManagerMemoryMb = clusterSpecification.getMasterMemoryMB();
- int taskManagerMemoryMb = clusterSpecification.getTaskManagerMemoryMB();
+ final int taskManagerMemoryMb = clusterSpecification.getTaskManagerMemoryMB();
if (jobManagerMemoryMb < yarnMinAllocationMB || taskManagerMemoryMb < yarnMinAllocationMB) {
LOG.warn("The JobManager or TaskManager memory is below the smallest possible YARN Container size. "
@@ -535,9 +533,6 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
if (jobManagerMemoryMb < yarnMinAllocationMB) {
jobManagerMemoryMb = yarnMinAllocationMB;
}
- if (taskManagerMemoryMb < yarnMinAllocationMB) {
- taskManagerMemoryMb = yarnMinAllocationMB;
- }
final String note = "Please check the 'yarn.scheduler.maximum-allocation-mb' and the 'yarn.nodemanager.resource.memory-mb' configuration values\n";
if (jobManagerMemoryMb > maximumResourceCapability.getMemory()) {
@@ -775,15 +770,6 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
paths.add(remotePathJar);
classPathBuilder.append(flinkJarPath.getName()).append(File.pathSeparator);
- // set the right configuration values for the TaskManager
- configuration.setInteger(
- TaskManagerOptions.NUM_TASK_SLOTS,
- clusterSpecification.getSlotsPerTaskManager());
-
- configuration.set(
- TaskManagerOptions.TOTAL_PROCESS_MEMORY,
- MemorySize.parse(clusterSpecification.getTaskManagerMemoryMB() + "m"));
-
// Upload the flink configuration
// write out configuration file
File tmpConfigurationFile = null;