[FLINK-9554][scala-shell] Respect customCommandlines

This closes apache#6140.
dw33qq · Jul 4, 2018 · 80be8e1 · 80be8e1
1 parent fb37d51
commit 80be8e1
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 7 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
@@ -166,6 +166,10 @@ public Configuration getConfiguration() {
 		return copiedConfiguration;
 	}
 
+	public Options getCustomCommandLineOptions() {
+		return customCommandLineOptions;
+	}
+
 	// --------------------------------------------------------------------------------------------
 	//  Execute Actions
 	// --------------------------------------------------------------------------------------------

diff --git a/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala b/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
@@ -20,6 +20,7 @@ package org.apache.flink.api.scala
 
 import java.io._
 
+import org.apache.commons.cli.{CommandLine, Options}
 import org.apache.flink.client.cli.{CliFrontend, CliFrontendParser}
 import org.apache.flink.client.deployment.ClusterDescriptor
 import org.apache.flink.client.program.ClusterClient
@@ -28,6 +29,7 @@ import org.apache.flink.runtime.akka.AkkaUtils
 import org.apache.flink.runtime.minicluster.{MiniCluster, MiniClusterConfiguration, StandaloneMiniCluster}
 
 import scala.collection.mutable.ArrayBuffer
+import scala.collection.JavaConverters._
 import scala.tools.nsc.Settings
 import scala.tools.nsc.interpreter._
 
@@ -273,14 +275,14 @@ object FlinkShell {
     yarnConfig.queue.foreach((queue) => args ++= Seq("-yqu", queue.toString))
     yarnConfig.slots.foreach((slots) => args ++= Seq("-ys", slots.toString))
 
-    val commandLine = CliFrontendParser.parse(
-      CliFrontendParser.getRunCommandOptions,
-      args.toArray,
-      true)
-
-    val frontend = new CliFrontend(
-      configuration,
+    val frontend = new CliFrontend(configuration,
       CliFrontend.loadCustomCommandLines(configuration, configurationDirectory))
+
+    val commandOptions = CliFrontendParser.getRunCommandOptions
+    val commandLineOptions = CliFrontendParser.mergeOptions(commandOptions,
+      frontend.getCustomCommandLineOptions());
+    val commandLine = CliFrontendParser.parse(commandLineOptions, args.toArray, true)
+
     val customCLI = frontend.getActiveCustomCommandLine(commandLine)
 
     val clusterDescriptor = customCLI.createClusterDescriptor(commandLine)