Revert "[FLINK-17935] Move set yarn.log-config-file to YarnClusterCli…

…entFactory.createClusterDescriptor()"

This reverts commit 2eb7377

flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java

@@ -1045,7 +1045,7 @@ static void setJobManagerAddressInConfig(Configuration config, InetSocketAddress
 
  public static List<CustomCommandLine> loadCustomCommandLines(Configuration configuration, String configurationDirectory) {
  List<CustomCommandLine> customCommandLines = new ArrayList<>();
- customCommandLines.add(new ExecutorCLI(configuration, configurationDirectory));
+ customCommandLines.add(new ExecutorCLI(configuration));
 
  // Command line interface of the YARN session, with a special initialization here
  // to prefix all options with y/yarn.

flink-clients/src/main/java/org/apache/flink/client/cli/ExecutorCLI.java

@@ -21,7 +21,6 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.DeploymentOptions;
-import org.apache.flink.configuration.DeploymentOptionsInternal;
 import org.apache.flink.configuration.UnmodifiableConfiguration;
 import org.apache.flink.core.execution.DefaultExecutorServiceLoader;
 import org.apache.flink.core.execution.PipelineExecutor;
@@ -73,11 +72,8 @@ public class ExecutorCLI implements CustomCommandLine {
 
  private final Configuration baseConfiguration;
 
- private final String configurationDir;
-
- public ExecutorCLI(final Configuration configuration, final String configDir) {
+ public ExecutorCLI(final Configuration configuration) {
  this.baseConfiguration = new UnmodifiableConfiguration(checkNotNull(configuration));
- this.configurationDir = checkNotNull(configDir);
  }
 
  @Override
@@ -119,7 +115,6 @@ public Configuration applyCommandLineOptionsToConfiguration(final CommandLine co
  }
 
  encodeDynamicProperties(commandLine, effectiveConfiguration);
- effectiveConfiguration.set(DeploymentOptionsInternal.CONF_DIR, configurationDir);
 
  if (LOG.isDebugEnabled()) {
  LOG.debug("Effective Configuration: {}", effectiveConfiguration);

flink-clients/src/test/java/org/apache/flink/client/cli/ExecutorCLITest.java

@@ -26,9 +26,7 @@
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Options;
 import org.junit.Before;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
 
 import java.util.Arrays;
 import java.util.List;
@@ -43,18 +41,13 @@
  */
 public class ExecutorCLITest {
 
- @Rule
- public TemporaryFolder tmp = new TemporaryFolder();
-
  private Options testOptions;
 
  @Before
  public void initOptions() {
  testOptions = new Options();
 
- final ExecutorCLI cliUnderTest = new ExecutorCLI(
- new Configuration(),
- tmp.getRoot().getAbsolutePath());
+ final ExecutorCLI cliUnderTest = new ExecutorCLI(new Configuration());
  cliUnderTest.addGeneralOptions(testOptions);
  }
 
@@ -64,9 +57,7 @@ public void testExecutorInBaseConfigIsPickedUp() throws CliArgsException {
  final Configuration loadedConfig = new Configuration();
  loadedConfig.set(DeploymentOptions.TARGET, expectedExecutorName);
 
- final ExecutorCLI cliUnderTest = new ExecutorCLI(
- loadedConfig,
- tmp.getRoot().getAbsolutePath());
+ final ExecutorCLI cliUnderTest = new ExecutorCLI(loadedConfig);
  final CommandLine emptyCommandLine = CliFrontendParser.parse(testOptions, new String[0], true);
 
  final Configuration configuration = cliUnderTest.applyCommandLineOptionsToConfiguration(emptyCommandLine);
@@ -96,9 +87,7 @@ public void testWithPreexistingConfigurationInConstructor() throws CliArgsExcept
  "-D" + CoreOptions.DEFAULT_PARALLELISM.key() + "=5"
  };
 
- final ExecutorCLI cliUnderTest = new ExecutorCLI(
- loadedConfig,
- tmp.getRoot().getAbsolutePath());
+ final ExecutorCLI cliUnderTest = new ExecutorCLI(loadedConfig);
  final CommandLine commandLine = CliFrontendParser.parse(testOptions, args, true);
 
  final Configuration configuration = cliUnderTest.applyCommandLineOptionsToConfiguration(commandLine);
@@ -124,9 +113,7 @@ private void testIsActiveHelper(final String executorOption) throws CliArgsExcep
  final ConfigOption<Integer> configOption = key("test.int").intType().noDefaultValue();
  final int expectedValue = 42;
 
- final ExecutorCLI cliUnderTest = new ExecutorCLI(
- new Configuration(),
- tmp.getRoot().getAbsolutePath());
+ final ExecutorCLI cliUnderTest = new ExecutorCLI(new Configuration());
 
  final String[] args = {executorOption, expectedExecutorName, "-D" + configOption.key() + "=" + expectedValue};
  final CommandLine commandLine = CliFrontendParser.parse(testOptions, args, true);

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java

@@ -22,7 +22,6 @@
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.client.cli.AbstractCustomCommandLine;
 import org.apache.flink.client.cli.CliArgsException;
-import org.apache.flink.client.cli.CliFrontend;
 import org.apache.flink.client.cli.ExecutorCLI;
 import org.apache.flink.client.deployment.ClusterClientFactory;
 import org.apache.flink.client.deployment.ClusterClientServiceLoader;
@@ -69,14 +68,14 @@ public class KubernetesSessionCli {
  private final ExecutorCLI cli;
  private final ClusterClientServiceLoader clusterClientServiceLoader;
 
- public KubernetesSessionCli(Configuration configuration, String configDir) {
- this(configuration, new DefaultClusterClientServiceLoader(), configDir);
+ public KubernetesSessionCli(Configuration configuration) {
+ this(configuration, new DefaultClusterClientServiceLoader());
  }
 
- public KubernetesSessionCli(Configuration configuration, ClusterClientServiceLoader clusterClientServiceLoader, String configDir) {
+ public KubernetesSessionCli(Configuration configuration, ClusterClientServiceLoader clusterClientServiceLoader) {
  this.baseConfiguration = new UnmodifiableConfiguration(checkNotNull(configuration));
  this.clusterClientServiceLoader = checkNotNull(clusterClientServiceLoader);
- this.cli = new ExecutorCLI(baseConfiguration, configDir);
+ this.cli = new ExecutorCLI(baseConfiguration);
  }
 
  public Configuration getEffectiveConfiguration(String[] args) throws CliArgsException {
@@ -179,12 +178,10 @@ private Tuple2<Boolean, Boolean> repStep(BufferedReader in) throws IOException,
  public static void main(String[] args) {
  final Configuration configuration = GlobalConfiguration.loadConfiguration();
 
- final String configDir = CliFrontend.getConfigurationDirectoryFromEnv();
-
  int retCode;
 
  try {
- final KubernetesSessionCli cli = new KubernetesSessionCli(configuration, configDir);
+ final KubernetesSessionCli cli = new KubernetesSessionCli(configuration);
  retCode = SecurityUtils.getInstalledContext().runSecured(() -> cli.run(args));
  } catch (CliArgsException e) {
  retCode = AbstractCustomCommandLine.handleCliArgsException(e, LOG);

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkConfMountDecorator.java

@@ -19,6 +19,7 @@
 package org.apache.flink.kubernetes.kubeclient.decorators;
 
 import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.client.cli.CliFrontend;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
 import org.apache.flink.kubernetes.kubeclient.FlinkPod;
@@ -166,7 +167,7 @@ String getFlinkConfData(Map<String, String> propertiesMap) throws IOException {
  }
 
  private List<File> getLocalLogConfFiles() {
- final String confDir = kubernetesComponentConf.getConfigDirectory();
+ final String confDir = CliFrontend.getConfigurationDirectoryFromEnv();
  final File logbackFile = new File(confDir, CONFIG_FILE_LOGBACK_NAME);
  final File log4jFile = new File(confDir, CONFIG_FILE_LOG4J_NAME);
 

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/AbstractKubernetesParameters.java

@@ -18,8 +18,8 @@
 
 package org.apache.flink.kubernetes.kubeclient.parameters;
 
+import org.apache.flink.client.cli.CliFrontend;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.DeploymentOptionsInternal;
 import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
 import org.apache.flink.kubernetes.utils.Constants;
 
@@ -54,13 +54,6 @@ public Configuration getFlinkConfiguration() {
  return flinkConfig;
  }
 
- @Override
- public String getConfigDirectory() {
- final String configDir = flinkConfig.get(DeploymentOptionsInternal.CONF_DIR);
- checkNotNull(configDir);
- return configDir;
- }
-
  @Override
  public String getClusterId() {
  final String clusterId = flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID);
@@ -137,14 +130,14 @@ public String getContainerEntrypoint() {
 
  @Override
  public boolean hasLogback() {
- final String confDir = getConfigDirectory();
+ final String confDir = CliFrontend.getConfigurationDirectoryFromEnv();
  final File logbackFile = new File(confDir, CONFIG_FILE_LOGBACK_NAME);
  return logbackFile.exists();
  }
 
  @Override
  public boolean hasLog4j() {
- final String confDir = getConfigDirectory();
+ final String confDir = CliFrontend.getConfigurationDirectoryFromEnv();
  final File log4jFile = new File(confDir, CONFIG_FILE_LOG4J_NAME);
  return log4jFile.exists();
  }

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesParameters.java

@@ -32,8 +32,6 @@
  */
 public interface KubernetesParameters {
 
- String getConfigDirectory();
-
  String getClusterId();
 
  String getNamespace();

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/cli/KubernetesSessionCliTest.java

@@ -25,17 +25,14 @@
 import org.apache.flink.client.deployment.DefaultClusterClientServiceLoader;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.DeploymentOptions;
-import org.apache.flink.configuration.DeploymentOptionsInternal;
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.configuration.MemorySize;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
 import org.apache.flink.kubernetes.executors.KubernetesSessionClusterExecutor;
 
 import org.junit.Assert;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
 
 import java.util.Map;
 
@@ -49,14 +46,9 @@
  */
 public class KubernetesSessionCliTest {
 
- @Rule
- public TemporaryFolder tmp = new TemporaryFolder();
-
  @Test
  public void testKubernetesSessionCliSetsDeploymentTargetCorrectly() throws CliArgsException {
- final KubernetesSessionCli cli = new KubernetesSessionCli(
- new Configuration(),
- tmp.getRoot().getAbsolutePath());
+ final KubernetesSessionCli cli = new KubernetesSessionCli(new Configuration());
 
  final String[] args = {};
  final Configuration configuration = cli.getEffectiveConfiguration(args);
@@ -67,9 +59,7 @@ public void testKubernetesSessionCliSetsDeploymentTargetCorrectly() throws CliAr
  @Test
  public void testDynamicProperties() throws Exception {
 
- final KubernetesSessionCli cli = new KubernetesSessionCli(
- new Configuration(),
- tmp.getRoot().getAbsolutePath());
+ final KubernetesSessionCli cli = new KubernetesSessionCli(new Configuration());
  final String[] args = new String[] {
  "-e", KubernetesSessionClusterExecutor.NAME,
  "-Dakka.ask.timeout=5 min",
@@ -82,10 +72,9 @@ public void testDynamicProperties() throws Exception {
  Assert.assertNotNull(clientFactory);
 
  final Map<String, String> executorConfigMap = executorConfig.toMap();
- assertEquals(4, executorConfigMap.size());
+ assertEquals(3, executorConfigMap.size());
  assertEquals("5 min", executorConfigMap.get("akka.ask.timeout"));
  assertEquals("-DappName=foobar", executorConfigMap.get("env.java.opts"));
- assertEquals(tmp.getRoot().getAbsolutePath(), executorConfig.get(DeploymentOptionsInternal.CONF_DIR));
  assertTrue(executorConfigMap.containsKey(DeploymentOptions.TARGET.key()));
  }
 
@@ -142,9 +131,7 @@ public void testCommandLineClusterSpecification() throws Exception {
  "-D" + TaskManagerOptions.NUM_TASK_SLOTS.key() + "=" + slotsPerTaskManager
  };
 
- final KubernetesSessionCli cli = new KubernetesSessionCli(
- configuration,
- tmp.getRoot().getAbsolutePath());
+ final KubernetesSessionCli cli = new KubernetesSessionCli(configuration);
 
  Configuration executorConfig = cli.getEffectiveConfiguration(args);
  ClusterClientFactory<String> clientFactory = getClusterClientFactory(executorConfig);
@@ -170,9 +157,7 @@ public void testConfigurationClusterSpecification() throws Exception {
  configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, slotsPerTaskManager);
 
  final String[] args = {"-e", KubernetesSessionClusterExecutor.NAME};
- final KubernetesSessionCli cli = new KubernetesSessionCli(
- configuration,
- tmp.getRoot().getAbsolutePath());
+ final KubernetesSessionCli cli = new KubernetesSessionCli(configuration);
 
  Configuration executorConfig = cli.getEffectiveConfiguration(args);
  ClusterClientFactory<String> clientFactory = getClusterClientFactory(executorConfig);
@@ -235,9 +220,7 @@ public void testHeapMemoryPropertyWithOldConfigKey() throws Exception {
  configuration.setInteger(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY_MB, 2048);
  configuration.setInteger(TaskManagerOptions.TASK_MANAGER_HEAP_MEMORY_MB, 4096);
 
- final KubernetesSessionCli cli = new KubernetesSessionCli(
- configuration,
- tmp.getRoot().getAbsolutePath());
+ final KubernetesSessionCli cli = new KubernetesSessionCli(configuration);
 
  final Configuration executorConfig = cli.getEffectiveConfiguration(new String[]{});
  final ClusterClientFactory<String> clientFactory = getClusterClientFactory(executorConfig);
@@ -275,8 +258,6 @@ private KubernetesSessionCli createFlinkKubernetesCustomCliWithJmAndTmTotalMemor
  Configuration configuration = new Configuration();
  configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalMemory));
  configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalMemory));
- return new KubernetesSessionCli(
- configuration,
- tmp.getRoot().getAbsolutePath());
+ return new KubernetesSessionCli(configuration);
  }
 }

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala

@@ -140,7 +140,10 @@ object FlinkShell {
  }
 
  private def getConfigDir(config: Config) = {
- config.configDir.getOrElse(CliFrontend.getConfigurationDirectoryFromEnv)
+ config.configDir match {
+ case Some(confDir) => confDir
+ case None => CliFrontend.getConfigurationDirectoryFromEnv
+ }
  }
 
  private def getGlobalConfig(config: Config) = {
@@ -229,7 +232,8 @@ object FlinkShell {
  val effectiveConfig = new Configuration(flinkConfig)
  val args = parseYarnArgList(config, "yarn-cluster")
 
- val configurationDirectory = getConfigDir(config)
+ val configurationDirectory =
+ config.configDir.getOrElse(CliFrontend.getConfigurationDirectoryFromEnv)
 
  val frontend = new CliFrontend(
  effectiveConfig,
@@ -267,7 +271,8 @@ object FlinkShell {
  val effectiveConfig = new Configuration(flinkConfig)
  val args = parseYarnArgList(config, mode)
 
- val configurationDirectory = getConfigDir(config)
+ val configurationDirectory =
+ config.configDir.getOrElse(CliFrontend.getConfigurationDirectoryFromEnv)
 
  val frontend = new CliFrontend(
  effectiveConfig,