[FLINK-17788][scala-shell] Fix yarn session support in scala shell

deepfield · Jun 11, 2020 · 6744033 · 6744033
1 parent 8ee8f29
commit 6744033
Showing 1 changed file with 6 additions and 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
@@ -222,12 +222,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 = getConfigDir(config)
 
@@ -253,6 +255,7 @@ object FlinkShell {
  .deploySessionCluster(clusterSpecification)
  .getClusterClient
  } finally {
+ executorConfig.set(DeploymentOptions.TARGET, "yarn-session")
  clusterDescriptor.close()
  }
 
@@ -265,7 +268,7 @@ object FlinkShell {
  mode: String) = {
 
  val effectiveConfig = new Configuration(flinkConfig)
- val args = parseYarnArgList(config, mode)
+ val args = parseArgList(config, mode)
 
  val configurationDirectory = getConfigDir(config)
 
@@ -284,7 +287,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 {