You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2023/12/04 10:13:30 UTC
(spark) branch master updated: [SPARK-46186][CONNECT] Fix illegal state transition when ExecuteThreadRunner interrupted before started
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new be9a710efb7b [SPARK-46186][CONNECT] Fix illegal state transition when ExecuteThreadRunner interrupted before started
be9a710efb7b is described below
commit be9a710efb7b1b9565c52766728d793fb497b483
Author: Juliusz Sompolski <ju...@databricks.com>
AuthorDate: Mon Dec 4 19:13:18 2023 +0900
[SPARK-46186][CONNECT] Fix illegal state transition when ExecuteThreadRunner interrupted before started
### What changes were proposed in this pull request?
A race condition sending interrupt (or releaseSession) just after execute could cause:
```
[info] org.apache.spark.SparkException: java.lang.IllegalStateException:
[info] operationId: fa653330-587a-41c7-a4a9-7987f070dcba with status Started
[info] is not within statuses List(Finished, Failed, Canceled) for event Closed
```
and
```
[info] org.apache.spark.SparkException: java.lang.IllegalStateException:
[info] operationId: fa653330-587a-41c7-a4a9-7987f070dcba with status Closed
[info] is not within statuses List(Started, Analyzed, ReadyForExecution, Finished, Failed) for event Canceled
```
when the interrupt arrives before the thread in ExecuteThreadRunner is started.
This would cause in ExecuteHolder close:
```
runner.interrupt() <- just sets interrupted = true
runner.join() <- thread didn't start yet, so join() returns immediately, doesn't wait for thread to be actually interrupted
...
eventsManager.postClosed() <- causes the first failure, because thread wasn't running and didn't move to Canceled
```
Afterwards, assuming we allowed the transition, the thread will get started, and then want to immediately move to Canceled, notice the `interrupted` state. Then it would hit the 2nd error, not allowing Canceled after Closed.
While we could consider allowing the first transition (Started -> Closed), we don't want any events to be coming after Closed, so that listeners can clean their state after Closed.
Fix is to handle interrupts coming before the thread started, and then prevent the thread from even starting if it was interruped.
### Why are the changes needed?
This was detected after grpc 1.56 to 1.59 upgrade and causes some tests in SparkConnectServiceE2ESuite and ReattachableExecuteSuite to be flaky.
With the grpc upgrade, execute is eagerly sent to the server, and in some test we cleanup and release the session without waiting for the execution to start. This has triggered this flakiness.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Test added. The test depends on timing, so may not fail reliably but only from time to time.
### Was this patch authored or co-authored using generative AI tooling?
Github Copilot was assisting in some boilerplate auto-completion.
Generated-by: Github Copilot
Closes #44095 from juliuszsompolski/SPARK-46186.
Authored-by: Juliusz Sompolski <ju...@databricks.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../connect/execution/ExecuteThreadRunner.scala | 33 ++++++++++++++++++----
.../spark/sql/connect/service/ExecuteHolder.scala | 7 -----
.../execution/ReattachableExecuteSuite.scala | 26 +++++++++++++++++
3 files changed, 54 insertions(+), 12 deletions(-)
diff --git a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala
index 24b3c302b759..0ecdc4bdef96 100644
--- a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala
+++ b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala
@@ -42,6 +42,8 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
// forwarding of thread locals needs to be taken into account.
private val executionThread: Thread = new ExecutionThread()
+ private var started: Boolean = false
+
private var interrupted: Boolean = false
private var completed: Boolean = false
@@ -49,12 +51,21 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
private val lock = new Object
/** Launches the execution in a background thread, returns immediately. */
- def start(): Unit = {
- executionThread.start()
+ private[connect] def start(): Unit = {
+ lock.synchronized {
+ assert(!started)
+ // Do not start if already interrupted.
+ if (!interrupted) {
+ executionThread.start()
+ started = true
+ }
+ }
}
/** Joins the background execution thread after it is finished. */
- def join(): Unit = {
+ private[connect] def join(): Unit = {
+ // only called when the execution is completed or interrupted.
+ assert(completed || interrupted)
executionThread.join()
}
@@ -63,9 +74,21 @@ private[connect] class ExecuteThreadRunner(executeHolder: ExecuteHolder) extends
* @return
* true if it was not interrupted before, false if it was already interrupted or completed.
*/
- def interrupt(): Boolean = {
+ private[connect] def interrupt(): Boolean = {
lock.synchronized {
- if (!interrupted && !completed) {
+ if (!started && !interrupted) {
+ // execution thread hasn't started yet, and will not be started.
+ // handle the interrupted error here directly.
+ interrupted = true
+ ErrorUtils.handleError(
+ "execute",
+ executeHolder.responseObserver,
+ executeHolder.sessionHolder.userId,
+ executeHolder.sessionHolder.sessionId,
+ Some(executeHolder.eventsManager),
+ interrupted)(new SparkSQLException("OPERATION_CANCELED", Map.empty))
+ true
+ } else if (!interrupted && !completed) {
// checking completed prevents sending interrupt onError after onCompleted
interrupted = true
executionThread.interrupt()
diff --git a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala
index 8b910154d2f4..9e97ded5bf8a 100644
--- a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala
+++ b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala
@@ -128,13 +128,6 @@ private[connect] class ExecuteHolder(
runner.start()
}
- /**
- * Wait for the execution thread to finish and join it.
- */
- def join(): Unit = {
- runner.join()
- }
-
def addObservation(name: String, observation: Observation): Unit = synchronized {
observations += (name -> observation)
}
diff --git a/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ReattachableExecuteSuite.scala b/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ReattachableExecuteSuite.scala
index 2a6b89620886..02b75f044952 100644
--- a/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ReattachableExecuteSuite.scala
+++ b/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ReattachableExecuteSuite.scala
@@ -23,6 +23,7 @@ import org.scalatest.concurrent.Eventually
import org.scalatest.time.SpanSugar._
import org.apache.spark.{SparkEnv, SparkException}
+import org.apache.spark.connect.proto
import org.apache.spark.sql.connect.SparkConnectServerTest
import org.apache.spark.sql.connect.config.Connect
import org.apache.spark.sql.connect.service.SparkConnectService
@@ -296,6 +297,31 @@ class ReattachableExecuteSuite extends SparkConnectServerTest {
}
}
+ test("SPARK-46186 interrupt directly after query start") {
+ // This test depends on fast timing.
+ // If something is wrong, it can fail only from time to time.
+ withRawBlockingStub { stub =>
+ val operationId = UUID.randomUUID().toString
+ val interruptRequest = proto.InterruptRequest.newBuilder
+ .setUserContext(userContext)
+ .setSessionId(defaultSessionId)
+ .setInterruptType(proto.InterruptRequest.InterruptType.INTERRUPT_TYPE_OPERATION_ID)
+ .setOperationId(operationId)
+ .build()
+ val iter = stub.executePlan(
+ buildExecutePlanRequest(buildPlan(MEDIUM_RESULTS_QUERY), operationId = operationId))
+ // wait for execute holder to exist, but the execute thread may not have started yet.
+ Eventually.eventually(timeout(eventuallyTimeout)) {
+ assert(SparkConnectService.executionManager.listExecuteHolders.length == 1)
+ }
+ stub.interrupt(interruptRequest)
+ val e = intercept[StatusRuntimeException] {
+ while (iter.hasNext) iter.next()
+ }
+ assert(e.getMessage.contains("OPERATION_CANCELED"))
+ }
+ }
+
// A few integration tests with large results.
// They should run significantly faster than the LARGE_QUERY_TIMEOUT
// - big query (4 seconds, 871 milliseconds)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org