[FLINK-6058] Don't read DEFAULT_PARALLELISM from GlobalConfiguration

cchepelov · Jul 11, 2017 · 4aa2ffc · 4aa2ffc
1 parent d0e5295
commit 4aa2ffc
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 7 deletions.
diff --git a/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.client;
 
+import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.common.JobID;
@@ -143,6 +144,8 @@ public class CliFrontend {
 
 	private final FiniteDuration clientTimeout;
 
+	private final int defaultParallelism;
+
 	/**
 	 *
 	 * @throws Exception Thrown if the configuration directory was not found, the configuration could not be loaded
@@ -169,6 +172,9 @@ public CliFrontend(String configDir) throws Exception {
 		}
 
 		this.clientTimeout = AkkaUtils.getClientTimeout(config);
+		this.defaultParallelism = GlobalConfiguration.loadConfiguration().getInteger(
+														ConfigConstants.DEFAULT_PARALLELISM_KEY,
+														ConfigConstants.DEFAULT_PARALLELISM);
 	}
 
 	// --------------------------------------------------------------------------------------------
@@ -250,6 +256,8 @@ protected int run(String[] args) {
 					+ client.getMaxSlots() + "). "
 					+ "To use another parallelism, set it at the ./bin/flink client.");
 				userParallelism = client.getMaxSlots();
+			} else if (ExecutionConfig.PARALLELISM_DEFAULT == userParallelism) {
+				userParallelism = defaultParallelism;
 			}
 
 			return executeProgram(program, client, userParallelism);
@@ -314,6 +322,9 @@ protected int info(String[] args) {
 
 		try {
 			int parallelism = options.getParallelism();
+			if (ExecutionConfig.PARALLELISM_DEFAULT == parallelism) {
+				parallelism = defaultParallelism;
+			}
 
 			LOG.info("Creating program plan dump");
 

diff --git a/flink-clients/src/test/java/org/apache/flink/client/CliFrontendRunTest.java b/flink-clients/src/test/java/org/apache/flink/client/CliFrontendRunTest.java
@@ -58,7 +58,7 @@ public void testRun() {
 			// test without parallelism
 			{
 				String[] parameters = {"-v", getTestJarPath()};
-				RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(-1, true, false);
+				RunTestingCliFrontend testFrontend = new RunTestingCliFrontend(1, true, false);
 				assertEquals(0, testFrontend.run(parameters));
 			}
 

diff --git a/...va/src/main/java/org/apache/flink/streaming/api/environment/StreamContextEnvironment.java b/...va/src/main/java/org/apache/flink/streaming/api/environment/StreamContextEnvironment.java
@@ -21,8 +21,6 @@
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.client.program.ContextEnvironment;
 import org.apache.flink.client.program.DetachedEnvironment;
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.streaming.api.graph.StreamGraph;
 import org.apache.flink.util.Preconditions;
 
@@ -45,10 +43,6 @@ protected StreamContextEnvironment(ContextEnvironment ctx) {
 		this.ctx = ctx;
 		if (ctx.getParallelism() > 0) {
 			setParallelism(ctx.getParallelism());
-		} else {
-			setParallelism(GlobalConfiguration.loadConfiguration().getInteger(
-					ConfigConstants.DEFAULT_PARALLELISM_KEY,
-					ConfigConstants.DEFAULT_PARALLELISM));
 		}
 	}