You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2020/03/19 16:04:54 UTC
[flink] branch release-1.10 updated: [FLINK-15852][cli] Prioritize
ExecutorCLI over YarnSessionCLI for active CLI
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.10 by this push:
new aa4d627 [FLINK-15852][cli] Prioritize ExecutorCLI over YarnSessionCLI for active CLI
aa4d627 is described below
commit aa4d6270e4e764daa392d70369b46df5b7ba4054
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Tue Mar 17 21:23:22 2020 +0100
[FLINK-15852][cli] Prioritize ExecutorCLI over YarnSessionCLI for active CLI
---
.../org/apache/flink/client/cli/CliFrontend.java | 26 ++++++--------
.../apache/flink/yarn/FlinkYarnSessionCliTest.java | 42 ++++++++++++++++++++++
2 files changed, 52 insertions(+), 16 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index c568f63..facf9d5 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -174,10 +174,7 @@ public class CliFrontend {
LOG.info("Running 'run' command.");
final Options commandOptions = CliFrontendParser.getRunCommandOptions();
-
- final Options commandLineOptions = CliFrontendParser.mergeOptions(commandOptions, customCommandLineOptions);
-
- final CommandLine commandLine = CliFrontendParser.parse(commandLineOptions, args, true);
+ final CommandLine commandLine = getCommandLine(commandOptions, args, true);
final ProgramOptions programOptions = new ProgramOptions(commandLine);
@@ -304,10 +301,7 @@ public class CliFrontend {
LOG.info("Running 'list' command.");
final Options commandOptions = CliFrontendParser.getListCommandOptions();
-
- final Options commandLineOptions = CliFrontendParser.mergeOptions(commandOptions, customCommandLineOptions);
-
- final CommandLine commandLine = CliFrontendParser.parse(commandLineOptions, args, false);
+ final CommandLine commandLine = getCommandLine(commandOptions, args, false);
ListOptions listOptions = new ListOptions(commandLine);
@@ -427,8 +421,7 @@ public class CliFrontend {
LOG.info("Running 'stop-with-savepoint' command.");
final Options commandOptions = CliFrontendParser.getStopCommandOptions();
- final Options commandLineOptions = CliFrontendParser.mergeOptions(commandOptions, customCommandLineOptions);
- final CommandLine commandLine = CliFrontendParser.parse(commandLineOptions, args, false);
+ final CommandLine commandLine = getCommandLine(commandOptions, args, false);
final StopOptions stopOptions = new StopOptions(commandLine);
if (stopOptions.isPrintHelp()) {
@@ -474,10 +467,7 @@ public class CliFrontend {
LOG.info("Running 'cancel' command.");
final Options commandOptions = CliFrontendParser.getCancelCommandOptions();
-
- final Options commandLineOptions = CliFrontendParser.mergeOptions(commandOptions, customCommandLineOptions);
-
- final CommandLine commandLine = CliFrontendParser.parse(commandLineOptions, args, false);
+ final CommandLine commandLine = getCommandLine(commandOptions, args, false);
CancelOptions cancelOptions = new CancelOptions(commandLine);
@@ -550,6 +540,11 @@ public class CliFrontend {
}
}
+ public CommandLine getCommandLine(final Options commandOptions, final String[] args, final boolean stopAtNonOptions) throws CliArgsException {
+ final Options commandLineOptions = CliFrontendParser.mergeOptions(commandOptions, customCommandLineOptions);
+ return CliFrontendParser.parse(commandLineOptions, args, stopAtNonOptions);
+ }
+
/**
* Executes the SAVEPOINT action.
*
@@ -1021,6 +1016,7 @@ public class CliFrontend {
public static List<CustomCommandLine> loadCustomCommandLines(Configuration configuration, String configurationDirectory) {
List<CustomCommandLine> customCommandLines = new ArrayList<>();
+ customCommandLines.add(new ExecutorCLI(configuration));
// Command line interface of the YARN session, with a special initialization here
// to prefix all options with y/yarn.
@@ -1036,8 +1032,6 @@ public class CliFrontend {
LOG.warn("Could not load CLI class {}.", flinkYarnSessionCLI, e);
}
- customCommandLines.add(new ExecutorCLI(configuration));
-
// Tips: DefaultCLI must be added at last, because getActiveCustomCommandLine(..) will get the
// active CustomCommandLine in order and DefaultCLI isActive always return true.
customCommandLines.add(new DefaultCLI(configuration));
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 9a4299c..a3831d6 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
@@ -18,6 +18,11 @@
package org.apache.flink.yarn;
+import org.apache.flink.client.cli.CliArgsException;
+import org.apache.flink.client.cli.CliFrontend;
+import org.apache.flink.client.cli.CliFrontendParser;
+import org.apache.flink.client.cli.CustomCommandLine;
+import org.apache.flink.client.cli.ExecutorCLI;
import org.apache.flink.client.deployment.ClusterClientFactory;
import org.apache.flink.client.deployment.ClusterClientServiceLoader;
import org.apache.flink.client.deployment.ClusterSpecification;
@@ -35,6 +40,7 @@ import org.apache.flink.util.FlinkException;
import org.apache.flink.util.TestLogger;
import org.apache.flink.yarn.cli.FlinkYarnSessionCli;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
+import org.apache.flink.yarn.executors.YarnJobClusterExecutor;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
@@ -49,6 +55,7 @@ import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.StandardOpenOption;
+import java.util.List;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
@@ -161,6 +168,41 @@ public class FlinkYarnSessionCliTest extends TestLogger {
assertEquals(nodeLabelCliInput, descriptor.getNodeLabel());
}
+ @Test
+ public void testExecutorCLIisPrioritised() throws Exception {
+ final File directoryPath = writeYarnPropertiesFile(validPropertiesFile);
+
+ final Configuration configuration = new Configuration();
+ configuration.setString(YarnConfigOptions.PROPERTIES_FILE_LOCATION, directoryPath.getAbsolutePath());
+
+ validateYarnCLIisActive(configuration);
+
+ final String[] argsUnderTest = new String[] {"-e", YarnJobClusterExecutor.NAME};
+
+ validateExecutorCLIisPrioritised(configuration, argsUnderTest);
+ }
+
+ private void validateExecutorCLIisPrioritised(Configuration configuration, String[] argsUnderTest) throws IOException, CliArgsException {
+ final List<CustomCommandLine> customCommandLines = CliFrontend.loadCustomCommandLines(
+ configuration,
+ tmp.newFile().getAbsolutePath());
+
+ final CliFrontend cli = new CliFrontend(configuration, customCommandLines);
+ final CommandLine commandLine = cli.getCommandLine(
+ CliFrontendParser.getRunCommandOptions(),
+ argsUnderTest,
+ true);
+
+ final CustomCommandLine customCommandLine = cli.getActiveCustomCommandLine(commandLine);
+ assertTrue(customCommandLine instanceof ExecutorCLI);
+ }
+
+ private void validateYarnCLIisActive(Configuration configuration) throws FlinkException, CliArgsException {
+ final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCli(configuration);
+ final CommandLine testCLIArgs = flinkYarnSessionCli.parseCommandLineOptions(new String[] {}, true);
+ assertTrue(flinkYarnSessionCli.isActive(testCLIArgs));
+ }
+
/**
* Test that the CliFrontend is able to pick up the .yarn-properties file from a specified location.
*/