You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2022/05/17 04:02:00 UTC
[flink] branch release-1.15 updated: [FLINK-26909][Client/Job Submission] Allow to set parallelism to be -1 from Cli
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new 5f2c9c7a255 [FLINK-26909][Client/Job Submission] Allow to set parallelism to be -1 from Cli
5f2c9c7a255 is described below
commit 5f2c9c7a255c4e2d9d739dcb3ee5477a11ada90d
Author: liujiangang <li...@kuaishou.com>
AuthorDate: Thu May 5 21:55:23 2022 +0800
[FLINK-26909][Client/Job Submission] Allow to set parallelism to be -1 from Cli
This closes #19651.
---
.../apache/flink/client/cli/ProgramOptions.java | 8 ++++--
.../flink/client/cli/CliFrontendRunTest.java | 33 ++++++++++++++++++++++
2 files changed, 39 insertions(+), 2 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java b/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
index 15844d3ca86..b215cabea1f 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/ProgramOptions.java
@@ -57,6 +57,8 @@ public class ProgramOptions extends CommandLineOptions {
private final String[] programArgs;
+ private final boolean hasParallelismOpt;
+
private final int parallelism;
private final boolean detachedMode;
@@ -93,10 +95,11 @@ public class ProgramOptions extends CommandLineOptions {
this.classpaths = classpaths;
if (line.hasOption(PARALLELISM_OPTION.getOpt())) {
+ hasParallelismOpt = true;
String parString = line.getOptionValue(PARALLELISM_OPTION.getOpt());
try {
parallelism = Integer.parseInt(parString);
- if (parallelism <= 0) {
+ if (parallelism <= 0 && parallelism != ExecutionConfig.PARALLELISM_DEFAULT) {
throw new NumberFormatException();
}
} catch (NumberFormatException e) {
@@ -104,6 +107,7 @@ public class ProgramOptions extends CommandLineOptions {
"The parallelism must be a positive number: " + parString);
}
} else {
+ hasParallelismOpt = false;
parallelism = ExecutionConfig.PARALLELISM_DEFAULT;
}
@@ -169,7 +173,7 @@ public class ProgramOptions extends CommandLineOptions {
}
public void applyToConfiguration(Configuration configuration) {
- if (getParallelism() != ExecutionConfig.PARALLELISM_DEFAULT) {
+ if (hasParallelismOpt) {
configuration.setInteger(CoreOptions.DEFAULT_PARALLELISM, getParallelism());
}
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
index ea82cc9bc1f..d1650c8cabc 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendRunTest.java
@@ -18,10 +18,12 @@
package org.apache.flink.client.cli;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.client.deployment.ClusterClientServiceLoader;
import org.apache.flink.client.deployment.DefaultClusterClientServiceLoader;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.runtime.jobgraph.RestoreMode;
import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
@@ -208,6 +210,37 @@ public class CliFrontendRunTest extends CliFrontendTestBase {
testFrontend.run(parameters);
}
+ @Test(expected = CliArgsException.class)
+ public void testInvalidNegativeParallelismOption() throws Exception {
+ String[] parameters = {"-p", "-2", getTestJarPath()};
+ Configuration configuration = new Configuration();
+ CliFrontend testFrontend =
+ new CliFrontend(configuration, Collections.singletonList(getCli()));
+ testFrontend.run(parameters);
+ }
+
+ @Test
+ public void testDefaultParallelismOption() throws Exception {
+ final Configuration configuration = getConfiguration();
+ String[] parameters = {
+ "-p", String.valueOf(ExecutionConfig.PARALLELISM_DEFAULT), getTestJarPath()
+ };
+ verifyCliFrontend(
+ configuration, getCli(), parameters, ExecutionConfig.PARALLELISM_DEFAULT, false);
+ }
+
+ @Test
+ public void testDefaultParallelismOptionOverridesConfiguration() throws Exception {
+ // The parallelism should be the same with Cli param -1 instead of config 1.
+ final Configuration configuration = getConfiguration();
+ configuration.set(CoreOptions.DEFAULT_PARALLELISM, 1);
+ String[] parameters = {
+ "-p", String.valueOf(ExecutionConfig.PARALLELISM_DEFAULT), getTestJarPath()
+ };
+ verifyCliFrontend(
+ configuration, getCli(), parameters, ExecutionConfig.PARALLELISM_DEFAULT, false);
+ }
+
// --------------------------------------------------------------------------------------------
public static void verifyCliFrontend(