Skip to content

Commit 47bad97

Browse files
author
Vitaliy Dmitriev
committed
ADH-7362: added method to get JAVA_HOME from env
1 parent 6de9af3 commit 47bad97

5 files changed

Lines changed: 81 additions & 11 deletions

File tree

‎flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import java.util.concurrent.ThreadLocalRandom;
4040
import java.util.stream.Collectors;
4141

42+
import static org.apache.flink.configuration.ConfigConstants.ENV_JAVA_HOME;
4243
import static org.apache.flink.configuration.MetricOptions.SYSTEM_RESOURCE_METRICS;
4344
import static org.apache.flink.configuration.MetricOptions.SYSTEM_RESOURCE_METRICS_PROBING_INTERVAL;
4445
import static org.apache.flink.configuration.StructuredOptionsSplitter.escapeWithSingleQuote;
@@ -665,6 +666,29 @@ public static boolean filterPrefixMapKey(String key, String candidate) {
665666
return candidate.startsWith(prefixKey);
666667
}
667668

669+
/**
670+
* Set the JAVA_HOME variable in the provided environment map.
671+
*
672+
* <p>This method follows a specific priority order to determine the JAVA_HOME value:
673+
*
674+
* <ol>
675+
* <li>If the environment map already contains the JAVA_HOME key, the method does nothing.
676+
* <li>Otherwise, it attempts to retrieve JAVA_HOME from the Flink configuration using {@link
677+
* CoreOptions#FLINK_JAVA_HOME}.
678+
* <li>If it isn't found in configuration, it falls back to the system environment variable.
679+
* <li>If a value is found through either source, it is added to the environment map.
680+
* </ol>
681+
*/
682+
public static void setJavaHomeEnv(Configuration configuration, Map<String, String> env) {
683+
if (!env.containsKey(ENV_JAVA_HOME)) {
684+
Optional.ofNullable(
685+
configuration
686+
.getOptional(CoreOptions.FLINK_JAVA_HOME)
687+
.orElse(System.getenv(ENV_JAVA_HOME)))
688+
.ifPresent(javaHomeStr -> env.put(ENV_JAVA_HOME, javaHomeStr));
689+
}
690+
}
691+
668692
static Map<String, String> convertToPropertiesPrefixed(
669693
Map<String, Object> confData, String key, boolean standardYaml) {
670694
final String prefixKey = key + ".";

‎flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/ContaineredTaskManagerParameters.java‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,12 @@
1919
package org.apache.flink.runtime.clusterframework;
2020

2121
import org.apache.flink.configuration.Configuration;
22-
import org.apache.flink.configuration.CoreOptions;
22+
import org.apache.flink.configuration.ConfigurationUtils;
2323
import org.apache.flink.configuration.ResourceManagerOptions;
2424

2525
import java.util.HashMap;
2626
import java.util.Map;
2727

28-
import static org.apache.flink.configuration.ConfigConstants.ENV_JAVA_HOME;
29-
3028
/** This class describes the basic parameters for launching a TaskManager process. */
3129
public class ContaineredTaskManagerParameters implements java.io.Serializable {
3230

@@ -94,8 +92,7 @@ public static ContaineredTaskManagerParameters create(
9492
}
9593

9694
// set JAVA_HOME
97-
config.getOptional(CoreOptions.FLINK_JAVA_HOME)
98-
.ifPresent(javaHome -> envVars.put(ENV_JAVA_HOME, javaHome));
95+
ConfigurationUtils.setJavaHomeEnv(config, envVars);
9996

10097
// done
10198
return new ContaineredTaskManagerParameters(taskExecutorProcessSpec, envVars);

‎flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -137,7 +137,6 @@
137137
import static org.apache.flink.configuration.ConfigConstants.DEFAULT_FLINK_USR_LIB_DIR;
138138
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_LIB_DIR;
139139
import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_OPT_DIR;
140-
import static org.apache.flink.configuration.ConfigConstants.ENV_JAVA_HOME;
141140
import static org.apache.flink.configuration.ResourceManagerOptions.CONTAINERIZED_MASTER_ENV_PREFIX;
142141
import static org.apache.flink.configuration.ResourceManagerOptions.CONTAINERIZED_TASK_MANAGER_ENV_PREFIX;
143142
import static org.apache.flink.runtime.entrypoint.component.FileJobGraphRetriever.JOB_GRAPH_FILE_PATH;
@@ -1970,10 +1969,12 @@ Map<String, String> generateApplicationMasterEnv(
19701969
ConfigurationUtils.getPrefixedKeyValuePairs(
19711970
ResourceManagerOptions.CONTAINERIZED_MASTER_ENV_PREFIX,
19721971
this.flinkConfiguration));
1972+
1973+
LOG.warn("JAVA_HOME:");
1974+
LOG.warn(System.getenv("JAVA_HOME"));
1975+
19731976
// set JAVA_HOME
1974-
this.flinkConfiguration
1975-
.getOptional(CoreOptions.FLINK_JAVA_HOME)
1976-
.ifPresent(javaHome -> env.put(ENV_JAVA_HOME, javaHome));
1977+
ConfigurationUtils.setJavaHomeEnv(this.flinkConfiguration, env);
19771978
// set Flink app class path
19781979
env.put(ENV_FLINK_CLASSPATH, classPathStr);
19791980
// Set FLINK_LIB_DIR to `lib` folder under working dir in container

‎flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.flink.configuration.Configuration;
2525
import org.apache.flink.configuration.CoreOptions;
2626
import org.apache.flink.configuration.MemorySize;
27+
import org.apache.flink.core.testutils.CommonTestUtils;
2728
import org.apache.flink.runtime.clusterframework.ContaineredTaskManagerParameters;
2829
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
2930
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
@@ -690,14 +691,37 @@ void testGetTaskManagerEnvsWithJavaHomeSet() {
690691
assertThat(envVars).containsEntry(ENV_JAVA_HOME, "/opt/jdk").containsEntry("key", "val");
691692
}
692693

694+
@Test
695+
void testGetTaskManagerEnvsWithEnvJavaHomeSet() {
696+
final Configuration cfg = new Configuration();
697+
final String newJavaHome = "/usr/lib/jvm/java-openjdk-17";
698+
final Map<String, String> oldEnv = System.getenv();
699+
try {
700+
Map<String, String> newEnv = new HashMap<>(System.getenv());
701+
newEnv.put(ConfigConstants.ENV_JAVA_HOME, newJavaHome);
702+
CommonTestUtils.setEnv(newEnv);
703+
704+
cfg.setString(CONTAINERIZED_TASK_MANAGER_ENV_PREFIX + "key", "val");
705+
final ContaineredTaskManagerParameters containeredParams =
706+
ContaineredTaskManagerParameters.create(cfg, TASK_EXECUTOR_PROCESS_SPEC);
707+
final Map<String, String> envVars = containeredParams.taskManagerEnv();
708+
assertThat(envVars)
709+
.containsEntry(ENV_JAVA_HOME, newJavaHome)
710+
.containsEntry("key", "val");
711+
} finally {
712+
CommonTestUtils.setEnv(oldEnv);
713+
}
714+
}
715+
693716
@Test
694717
void testGetTaskManagerEnvsWithoutJavaHomeSet() {
695718
final Configuration cfg = new Configuration();
719+
final String origJavaHome = System.getenv(ConfigConstants.ENV_JAVA_HOME);
696720
cfg.setString(CONTAINERIZED_TASK_MANAGER_ENV_PREFIX + "key", "val");
697721
final ContaineredTaskManagerParameters containeredParams =
698722
ContaineredTaskManagerParameters.create(cfg, TASK_EXECUTOR_PROCESS_SPEC);
699723
final Map<String, String> envVars = containeredParams.taskManagerEnv();
700-
assertThat(envVars).doesNotContainKey(ENV_JAVA_HOME);
724+
assertThat(envVars.get(ConfigConstants.ENV_JAVA_HOME)).isEqualTo(origJavaHome);
701725
}
702726

703727
private static void verifyUnitResourceVariousSchedulers(

‎flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -947,14 +947,38 @@ public void testGenerateApplicationMasterEnv(@TempDir File flinkHomeDir) throws
947947
public void testContainerEnvJavaHomeNotOverriddenByDefault(@TempDir File flinkHomeDir)
948948
throws IOException {
949949
final Configuration flinkConfig = new Configuration();
950+
final String origJavaHome = System.getenv(ConfigConstants.ENV_JAVA_HOME);
950951
final Map<String, String> masterEnv =
951952
getTestMasterEnv(
952953
flinkConfig,
953954
flinkHomeDir,
954955
"",
955956
"./lib/flink_dist.jar",
956957
ApplicationId.newInstance(0, 0));
957-
assertThat(masterEnv).doesNotContainKey(ConfigConstants.ENV_JAVA_HOME);
958+
assertThat(masterEnv.get(ConfigConstants.ENV_JAVA_HOME)).isEqualTo(origJavaHome);
959+
}
960+
961+
@Test
962+
public void testContainerEnvJavaHomeNewValue(@TempDir File flinkHomeDir) throws IOException {
963+
final Configuration flinkConfig = new Configuration();
964+
final String newJavaHome = "/usr/lib/jvm/java-openjdk-17";
965+
final Map<String, String> oldEnv = System.getenv();
966+
967+
try {
968+
Map<String, String> newEnv = new HashMap<>(System.getenv());
969+
newEnv.put(ConfigConstants.ENV_JAVA_HOME, newJavaHome);
970+
CommonTestUtils.setEnv(newEnv);
971+
final Map<String, String> masterEnv =
972+
getTestMasterEnv(
973+
flinkConfig,
974+
flinkHomeDir,
975+
"",
976+
"./lib/flink_dist.jar",
977+
ApplicationId.newInstance(0, 0));
978+
assertThat(masterEnv.get(ConfigConstants.ENV_JAVA_HOME)).isEqualTo(newJavaHome);
979+
} finally {
980+
CommonTestUtils.setEnv(oldEnv);
981+
}
958982
}
959983

960984
@Test

0 commit comments

Comments
 (0)