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:24 UTC
[flink] branch release-1.10 updated (1950efd -> 91d96ab)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 1950efd [hotfix] Add space after period in TTL_COMPACT_FILTER_ENABLED description
new f5d742f [FLINK-15564][yarn][test] Fix that YarnClusterDescriptorTest#testFailIfTaskSlotsHigherThanMaxVcores and #testConfigOverwrite not validating the original intended behaviors.
new f46048c [FLINK-15564][yarn] Factor max vcores retrieval out of YarnClusterDescriptor
new 3a51e98 [hotfix][yarn] Remove yarn app master env keys that are not used.
new daf5da4 [hotfix][yarn][client] Remove invalid numberTaskManagers from ClusterSpecification.
new 7b6aa1a [FLINK-15598][yarn] Remove unnecessary cluster specification validation in YarnClusterDescriptor#deployInternal.
new 91d96ab [FLINK-15598][yarn] Do not overwrite configuration with cluster specification in YarnClusterDescriptor#startAppMaster.
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../client/deployment/ClusterSpecification.java | 18 +---
.../KubernetesClusterDescriptorTest.java | 1 -
.../flink/kubernetes/KubernetesSessionCliTest.java | 1 -
.../flink/yarn/YARNHighAvailabilityITCase.java | 3 +
.../java/org/apache/flink/yarn/YARNITCase.java | 4 +-
.../apache/flink/yarn/YarnConfigurationITCase.java | 5 +-
.../YarnClientYarnClusterInformationRetriever.java | 56 +++++++++++
.../flink/yarn/YarnClusterClientFactory.java | 1 +
.../apache/flink/yarn/YarnClusterDescriptor.java | 103 ++-------------------
.../yarn/YarnClusterInformationRetriever.java | 14 +--
.../java/org/apache/flink/yarn/YarnConfigKeys.java | 3 -
.../apache/flink/yarn/cli/FlinkYarnSessionCli.java | 22 -----
.../apache/flink/yarn/FlinkYarnSessionCliTest.java | 1 -
.../flink/yarn/YarnClusterDescriptorBuilder.java | 82 ++++++++++++++++
.../flink/yarn/YarnClusterDescriptorTest.java | 22 ++---
.../java/org/apache/flink/yarn/YarnTestUtils.java | 13 ++-
16 files changed, 186 insertions(+), 163 deletions(-)
create mode 100644 flink-yarn/src/main/java/org/apache/flink/yarn/YarnClientYarnClusterInformationRetriever.java
copy flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobGraphStoreFactory.java => flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterInformationRetriever.java (69%)
create mode 100644 flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorBuilder.java
[flink] 03/06: [hotfix][yarn] Remove yarn app master env keys that
are not used.
Posted by tr...@apache.org.
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 3a51e98b9dc65235301b874553c7110593f53b33
Author: Xintong Song <to...@gmail.com>
AuthorDate: Fri Jan 10 13:15:16 2020 +0800
[hotfix][yarn] Remove yarn app master env keys that are not used.
---
.../src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java | 3 ---
flink-yarn/src/main/java/org/apache/flink/yarn/YarnConfigKeys.java | 3 ---
2 files changed, 6 deletions(-)
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 bc5f993..8f48de1 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
@@ -999,13 +999,10 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
appMasterEnv.put(YarnConfigKeys.ENV_FLINK_CLASSPATH, classPathBuilder.toString());
// set Flink on YARN internal configuration values
- appMasterEnv.put(YarnConfigKeys.ENV_TM_COUNT, String.valueOf(clusterSpecification.getNumberTaskManagers()));
- appMasterEnv.put(YarnConfigKeys.ENV_TM_MEMORY, String.valueOf(clusterSpecification.getTaskManagerMemoryMB()));
appMasterEnv.put(YarnConfigKeys.FLINK_JAR_PATH, remotePathJar.toString());
appMasterEnv.put(YarnConfigKeys.ENV_APP_ID, appId.toString());
appMasterEnv.put(YarnConfigKeys.ENV_CLIENT_HOME_DIR, homeDir.toString());
appMasterEnv.put(YarnConfigKeys.ENV_CLIENT_SHIP_FILES, envShipFileList.toString());
- appMasterEnv.put(YarnConfigKeys.ENV_SLOTS, String.valueOf(clusterSpecification.getSlotsPerTaskManager()));
appMasterEnv.put(YarnConfigKeys.ENV_ZOOKEEPER_NAMESPACE, getZookeeperNamespace());
appMasterEnv.put(YarnConfigKeys.FLINK_YARN_FILES, yarnFilesDir.toUri().toString());
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnConfigKeys.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnConfigKeys.java
index 0fc54a2..c73ec3e 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnConfigKeys.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnConfigKeys.java
@@ -27,12 +27,9 @@ public class YarnConfigKeys {
// Environment variable names
// ------------------------------------------------------------------------
- public static final String ENV_TM_MEMORY = "_CLIENT_TM_MEMORY";
- public static final String ENV_TM_COUNT = "_CLIENT_TM_COUNT";
public static final String ENV_APP_ID = "_APP_ID";
public static final String ENV_CLIENT_HOME_DIR = "_CLIENT_HOME_DIR";
public static final String ENV_CLIENT_SHIP_FILES = "_CLIENT_SHIP_FILES";
- public static final String ENV_SLOTS = "_SLOTS";
public static final String ENV_FLINK_CLASSPATH = "_FLINK_CLASSPATH";
[flink] 02/06: [FLINK-15564][yarn] Factor max vcores retrieval out
of YarnClusterDescriptor
Posted by tr...@apache.org.
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 f46048ce3b41a18d750f1be2998f1671f52778fb
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Thu Jan 16 14:55:17 2020 +0100
[FLINK-15564][yarn] Factor max vcores retrieval out of YarnClusterDescriptor
For better testability this commit introduces the YarnClusterInformationRetriever which
is responsible for retrieving the maximum number of vcores.
This closes #10852.
---
.../YarnClientYarnClusterInformationRetriever.java | 56 +++++++++++++++
.../flink/yarn/YarnClusterClientFactory.java | 1 +
.../apache/flink/yarn/YarnClusterDescriptor.java | 22 ++----
.../yarn/YarnClusterInformationRetriever.java | 34 +++++++++
.../flink/yarn/YarnClusterDescriptorBuilder.java | 82 ++++++++++++++++++++++
.../flink/yarn/YarnClusterDescriptorTest.java | 15 ++--
.../java/org/apache/flink/yarn/YarnTestUtils.java | 20 +++---
7 files changed, 199 insertions(+), 31 deletions(-)
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClientYarnClusterInformationRetriever.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClientYarnClusterInformationRetriever.java
new file mode 100644
index 0000000..78f65e0
--- /dev/null
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClientYarnClusterInformationRetriever.java
@@ -0,0 +1,56 @@
+/*
+ * 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.yarn;
+
+import org.apache.flink.util.FlinkException;
+
+import org.apache.hadoop.yarn.api.records.NodeState;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+import java.io.IOException;
+
+/**
+ * {@link YarnClusterInformationRetriever} implementation which queries the Yarn cluster using
+ * a {@link YarnClient} instance.
+ */
+public final class YarnClientYarnClusterInformationRetriever implements YarnClusterInformationRetriever {
+ private final YarnClient yarnClient;
+
+ private YarnClientYarnClusterInformationRetriever(YarnClient yarnClient) {
+ this.yarnClient = yarnClient;
+ }
+
+ @Override
+ public int getMaxVcores() throws FlinkException {
+ try {
+ return yarnClient.getNodeReports(NodeState.RUNNING)
+ .stream()
+ .mapToInt(report -> report.getCapability().getVirtualCores())
+ .max()
+ .orElse(0);
+ } catch (YarnException | IOException e) {
+ throw new FlinkException("Couldn't get cluster description, please check on the YarnConfiguration", e);
+ }
+ }
+
+ public static YarnClientYarnClusterInformationRetriever create(YarnClient yarnClient) {
+ return new YarnClientYarnClusterInformationRetriever(yarnClient);
+ }
+}
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClientFactory.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClientFactory.java
index 7fca2fc..0b5beb2 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClientFactory.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClientFactory.java
@@ -75,6 +75,7 @@ public class YarnClusterClientFactory extends AbstractContainerizedClusterClient
configuration,
yarnConfiguration,
yarnClient,
+ YarnClientYarnClusterInformationRetriever.create(yarnClient),
false);
}
}
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 4b2dff6..bc5f993 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
@@ -131,6 +131,8 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
private final YarnClient yarnClient;
+ private final YarnClusterInformationRetriever yarnClusterInformationRetriever;
+
/** True if the descriptor must not shut down the YarnClient. */
private final boolean sharedYarnClient;
@@ -157,10 +159,12 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
Configuration flinkConfiguration,
YarnConfiguration yarnConfiguration,
YarnClient yarnClient,
+ YarnClusterInformationRetriever yarnClusterInformationRetriever,
boolean sharedYarnClient) {
this.yarnConfiguration = Preconditions.checkNotNull(yarnConfiguration);
this.yarnClient = Preconditions.checkNotNull(yarnClient);
+ this.yarnClusterInformationRetriever = Preconditions.checkNotNull(yarnClusterInformationRetriever);
this.sharedYarnClient = sharedYarnClient;
this.flinkConfiguration = Preconditions.checkNotNull(flinkConfiguration);
@@ -265,7 +269,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
this.shipFiles.addAll(shipFiles);
}
- private void isReadyForDeployment(ClusterSpecification clusterSpecification) throws YarnDeploymentException {
+ private void isReadyForDeployment(ClusterSpecification clusterSpecification) throws Exception {
if (clusterSpecification.getNumberTaskManagers() <= 0) {
throw new YarnDeploymentException("Taskmanager count must be positive");
@@ -278,7 +282,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
}
// Check if we don't exceed YARN's maximum virtual cores.
- final int numYarnMaxVcores = getNumYarnMaxVcores();
+ final int numYarnMaxVcores = yarnClusterInformationRetriever.getMaxVcores();
int configuredAmVcores = flinkConfiguration.getInteger(YarnConfigOptions.APP_MASTER_VCORES);
if (configuredAmVcores > numYarnMaxVcores) {
@@ -308,20 +312,6 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
}
}
- @VisibleForTesting
- protected int getNumYarnMaxVcores() throws YarnDeploymentException {
- // Fetch numYarnMaxVcores from all the RUNNING nodes via yarnClient
- try {
- return yarnClient.getNodeReports(NodeState.RUNNING)
- .stream()
- .mapToInt(report -> report.getCapability().getVirtualCores())
- .max()
- .orElse(0);
- } catch (Exception e) {
- throw new YarnDeploymentException("Couldn't get cluster description, please check on the YarnConfiguration", e);
- }
- }
-
private static boolean allocateResource(int[] nodeManagers, int toAllocate) {
for (int i = 0; i < nodeManagers.length; i++) {
if (nodeManagers[i] >= toAllocate) {
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterInformationRetriever.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterInformationRetriever.java
new file mode 100644
index 0000000..1d8bdfa
--- /dev/null
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterInformationRetriever.java
@@ -0,0 +1,34 @@
+/*
+ * 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.yarn;
+
+import org.apache.flink.util.FlinkException;
+
+/**
+ * Interface which allows to retrieve information about a Yarn cluster.
+ */
+public interface YarnClusterInformationRetriever {
+
+ /**
+ * Get the maximum number of vcores supported by the Yarn cluster.
+ *
+ * @return max vcores supported by the cluster
+ */
+ int getMaxVcores() throws FlinkException;
+}
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorBuilder.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorBuilder.java
new file mode 100644
index 0000000..20b8d35
--- /dev/null
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorBuilder.java
@@ -0,0 +1,82 @@
+/*
+ * 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.yarn;
+
+import org.apache.flink.configuration.Configuration;
+
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+
+import javax.annotation.Nullable;
+
+/**
+ * Builder for the {@link YarnClusterDescriptor}.
+ */
+public final class YarnClusterDescriptorBuilder {
+
+ private final YarnClient yarnClient;
+ private final boolean sharedYarnClient;
+
+ private Configuration flinkConfiguration = new Configuration();
+ private YarnConfiguration yarnConfiguration = new YarnConfiguration();
+
+ @Nullable
+ private YarnClusterInformationRetriever yarnClusterInformationRetriever = null;
+
+ private YarnClusterDescriptorBuilder(YarnClient yarnClient, boolean sharedYarnClient) {
+ this.yarnClient = yarnClient;
+ this.sharedYarnClient = sharedYarnClient;
+ }
+
+ public YarnClusterDescriptorBuilder setFlinkConfiguration(Configuration flinkConfiguration) {
+ this.flinkConfiguration = flinkConfiguration;
+ return this;
+ }
+
+ public YarnClusterDescriptorBuilder setYarnConfiguration(YarnConfiguration yarnConfiguration) {
+ this.yarnConfiguration = yarnConfiguration;
+ return this;
+ }
+
+ public YarnClusterDescriptorBuilder setYarnClusterInformationRetriever(YarnClusterInformationRetriever yarnClusterInformationRetriever) {
+ this.yarnClusterInformationRetriever = yarnClusterInformationRetriever;
+ return this;
+ }
+
+ public YarnClusterDescriptor build() {
+ final YarnClusterInformationRetriever clusterInformationRetriever;
+
+ if (yarnClusterInformationRetriever == null) {
+ clusterInformationRetriever = YarnClientYarnClusterInformationRetriever.create(yarnClient);
+ } else {
+ clusterInformationRetriever = yarnClusterInformationRetriever;
+ }
+
+ return new YarnClusterDescriptor(
+ flinkConfiguration,
+ yarnConfiguration,
+ yarnClient,
+ clusterInformationRetriever,
+ sharedYarnClient);
+ }
+
+ public static YarnClusterDescriptorBuilder newBuilder(YarnClient yarnClient, boolean sharedYarnClient) {
+ return new YarnClusterDescriptorBuilder(yarnClient, sharedYarnClient);
+ }
+}
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 1b32380..7c10211 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
@@ -66,6 +66,7 @@ import static org.junit.Assert.fail;
*/
public class YarnClusterDescriptorTest extends TestLogger {
+ private static final int YARN_MAX_VCORES = 16;
private static YarnConfiguration yarnConfiguration;
private static YarnClient yarnClient;
@@ -535,11 +536,13 @@ public class YarnClusterDescriptorTest extends TestLogger {
}
private YarnClusterDescriptor createYarnClusterDescriptor(Configuration configuration) {
- return YarnTestUtils.createClusterDescriptorWithLogging(
- temporaryFolder.getRoot().getAbsolutePath(),
- configuration,
- yarnConfiguration,
- yarnClient,
- true);
+ YarnTestUtils.configureLogFile(configuration, temporaryFolder.getRoot().getAbsolutePath());
+
+ return YarnClusterDescriptorBuilder.newBuilder(yarnClient, true)
+ .setFlinkConfiguration(configuration)
+ .setYarnConfiguration(yarnConfiguration)
+ .setYarnClusterInformationRetriever(() -> YARN_MAX_VCORES)
+ .build();
+
}
}
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnTestUtils.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnTestUtils.java
index e5ed6d5..ec63d6d 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnTestUtils.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnTestUtils.java
@@ -32,8 +32,6 @@ import java.util.Arrays;
*/
public class YarnTestUtils {
- private static final int NUM_YARN_MAX_VCORES = 16;
-
static boolean isHadoopVersionGreaterThanOrEquals(final int major, final int minor) {
final String[] splitVersion = VersionInfo.getVersion().split("\\.");
final int[] versions = Arrays.stream(splitVersion).mapToInt(Integer::parseInt).toArray();
@@ -50,12 +48,16 @@ public class YarnTestUtils {
final YarnConfiguration yarnConfiguration,
final YarnClient yarnClient,
final boolean sharedYarnClient) {
- final Configuration effectiveConfiguration = FlinkYarnSessionCli.setLogConfigFileInConfig(flinkConfiguration, flinkConfDir);
- return new YarnClusterDescriptor(effectiveConfiguration, yarnConfiguration, yarnClient, sharedYarnClient) {
- @Override
- protected int getNumYarnMaxVcores() {
- return NUM_YARN_MAX_VCORES;
- }
- };
+ final Configuration effectiveConfiguration = configureLogFile(flinkConfiguration, flinkConfDir);
+ return new YarnClusterDescriptor(
+ effectiveConfiguration,
+ yarnConfiguration,
+ yarnClient,
+ YarnClientYarnClusterInformationRetriever.create(yarnClient),
+ sharedYarnClient);
+ }
+
+ public static Configuration configureLogFile(Configuration flinkConfiguration, String flinkConfDir) {
+ return FlinkYarnSessionCli.setLogConfigFileInConfig(flinkConfiguration, flinkConfDir);
}
}
[flink] 04/06: [hotfix][yarn][client] Remove invalid
numberTaskManagers from ClusterSpecification.
Posted by tr...@apache.org.
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 daf5da42e421a786a4bdceec7bfd578b9070b91b
Author: Xintong Song <to...@gmail.com>
AuthorDate: Fri Jan 10 13:33:38 2020 +0800
[hotfix][yarn][client] Remove invalid numberTaskManagers from ClusterSpecification.
After FLIP-6, we can no longer specifiy number of task managers from client side on active deployment. Thus numberTaskManagers in ClusterSpecification is invalid.
This also includes the following changes:
- YarnClusterDescriptor#validateClusterResources no longer validate whether yarn have enough resource for the flink cluster, because the total resource requirements of the flink cluster is not known.
- Remove parallelism from yarn properties file, because we can no longer infer the parallelism from numberTaskManagers * slotsPerTaskManager.
---
.../client/deployment/ClusterSpecification.java | 16 +--------
.../KubernetesClusterDescriptorTest.java | 1 -
.../flink/kubernetes/KubernetesSessionCliTest.java | 1 -
.../java/org/apache/flink/yarn/YARNITCase.java | 1 -
.../apache/flink/yarn/YarnClusterDescriptor.java | 42 ----------------------
.../apache/flink/yarn/cli/FlinkYarnSessionCli.java | 22 ------------
.../apache/flink/yarn/FlinkYarnSessionCliTest.java | 1 -
7 files changed, 1 insertion(+), 83 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterSpecification.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterSpecification.java
index 5c21b17..51bd522 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterSpecification.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterSpecification.java
@@ -24,13 +24,11 @@ package org.apache.flink.client.deployment;
public final class ClusterSpecification {
private final int masterMemoryMB;
private final int taskManagerMemoryMB;
- private final int numberTaskManagers;
private final int slotsPerTaskManager;
- private ClusterSpecification(int masterMemoryMB, int taskManagerMemoryMB, int numberTaskManagers, int slotsPerTaskManager) {
+ private ClusterSpecification(int masterMemoryMB, int taskManagerMemoryMB, int slotsPerTaskManager) {
this.masterMemoryMB = masterMemoryMB;
this.taskManagerMemoryMB = taskManagerMemoryMB;
- this.numberTaskManagers = numberTaskManagers;
this.slotsPerTaskManager = slotsPerTaskManager;
}
@@ -42,10 +40,6 @@ public final class ClusterSpecification {
return taskManagerMemoryMB;
}
- public int getNumberTaskManagers() {
- return numberTaskManagers;
- }
-
public int getSlotsPerTaskManager() {
return slotsPerTaskManager;
}
@@ -55,7 +49,6 @@ public final class ClusterSpecification {
return "ClusterSpecification{" +
"masterMemoryMB=" + masterMemoryMB +
", taskManagerMemoryMB=" + taskManagerMemoryMB +
- ", numberTaskManagers=" + numberTaskManagers +
", slotsPerTaskManager=" + slotsPerTaskManager +
'}';
}
@@ -66,7 +59,6 @@ public final class ClusterSpecification {
public static class ClusterSpecificationBuilder {
private int masterMemoryMB = 768;
private int taskManagerMemoryMB = 1024;
- private int numberTaskManagers = 1;
private int slotsPerTaskManager = 1;
public ClusterSpecificationBuilder setMasterMemoryMB(int masterMemoryMB) {
@@ -79,11 +71,6 @@ public final class ClusterSpecification {
return this;
}
- public ClusterSpecificationBuilder setNumberTaskManagers(int numberTaskManagers) {
- this.numberTaskManagers = numberTaskManagers;
- return this;
- }
-
public ClusterSpecificationBuilder setSlotsPerTaskManager(int slotsPerTaskManager) {
this.slotsPerTaskManager = slotsPerTaskManager;
return this;
@@ -93,7 +80,6 @@ public final class ClusterSpecification {
return new ClusterSpecification(
masterMemoryMB,
taskManagerMemoryMB,
- numberTaskManagers,
slotsPerTaskManager);
}
}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java
index ae77505..167f164 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java
@@ -51,7 +51,6 @@ public class KubernetesClusterDescriptorTest extends KubernetesTestBase {
final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
.setMasterMemoryMB(1234)
.setTaskManagerMemoryMB(1222)
- .setNumberTaskManagers(1)
.setSlotsPerTaskManager(1)
.createClusterSpecification();
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesSessionCliTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesSessionCliTest.java
index 7f4319e..cee44a8 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesSessionCliTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesSessionCliTest.java
@@ -93,7 +93,6 @@ public class KubernetesSessionCliTest {
// each task manager has 3 slots but the parallelism is 7. Thus the slots should be increased.
assertEquals(3, clusterSpecification.getSlotsPerTaskManager());
- assertEquals(1, clusterSpecification.getNumberTaskManagers());
}
@Test
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 7f6f1d9..1252d0b 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
@@ -98,7 +98,6 @@ public class YARNITCase extends YarnTestBase {
.setMasterMemoryMB(768)
.setTaskManagerMemoryMB(1024)
.setSlotsPerTaskManager(1)
- .setNumberTaskManagers(1)
.createClusterSpecification();
File testingJar = YarnTestBase.findFile("..", new YarnTestUtils.TestJarFinder("flink-yarn-tests"));
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 8f48de1..f84de95 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
@@ -101,7 +101,6 @@ import java.nio.file.Files;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -271,9 +270,6 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
private void isReadyForDeployment(ClusterSpecification clusterSpecification) throws Exception {
- if (clusterSpecification.getNumberTaskManagers() <= 0) {
- throw new YarnDeploymentException("Taskmanager count must be positive");
- }
if (this.flinkJarPath == null) {
throw new YarnDeploymentException("The Flink jar path is null");
}
@@ -312,16 +308,6 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
}
}
- private static boolean allocateResource(int[] nodeManagers, int toAllocate) {
- for (int i = 0; i < nodeManagers.length; i++) {
- if (nodeManagers[i] >= toAllocate) {
- nodeManagers[i] -= toAllocate;
- return true;
- }
- }
- return false;
- }
-
public String getZookeeperNamespace() {
return zookeeperNamespace;
}
@@ -559,7 +545,6 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
Resource maximumResourceCapability,
ClusterResourceDescription freeClusterResources) throws YarnDeploymentException {
- int taskManagerCount = clusterSpecification.getNumberTaskManagers();
int jobManagerMemoryMb = clusterSpecification.getMasterMemoryMB();
int taskManagerMemoryMb = clusterSpecification.getTaskManagerMemoryMB();
@@ -593,13 +578,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
"connecting from the beginning because the resources are currently not available in the cluster. " +
"The allocation might take more time than usual because the Flink YARN client needs to wait until " +
"the resources become available.";
- int totalMemoryRequired = jobManagerMemoryMb + taskManagerMemoryMb * taskManagerCount;
-
- if (freeClusterResources.totalFreeMemory < totalMemoryRequired) {
- LOG.warn("This YARN session requires " + totalMemoryRequired + "MB of memory in the cluster. "
- + "There are currently only " + freeClusterResources.totalFreeMemory + "MB available." + noteRsc);
- }
if (taskManagerMemoryMb > freeClusterResources.containerLimit) {
LOG.warn("The requested amount of memory for the TaskManagers (" + taskManagerMemoryMb + "MB) is more than "
+ "the largest possible YARN container: " + freeClusterResources.containerLimit + noteRsc);
@@ -609,30 +588,9 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
+ "the largest possible YARN container: " + freeClusterResources.containerLimit + noteRsc);
}
- // ----------------- check if the requested containers fit into the cluster.
-
- int[] nmFree = Arrays.copyOf(freeClusterResources.nodeManagersFree, freeClusterResources.nodeManagersFree.length);
- // first, allocate the jobManager somewhere.
- if (!allocateResource(nmFree, jobManagerMemoryMb)) {
- LOG.warn("Unable to find a NodeManager that can fit the JobManager/Application master. " +
- "The JobManager requires " + jobManagerMemoryMb + "MB. NodeManagers available: " +
- Arrays.toString(freeClusterResources.nodeManagersFree) + noteRsc);
- }
- // allocate TaskManagers
- for (int i = 0; i < taskManagerCount; i++) {
- if (!allocateResource(nmFree, taskManagerMemoryMb)) {
- LOG.warn("There is not enough memory available in the YARN cluster. " +
- "The TaskManager(s) require " + taskManagerMemoryMb + "MB each. " +
- "NodeManagers available: " + Arrays.toString(freeClusterResources.nodeManagersFree) + "\n" +
- "After allocating the JobManager (" + jobManagerMemoryMb + "MB) and (" + i + "/" + taskManagerCount + ") TaskManagers, " +
- "the following NodeManagers are available: " + Arrays.toString(nmFree) + noteRsc);
- }
- }
-
return new ClusterSpecification.ClusterSpecificationBuilder()
.setMasterMemoryMB(jobManagerMemoryMb)
.setTaskManagerMemoryMB(taskManagerMemoryMb)
- .setNumberTaskManagers(clusterSpecification.getNumberTaskManagers())
.setSlotsPerTaskManager(clusterSpecification.getSlotsPerTaskManager())
.createClusterSpecification();
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 64a1b17..4dc20c7 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
@@ -30,7 +30,6 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.configuration.ConfigUtils;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.JobManagerOptions;
@@ -109,7 +108,6 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
// YARN-session related constants
private static final String YARN_PROPERTIES_FILE = ".yarn-properties-";
private static final String YARN_APPLICATION_ID_KEY = "applicationID";
- private static final String YARN_PROPERTIES_PARALLELISM = "parallelism";
private static final String YARN_PROPERTIES_DYNAMIC_PROPERTIES_STRING = "dynamicPropertiesString";
private static final String YARN_DYNAMIC_PROPERTIES_SEPARATOR = "@@"; // this has to be a regex for String.split()
@@ -499,21 +497,6 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
private Configuration applyYarnProperties(Configuration configuration) throws FlinkException {
final Configuration effectiveConfiguration = new Configuration(configuration);
- // configure the default parallelism from YARN
- String propParallelism = yarnPropertiesFile.getProperty(YARN_PROPERTIES_PARALLELISM);
- if (propParallelism != null) { // maybe the property is not set
- try {
- int parallelism = Integer.parseInt(propParallelism);
- effectiveConfiguration.setInteger(CoreOptions.DEFAULT_PARALLELISM, parallelism);
-
- logAndSysout("YARN properties set default parallelism to " + parallelism);
- }
- catch (NumberFormatException e) {
- throw new FlinkException("Error while parsing the YARN properties: " +
- "Property " + YARN_PROPERTIES_PARALLELISM + " is not an integer.", e);
- }
- }
-
String applicationId = yarnPropertiesFile.getProperty(YARN_APPLICATION_ID_KEY);
if (applicationId != null) {
effectiveConfiguration.setString(YarnConfigOptions.APPLICATION_ID, applicationId);
@@ -573,7 +556,6 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
writeYarnPropertiesFile(
yarnApplicationId,
- clusterSpecification.getNumberTaskManagers() * clusterSpecification.getSlotsPerTaskManager(),
dynamicPropertiesEncoded);
} catch (Exception e) {
try {
@@ -718,16 +700,12 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
private void writeYarnPropertiesFile(
ApplicationId yarnApplicationId,
- int parallelism,
@Nullable String dynamicProperties) {
// file that we write into the conf/ dir containing the jobManager address and the dop.
final File yarnPropertiesFile = getYarnPropertiesLocation(yarnPropertiesFileLocation);
Properties yarnProps = new Properties();
yarnProps.setProperty(YARN_APPLICATION_ID_KEY, yarnApplicationId.toString());
- if (parallelism > 0) {
- yarnProps.setProperty(YARN_PROPERTIES_PARALLELISM, Integer.toString(parallelism));
- }
// add dynamic properties
if (dynamicProperties != null) {
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 e1bd658..5cfaa64 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
@@ -114,7 +114,6 @@ public class FlinkYarnSessionCliTest extends TestLogger {
// each task manager has 3 slots but the parallelism is 7. Thus the slots should be increased.
assertEquals(3, clusterSpecification.getSlotsPerTaskManager());
- assertEquals(1, clusterSpecification.getNumberTaskManagers());
}
@Test
[flink] 06/06: [FLINK-15598][yarn] Do not overwrite configuration
with cluster specification in YarnClusterDescriptor#startAppMaster.
Posted by tr...@apache.org.
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;
[flink] 01/06: [FLINK-15564][yarn][test] Fix that
YarnClusterDescriptorTest#testFailIfTaskSlotsHigherThanMaxVcores and
#testConfigOverwrite not validating the original intended behaviors.
Posted by tr...@apache.org.
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 f5d742fdb6ebefbdd5799f6911855f493bc8fca7
Author: Xintong Song <to...@gmail.com>
AuthorDate: Mon Jan 13 19:18:58 2020 +0800
[FLINK-15564][yarn][test] Fix that YarnClusterDescriptorTest#testFailIfTaskSlotsHigherThanMaxVcores and #testConfigOverwrite not validating the original intended behaviors.
---
.../client/deployment/ClusterSpecification.java | 2 +-
.../apache/flink/yarn/YarnClusterDescriptor.java | 26 +++++++++++++---------
.../flink/yarn/YarnClusterDescriptorTest.java | 7 ------
.../java/org/apache/flink/yarn/YarnTestUtils.java | 9 +++++++-
4 files changed, 24 insertions(+), 20 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterSpecification.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterSpecification.java
index 4205465..5c21b17 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterSpecification.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterSpecification.java
@@ -65,7 +65,7 @@ public final class ClusterSpecification {
*/
public static class ClusterSpecificationBuilder {
private int masterMemoryMB = 768;
- private int taskManagerMemoryMB = 768;
+ private int taskManagerMemoryMB = 1024;
private int numberTaskManagers = 1;
private int slotsPerTaskManager = 1;
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 2525d08..4b2dff6 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
@@ -278,17 +278,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
}
// Check if we don't exceed YARN's maximum virtual cores.
- // Fetch numYarnMaxVcores from all the RUNNING nodes via yarnClient
- final int numYarnMaxVcores;
- try {
- numYarnMaxVcores = yarnClient.getNodeReports(NodeState.RUNNING)
- .stream()
- .mapToInt(report -> report.getCapability().getVirtualCores())
- .max()
- .orElse(0);
- } catch (Exception e) {
- throw new YarnDeploymentException("Couldn't get cluster description, please check on the YarnConfiguration", e);
- }
+ final int numYarnMaxVcores = getNumYarnMaxVcores();
int configuredAmVcores = flinkConfiguration.getInteger(YarnConfigOptions.APP_MASTER_VCORES);
if (configuredAmVcores > numYarnMaxVcores) {
@@ -318,6 +308,20 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
}
}
+ @VisibleForTesting
+ protected int getNumYarnMaxVcores() throws YarnDeploymentException {
+ // Fetch numYarnMaxVcores from all the RUNNING nodes via yarnClient
+ try {
+ return yarnClient.getNodeReports(NodeState.RUNNING)
+ .stream()
+ .mapToInt(report -> report.getCapability().getVirtualCores())
+ .max()
+ .orElse(0);
+ } catch (Exception e) {
+ throw new YarnDeploymentException("Couldn't get cluster description, please check on the YarnConfiguration", e);
+ }
+ }
+
private static boolean allocateResource(int[] nodeManagers, int toAllocate) {
for (int i = 0; i < nodeManagers.length; i++) {
if (nodeManagers[i] >= toAllocate) {
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 af24dd8..1b32380 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
@@ -104,9 +104,6 @@ public class YarnClusterDescriptorTest extends TestLogger {
clusterDescriptor.setLocalJarPath(new Path(flinkJar.getPath()));
ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
- .setMasterMemoryMB(1)
- .setTaskManagerMemoryMB(1)
- .setNumberTaskManagers(1)
.setSlotsPerTaskManager(Integer.MAX_VALUE)
.createClusterSpecification();
@@ -137,10 +134,6 @@ public class YarnClusterDescriptorTest extends TestLogger {
// configure slots
ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
- .setMasterMemoryMB(1)
- .setTaskManagerMemoryMB(1)
- .setNumberTaskManagers(1)
- .setSlotsPerTaskManager(1)
.createClusterSpecification();
try {
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnTestUtils.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnTestUtils.java
index 4aef1de..e5ed6d5 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnTestUtils.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnTestUtils.java
@@ -32,6 +32,8 @@ import java.util.Arrays;
*/
public class YarnTestUtils {
+ private static final int NUM_YARN_MAX_VCORES = 16;
+
static boolean isHadoopVersionGreaterThanOrEquals(final int major, final int minor) {
final String[] splitVersion = VersionInfo.getVersion().split("\\.");
final int[] versions = Arrays.stream(splitVersion).mapToInt(Integer::parseInt).toArray();
@@ -49,6 +51,11 @@ public class YarnTestUtils {
final YarnClient yarnClient,
final boolean sharedYarnClient) {
final Configuration effectiveConfiguration = FlinkYarnSessionCli.setLogConfigFileInConfig(flinkConfiguration, flinkConfDir);
- return new YarnClusterDescriptor(effectiveConfiguration, yarnConfiguration, yarnClient, sharedYarnClient);
+ return new YarnClusterDescriptor(effectiveConfiguration, yarnConfiguration, yarnClient, sharedYarnClient) {
+ @Override
+ protected int getNumYarnMaxVcores() {
+ return NUM_YARN_MAX_VCORES;
+ }
+ };
}
}
[flink] 05/06: [FLINK-15598][yarn] Remove unnecessary cluster
specification validation in YarnClusterDescriptor#deployInternal.
Posted by tr...@apache.org.
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 7b6aa1a63f45742557efc5d0bef2db40ebcc683d
Author: Xintong Song <to...@gmail.com>
AuthorDate: Fri Jan 10 12:17:24 2020 +0800
[FLINK-15598][yarn] Remove unnecessary cluster specification validation in YarnClusterDescriptor#deployInternal.
Currently, the only thing YarnClusterDescriptor#validateClusterSpecification does is to validate that ClusterSpecification#taskManagerMemoryMB and flinkConfiguration together are valid for creating TaskExecutorResourceSpec.
However, such validation is already covered by AbstractContainerizedClusterClientFactory#getClusterSpecification.
---
.../apache/flink/yarn/YarnClusterDescriptor.java | 24 ----------------------
1 file changed, 24 deletions(-)
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 f84de95..ab35ce2 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
@@ -42,7 +42,6 @@ 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;
-import org.apache.flink.runtime.clusterframework.TaskExecutorResourceUtils;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
@@ -414,26 +413,6 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
}
/**
- * Method to validate cluster specification before deploy it, it will throw
- * an {@link FlinkException} if the {@link ClusterSpecification} is invalid.
- *
- * @param clusterSpecification cluster specification to check against the configuration of the
- * YarnClusterDescriptor
- * @throws FlinkException if the cluster cannot be started with the provided {@link ClusterSpecification}
- */
- private void validateClusterSpecification(ClusterSpecification clusterSpecification) throws FlinkException {
- MemorySize totalProcessMemory = MemorySize.parse(clusterSpecification.getTaskManagerMemoryMB() + "m");
- try {
- TaskExecutorResourceUtils
- .newResourceSpecBuilder(flinkConfiguration)
- .withTotalProcessMemory(totalProcessMemory)
- .build();
- } catch (IllegalArgumentException iae) {
- throw new FlinkException("Inconsistent cluster specification.", iae);
- }
- }
-
- /**
* 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
@@ -449,9 +428,6 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
@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),