You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by hv...@apache.org on 2023/01/31 01:49:36 UTC
[spark] branch branch-3.4 updated: [SPARK-42202][CONNECT][TEST] Improve the E2E test server stop logic
This is an automated email from the ASF dual-hosted git repository.
hvanhovell pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new fb2274a166c [SPARK-42202][CONNECT][TEST] Improve the E2E test server stop logic
fb2274a166c is described below
commit fb2274a166c65ddbd282d303ff30033c3c5439d0
Author: Zhen Li <zh...@users.noreply.github.com>
AuthorDate: Mon Jan 30 21:49:03 2023 -0400
[SPARK-42202][CONNECT][TEST] Improve the E2E test server stop logic
### What changes were proposed in this pull request?
Stop the E2E test server without leaving bad error messages due to server being killed.
### Why are the changes needed?
Improvement to better testing experience.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Existing tests
Closes #39760 from zhenlineo/e2e.
Authored-by: Zhen Li <zh...@users.noreply.github.com>
Signed-off-by: Herman van Hovell <he...@databricks.com>
(cherry picked from commit 3a0ec35c72410e748a435401981a42deade8f65a)
Signed-off-by: Herman van Hovell <he...@databricks.com>
---
.../connect/client/util/RemoteSparkSession.scala | 38 +++++++++-----
.../sql/connect/SimpleSparkConnectService.scala | 61 ++++++++++++++++++++++
.../sql/connect/service/SparkConnectService.scala | 9 +++-
3 files changed, 92 insertions(+), 16 deletions(-)
diff --git a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/util/RemoteSparkSession.scala b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/util/RemoteSparkSession.scala
index 4266b84ac9b..552799d5229 100644
--- a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/util/RemoteSparkSession.scala
+++ b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/util/RemoteSparkSession.scala
@@ -67,31 +67,32 @@ object SparkConnectServerUtils {
@volatile private var stopped = false
+ private var consoleOut: BufferedOutputStream = _
+ private val serverStopCommand = "q"
+
private lazy val sparkConnect: Process = {
debug("Starting the Spark Connect Server...")
val jar = findSparkConnectJar
val builder = Process(
- new File(sparkHome, "bin/spark-shell").getCanonicalPath,
Seq(
- "--jars",
- jar,
+ "bin/spark-submit",
"--driver-class-path",
jar,
"--conf",
- "spark.plugins=org.apache.spark.sql.connect.SparkConnectPlugin",
- "--conf",
- s"spark.connect.grpc.binding.port=$port"))
+ s"spark.connect.grpc.binding.port=$port",
+ "--class",
+ "org.apache.spark.sql.connect.SimpleSparkConnectService",
+ jar),
+ new File(sparkHome))
val io = new ProcessIO(
- // Hold the input channel to the spark console to keep the console open
- in => new BufferedOutputStream(in),
- // Only redirect output if debug to avoid channel interruption error on process termination.
- out => if (isDebug) Source.fromInputStream(out).getLines.foreach(debug),
- err => if (isDebug) Source.fromInputStream(err).getLines.foreach(debug))
+ in => consoleOut = new BufferedOutputStream(in),
+ out => Source.fromInputStream(out).getLines.foreach(debug),
+ err => Source.fromInputStream(err).getLines.foreach(debug))
val process = builder.run(io)
// Adding JVM shutdown hook
- sys.addShutdownHook(kill())
+ sys.addShutdownHook(stop())
process
}
@@ -100,10 +101,19 @@ object SparkConnectServerUtils {
sparkConnect
}
- def kill(): Int = {
+ def stop(): Int = {
stopped = true
debug("Stopping the Spark Connect Server...")
- sparkConnect.destroy()
+ try {
+ consoleOut.write(serverStopCommand.getBytes)
+ consoleOut.flush()
+ consoleOut.close()
+ } catch {
+ case e: Throwable =>
+ debug(e)
+ sparkConnect.destroy()
+ }
+
val code = sparkConnect.exitValue()
debug(s"Spark Connect Server is stopped with exit code: $code")
code
diff --git a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/SimpleSparkConnectService.scala b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/SimpleSparkConnectService.scala
new file mode 100644
index 00000000000..46f05389ad7
--- /dev/null
+++ b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/SimpleSparkConnectService.scala
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect
+
+import java.util.concurrent.TimeUnit
+
+import scala.io.StdIn
+import scala.sys.exit
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.connect.service.SparkConnectService
+
+/**
+ * A simple main class method to start the spark connect server as a service for client tests
+ * using spark-submit:
+ * {{{
+ * bin/spark-submit --class org.apache.spark.sql.connect.SimpleSparkConnectService
+ * }}}
+ * The service can be stopped by receiving a stop command or until the service get killed.
+ */
+private[sql] object SimpleSparkConnectService {
+ private val stopCommand = "q"
+
+ def main(args: Array[String]): Unit = {
+ while (true) {
+ val conf = new SparkConf()
+ val sparkSession = SparkSession.builder().config(conf).getOrCreate()
+ val sparkContext = sparkSession.sparkContext // init spark context
+ SparkConnectService.start()
+ // scalastyle:off println
+ println("Ready for client connections.")
+ // scalastyle:on println
+ val code = StdIn.readLine()
+ if (code == stopCommand) {
+ // scalastyle:off println
+ println("No more client connections.")
+ // scalastyle:on println
+ // Wait for 1 min for the server to stop
+ SparkConnectService.stop(Some(1), Some(TimeUnit.MINUTES))
+ sparkSession.close()
+ exit(0)
+ }
+ }
+ }
+}
diff --git a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala
index 9b05d90f1e3..258888d70f9 100644
--- a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala
+++ b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala
@@ -283,9 +283,14 @@ object SparkConnectService {
startGRPCService()
}
- def stop(): Unit = {
+ def stop(timeout: Option[Long] = None, unit: Option[TimeUnit] = None): Unit = {
if (server != null) {
- server.shutdownNow()
+ if (timeout.isDefined && unit.isDefined) {
+ server.shutdown()
+ server.awaitTermination(timeout.get, unit.get)
+ } else {
+ server.shutdownNow()
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org