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:36 UTC
[flink] 02/02: [FLINK-15483][kubernetes] Starting jobmanager pod
should respect environment config option
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