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:
[email protected]
With regards,
Apache Git Services
