[GitHub] [flink] zjffdu commented on a change in pull request #8533: [FLINK-12596][scala-shell] Unify all kinds of cluster via ClusterClient in FlinkShell

[GitBox <gi...@apache.org>]

zjffdu commented on a change in pull request #8533: [FLINK-12596][scala-shell] Unify all kinds of cluster via ClusterClient in FlinkShell
URL: https://github.com/apache/flink/pull/8533#discussion_r287630879
 
 

 ##########
 File path: flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
 ##########
 @@ -194,47 +200,28 @@ object FlinkShell {
     val configDirectory = new File(confDirPath)
     val configuration = GlobalConfiguration.loadConfiguration(configDirectory.getAbsolutePath)
 
-    val (repl, cluster) = try {
-      val (host, port, cluster) = fetchConnectionInfo(configuration, config)
-      val conf = cluster match {
-        case Some(Left(_)) => configuration
-        case Some(Right(yarnCluster)) => yarnCluster.getFlinkConfiguration
-        case None => configuration
-      }
-
+    try {
+      val (host, port, shouldShutdownCluster, clusterClient) = fetchConnectionInfo(configuration, config)
+      val conf = clusterClient.getFlinkConfiguration
       println(s"\nConnecting to Flink cluster (host: $host, port: $port).\n")
-      val repl = bufferedReader match {
-        case Some(reader) =>
-          val out = new StringWriter()
-          new FlinkILoop(host, port, conf, config.externalJars, reader, new JPrintWriter(out))
-        case None =>
-          new FlinkILoop(host, port, conf, config.externalJars)
-      }
-
-      (repl, cluster)
+      new FlinkILoop(host, port, conf, config.externalJars, clusterClient, shouldShutdownCluster, in,
+        new JPrintWriter(out))
     } catch {
       case e: IllegalArgumentException =>
         println(s"Error: ${e.getMessage}")
         sys.exit()
     }
+  }
+
+  def startShell(config: Config, in: Option[BufferedReader], out: JPrintWriter): Unit = {
+    println("Starting Flink Shell:")
 
+    val flinkILoop = createFlinkILoop(config, in, out)
     val settings = new Settings()
     settings.usejavacp.value = true
     settings.Yreplsync.value = true
 
-    try {
-      repl.process(settings)
-    } finally {
-      repl.closeInterpreter()
-      cluster match {
-        case Some(Left(miniCluster)) => miniCluster.close()
-        case Some(Right(yarnCluster)) =>
-          yarnCluster.shutDownCluster()
-          yarnCluster.shutdown()
-        case _ =>
-      }
-    }
-
+    flinkILoop.process(settings)
 
 Review comment:
   It is not necessary to call `closeInterpreter` implicitly, as it will be called in method `process` https://github.com/scala/scala/blob/2.11.x/src/repl/scala/tools/nsc/interpreter/ILoop.scala#L998

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services