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/08 13:22:34 UTC
[flink] branch release-1.10 updated (90d5b76 -> f1ee817)
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 90d5b76 [hotfix][doc] Add missing execution configuration in config.zh.md
new e2ce35c [hotfix][yarn] Move yarn/Utils#getEnvironmentVariables to BootstrapTools
new f1ee817 [FLINK-15483][kubernetes] Starting jobmanager pod should respect environment config option
The 2 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:
.../decorators/FlinkMasterDeploymentDecorator.java | 10 +++++++++
.../flink/kubernetes/KubernetesTestBase.java | 8 ++++++++
.../kubernetes/kubeclient/Fabric8ClientTest.java | 3 +++
.../runtime/clusterframework/BootstrapTools.java | 18 ++++++++++++++++
.../clusterframework/BootstrapToolsTest.java | 24 ++++++++++++++++++++++
.../test/java/org/apache/flink/yarn/UtilsTest.java | 23 ---------------------
.../src/main/java/org/apache/flink/yarn/Utils.java | 18 ----------------
.../apache/flink/yarn/YarnClusterDescriptor.java | 3 ++-
8 files changed, 65 insertions(+), 42 deletions(-)
[flink] 02/02: [FLINK-15483][kubernetes] Starting jobmanager pod
should respect environment config option
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 f1ee817b951b5b9140798d23dee9f871711ca0e3
Author: wangyang0918 <da...@alibaba-inc.com>
AuthorDate: Mon Jan 6 16:28:06 2020 +0800
[FLINK-15483][kubernetes] Starting jobmanager pod should respect environment config option
The config option "containerized.master.env." could be used to set the environment variables. For example, containerized.master.env.LD_LIBRARY_PATH: "/usr/lib/native".
This closes #10788.
---
.../kubeclient/decorators/FlinkMasterDeploymentDecorator.java | 10 ++++++++++
.../java/org/apache/flink/kubernetes/KubernetesTestBase.java | 8 ++++++++
.../apache/flink/kubernetes/kubeclient/Fabric8ClientTest.java | 3 +++
3 files changed, 21 insertions(+)
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkMasterDeploymentDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkMasterDeploymentDecorator.java
index 996359e..5b061f5 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkMasterDeploymentDecorator.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkMasterDeploymentDecorator.java
@@ -23,16 +23,19 @@ import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.configuration.BlobServerOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptionsInternal;
import org.apache.flink.kubernetes.kubeclient.resources.KubernetesDeployment;
import org.apache.flink.kubernetes.utils.Constants;
import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.clusterframework.BootstrapTools;
import io.fabric8.kubernetes.api.model.Container;
import io.fabric8.kubernetes.api.model.ContainerBuilder;
import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
+import io.fabric8.kubernetes.api.model.EnvVar;
import io.fabric8.kubernetes.api.model.PodSpec;
import io.fabric8.kubernetes.api.model.PodSpecBuilder;
import io.fabric8.kubernetes.api.model.ResourceRequirements;
@@ -43,6 +46,7 @@ import io.fabric8.kubernetes.api.model.apps.DeploymentSpecBuilder;
import java.io.File;
import java.util.Arrays;
import java.util.Map;
+import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -134,6 +138,12 @@ public class FlinkMasterDeploymentDecorator extends Decorator<Deployment, Kubern
new ContainerPortBuilder().withContainerPort(flinkConfig.getInteger(RestOptions.PORT)).build(),
new ContainerPortBuilder().withContainerPort(flinkConfig.getInteger(JobManagerOptions.PORT)).build(),
new ContainerPortBuilder().withContainerPort(blobServerPort).build()))
+ .withEnv(
+ BootstrapTools.getEnvironmentVariables(ResourceManagerOptions.CONTAINERIZED_MASTER_ENV_PREFIX, flinkConfig)
+ .entrySet()
+ .stream()
+ .map(kv -> new EnvVar(kv.getKey(), kv.getValue(), null))
+ .collect(Collectors.toList()))
.withVolumeMounts(KubernetesUtils.getConfigMapVolumeMount(flinkConfDirInPod, hasLogback, hasLog4j))
.build();
}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java
index 551f11f..32ce9e8 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java
@@ -21,6 +21,7 @@ package org.apache.flink.kubernetes;
import org.apache.flink.configuration.BlobServerOptions;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptionsInternal;
@@ -82,6 +83,10 @@ public class KubernetesTestBase extends TestLogger {
protected static final String MOCK_SERVICE_IP = "192.168.0.1";
+ protected static final String FLINK_MASTER_ENV_KEY = "LD_LIBRARY_PATH";
+
+ protected static final String FLINK_MASTER_ENV_VALUE = "/usr/lib/native";
+
@Before
public void setUp() throws IOException {
FLINK_CONFIG.setString(KubernetesConfigOptions.NAMESPACE, NAMESPACE);
@@ -91,6 +96,9 @@ public class KubernetesTestBase extends TestLogger {
FLINK_CONFIG.setString(KubernetesConfigOptionsInternal.ENTRY_POINT_CLASS, "main-class");
FLINK_CONFIG.setString(BlobServerOptions.PORT, String.valueOf(Constants.BLOB_SERVER_PORT));
FLINK_CONFIG.setString(TaskManagerOptions.RPC_PORT, String.valueOf(Constants.TASK_MANAGER_RPC_PORT));
+ FLINK_CONFIG.setString(
+ ResourceManagerOptions.CONTAINERIZED_MASTER_ENV_PREFIX + FLINK_MASTER_ENV_KEY,
+ FLINK_MASTER_ENV_VALUE);
flinkConfDir = temporaryFolder.newFolder().getAbsoluteFile();
BootstrapTools.writeConfiguration(new Configuration(), new File(flinkConfDir, "flink-conf.yaml"));
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8ClientTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8ClientTest.java
index 6d54617..3271a92 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8ClientTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8ClientTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.kubernetes.utils.Constants;
import io.fabric8.kubernetes.api.model.ConfigMap;
import io.fabric8.kubernetes.api.model.Container;
import io.fabric8.kubernetes.api.model.ContainerPort;
+import io.fabric8.kubernetes.api.model.EnvVar;
import io.fabric8.kubernetes.api.model.EnvVarBuilder;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodSpec;
@@ -187,6 +188,8 @@ public class Fabric8ClientTest extends KubernetesTestBase {
assertEquals(new File(mountPath, FLINK_CONF_FILENAME).getPath(),
jmContainer.getVolumeMounts().get(0).getMountPath());
assertEquals(FLINK_CONF_FILENAME, jmContainer.getVolumeMounts().get(0).getSubPath());
+
+ assertThat(jmContainer.getEnv(), Matchers.contains(new EnvVar(FLINK_MASTER_ENV_KEY, FLINK_MASTER_ENV_VALUE, null)));
}
@Test
[flink] 01/02: [hotfix][yarn] Move
yarn/Utils#getEnvironmentVariables to BootstrapTools
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 e2ce35c3580a71d74772fc3301eb2bbf6365b7e3
Author: wangyang0918 <da...@alibaba-inc.com>
AuthorDate: Mon Jan 6 16:25:49 2020 +0800
[hotfix][yarn] Move yarn/Utils#getEnvironmentVariables to BootstrapTools
The method BootstrapTools#getEnvironmentVariables could be reused by Yarn and Kubernetes.
---
.../runtime/clusterframework/BootstrapTools.java | 18 ++++++++++++++++
.../clusterframework/BootstrapToolsTest.java | 24 ++++++++++++++++++++++
.../test/java/org/apache/flink/yarn/UtilsTest.java | 23 ---------------------
.../src/main/java/org/apache/flink/yarn/Utils.java | 18 ----------------
.../apache/flink/yarn/YarnClusterDescriptor.java | 3 ++-
5 files changed, 44 insertions(+), 42 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
index d6d577f..04c8433 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
@@ -712,4 +712,22 @@ public class BootstrapTools {
}
return memory - heapLimit;
}
+
+ /**
+ * Method to extract environment variables from the flinkConfiguration based on the given prefix String.
+ *
+ * @param envPrefix Prefix for the environment variables key
+ * @param flinkConfiguration The Flink config to get the environment variable definition from
+ */
+ public static Map<String, String> getEnvironmentVariables(String envPrefix, Configuration flinkConfiguration) {
+ Map<String, String> result = new HashMap<>();
+ for (Map.Entry<String, String> entry: flinkConfiguration.toMap().entrySet()) {
+ if (entry.getKey().startsWith(envPrefix) && entry.getKey().length() > envPrefix.length()) {
+ // remove prefix
+ String key = entry.getKey().substring(envPrefix.length());
+ result.put(key, entry.getValue());
+ }
+ }
+ return result;
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java
index d18d288..71666f6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java
@@ -43,6 +43,7 @@ import java.net.InetAddress;
import java.net.ServerSocket;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
@@ -509,4 +510,27 @@ public class BootstrapToolsTest extends TestLogger {
conf.setString(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key(), "6000");
BootstrapTools.calculateHeapSize(4000, conf);
}
+
+ @Test
+ public void testGetEnvironmentVariables() {
+ Configuration testConf = new Configuration();
+ testConf.setString("containerized.master.env.LD_LIBRARY_PATH", "/usr/lib/native");
+
+ Map<String, String> res = BootstrapTools.getEnvironmentVariables("containerized.master.env.", testConf);
+
+ Assert.assertEquals(1, res.size());
+ Map.Entry<String, String> entry = res.entrySet().iterator().next();
+ Assert.assertEquals("LD_LIBRARY_PATH", entry.getKey());
+ Assert.assertEquals("/usr/lib/native", entry.getValue());
+ }
+
+ @Test
+ public void testGetEnvironmentVariablesErroneous() {
+ Configuration testConf = new Configuration();
+ testConf.setString("containerized.master.env.", "/usr/lib/native");
+
+ Map<String, String> res = BootstrapTools.getEnvironmentVariables("containerized.master.env.", testConf);
+
+ Assert.assertEquals(0, res.size());
+ }
}
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/UtilsTest.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/UtilsTest.java
index ae16cd6..276c2c9 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/UtilsTest.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/UtilsTest.java
@@ -81,29 +81,6 @@ public class UtilsTest extends TestLogger {
}
@Test
- public void testGetEnvironmentVariables() {
- Configuration testConf = new Configuration();
- testConf.setString("yarn.application-master.env.LD_LIBRARY_PATH", "/usr/lib/native");
-
- Map<String, String> res = Utils.getEnvironmentVariables("yarn.application-master.env.", testConf);
-
- Assert.assertEquals(1, res.size());
- Map.Entry<String, String> entry = res.entrySet().iterator().next();
- Assert.assertEquals("LD_LIBRARY_PATH", entry.getKey());
- Assert.assertEquals("/usr/lib/native", entry.getValue());
- }
-
- @Test
- public void testGetEnvironmentVariablesErroneous() {
- Configuration testConf = new Configuration();
- testConf.setString("yarn.application-master.env.", "/usr/lib/native");
-
- Map<String, String> res = Utils.getEnvironmentVariables("yarn.application-master.env.", testConf);
-
- Assert.assertEquals(0, res.size());
- }
-
- @Test
public void testCreateTaskExecutorCredentials() throws Exception {
File root = temporaryFolder.getRoot();
File home = new File(root, "home");
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java b/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
index d2c40e2..6d61c3d 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/Utils.java
@@ -361,24 +361,6 @@ public final class Utils {
}
/**
- * Method to extract environment variables from the flinkConfiguration based on the given prefix String.
- *
- * @param envPrefix Prefix for the environment variables key
- * @param flinkConfiguration The Flink config to get the environment variable definition from
- */
- public static Map<String, String> getEnvironmentVariables(String envPrefix, org.apache.flink.configuration.Configuration flinkConfiguration) {
- Map<String, String> result = new HashMap<>();
- for (Map.Entry<String, String> entry: flinkConfiguration.toMap().entrySet()) {
- if (entry.getKey().startsWith(envPrefix) && entry.getKey().length() > envPrefix.length()) {
- // remove prefix
- String key = entry.getKey().substring(envPrefix.length());
- result.put(key, entry.getValue());
- }
- }
- return result;
- }
-
- /**
* Creates the launch context, which describes how to bring up a TaskExecutor / TaskManager process in
* an allocated YARN container.
*
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 7940c8a..2525d08 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,7 +999,8 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
// Setup CLASSPATH and environment variables for ApplicationMaster
final Map<String, String> appMasterEnv = new HashMap<>();
// set user specified app master environment variables
- appMasterEnv.putAll(Utils.getEnvironmentVariables(ResourceManagerOptions.CONTAINERIZED_MASTER_ENV_PREFIX, configuration));
+ appMasterEnv.putAll(
+ BootstrapTools.getEnvironmentVariables(ResourceManagerOptions.CONTAINERIZED_MASTER_ENV_PREFIX, configuration));
// set Flink app class path
appMasterEnv.put(YarnConfigKeys.ENV_FLINK_CLASSPATH, classPathBuilder.toString());