[flink] branch release-1.10 updated: [FLINK-17788][scala-shell] Fix yarn session support in scala shell

[kkloudas]

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 a5de333  [FLINK-17788][scala-shell] Fix yarn session support in scala 
shell
a5de333 is described below

commit a5de333d20fd56dfd77243f0a43ed4803309bfda
Author: Kostas Kloudas <kklou...@gmail.com>
AuthorDate: Wed Jun 10 15:25:39 2020 +0200

    [FLINK-17788][scala-shell] Fix yarn session support in scala shell
---
 .../src/main/scala/org/apache/flink/api/scala/FlinkShell.scala   | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

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
index d3f65c2..c136f8f 100644
--- 
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
@@ -225,12 +225,14 @@ object FlinkShell {
       case None => fetchDeployedYarnClusterInfo(config, clusterConfig, 
"default")
     }
 
+    println("Configuration: " + effectiveConfig)
+
     (effectiveConfig, clusterClient)
   }
 
   private def deployNewYarnCluster(config: Config, flinkConfig: Configuration) 
= {
     val effectiveConfig = new Configuration(flinkConfig)
-    val args = parseYarnArgList(config, "yarn-cluster")
+    val args = parseArgList(config, "yarn-cluster")
 
     val configurationDirectory =
       config.configDir.getOrElse(CliFrontend.getConfigurationDirectoryFromEnv)
@@ -257,6 +259,7 @@ object FlinkShell {
         .deploySessionCluster(clusterSpecification)
         .getClusterClient
     } finally {
+      executorConfig.set(DeploymentOptions.TARGET, "yarn-session")
       clusterDescriptor.close()
     }
 
@@ -269,7 +272,7 @@ object FlinkShell {
       mode: String) = {
 
     val effectiveConfig = new Configuration(flinkConfig)
-    val args = parseYarnArgList(config, mode)
+    val args = parseArgList(config, mode)
 
     val configurationDirectory =
       config.configDir.getOrElse(CliFrontend.getConfigurationDirectoryFromEnv)
@@ -289,7 +292,7 @@ object FlinkShell {
     (executorConfig, None)
   }
 
-  def parseYarnArgList(config: Config, mode: String): Array[String] = {
+  def parseArgList(config: Config, mode: String): Array[String] = {
     val args = if (mode == "default") {
       ArrayBuffer[String]()
     } else {