You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "changgyoopark-db (via GitHub)" <gi...@apache.org> on 2024/09/02 12:49:23 UTC
[PR] [SPARK-49492] Reattach attempted on inactive ExecutionHolder [spark]
changgyoopark-db opened a new pull request, #47955:
URL: https://github.com/apache/spark/pull/47955
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Check the status of the ExecutionHolder before reattaching it to the client.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
An ExecutionHolder may fail to spawn an ExecuteThreadRunner if an exception is thrown before doing so. In that case, reattaching the ExecutionHolder will succeed but no progress will be made.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Add a test case to ReattachableExecuteSuite.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1745840249
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutePlanHandler.scala:
##########
@@ -31,6 +32,18 @@ class SparkConnectExecutePlanHandler(responseObserver: StreamObserver[proto.Exec
try {
executeHolder.eventsManager.postStarted()
executeHolder.start()
+ } catch {
+ // Errors raised before the execution holder has finished spawning a thread are considered
+ // plan execution failure, and the client should not try reattaching it afterwards.
+ case s: SparkThrowable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw s
+ case t: Throwable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw SparkException.internalError(t.getMessage(), t)
Review Comment:
You cannot wrap an OOM in a SparkException. Please use NonFatal here.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1746001339
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutePlanHandler.scala:
##########
@@ -31,6 +32,18 @@ class SparkConnectExecutePlanHandler(responseObserver: StreamObserver[proto.Exec
try {
executeHolder.eventsManager.postStarted()
executeHolder.start()
+ } catch {
+ // Errors raised before the execution holder has finished spawning a thread are considered
+ // plan execution failure, and the client should not try reattaching it afterwards.
+ case s: SparkThrowable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw s
+ case t: Throwable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw SparkException.internalError(t.getMessage(), t)
Review Comment:
Hm.. good point, yeah, OOM is definitely not a spark error.
On the other hand, OOM is definitely not NonFatal, so not sure if wrapping it in a NonFatal error type is a good idea. Unfortunately, not wrapping OOM is also problematic because (EDGE) the client will retry (OOM is translated into UNKNOWN) and will eventually get OPERATION_NOT_FOUND <- this isn't good either.
I'll firstly remove the conversion, and address the client error issue (EDGE) separately.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #47955:
URL: https://github.com/apache/spark/pull/47955#issuecomment-2333145413
Merged to master.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1743557616
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutePlanHandler.scala:
##########
@@ -31,6 +32,18 @@ class SparkConnectExecutePlanHandler(responseObserver: StreamObserver[proto.Exec
try {
executeHolder.eventsManager.postStarted()
executeHolder.start()
+ } catch {
+ // Errors raised before the execution holder has finished spawning a thread are considered
+ // plan execution failure, and the client should not try reattaching it afterwards.
+ case s: SparkThrowable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw s
+ case t: Throwable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw SparkException.internalError(t.getMessage(), t)
Review Comment:
Hmm, how would these exceptions arise from `executeHolder.start()` or `executeHolder.eventsManager.postStarted()`? These seem like simple operations that simply boot up other async process
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1743563161
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutePlanHandler.scala:
##########
@@ -31,6 +32,18 @@ class SparkConnectExecutePlanHandler(responseObserver: StreamObserver[proto.Exec
try {
executeHolder.eventsManager.postStarted()
executeHolder.start()
+ } catch {
+ // Errors raised before the execution holder has finished spawning a thread are considered
+ // plan execution failure, and the client should not try reattaching it afterwards.
+ case s: SparkThrowable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw s
+ case t: Throwable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw SparkException.internalError(t.getMessage(), t)
Review Comment:
java.lang.OutOfMemoryError can be thrown *anywhere* heap memory is allocated (confirmed).
Ex:
```
java.lang.OutOfMemoryError: null
at java.lang.AbstractStringBuilder.hugeCapacity(AbstractStringBuilder.java:161) ~[?:1.8.0_392]
at java.lang.AbstractStringBuilder.newCapacity(AbstractStringBuilder.java:155) ~[?:1.8.0_392]
at java.lang.AbstractStringBuilder.ensureCapacityInternal(AbstractStringBuilder.java:125) ~[?:1.8.0_392]
at java.lang.AbstractStringBuilder.append(AbstractStringBuilder.java:472) ~[?:1.8.0_392]
at java.lang.AbstractStringBuilder.append(AbstractStringBuilder.java:486) ~[?:1.8.0_392]
at java.lang.StringBuilder.append(StringBuilder.java:171) ~[?:1.8.0_392]
at java.lang.StringBuilder.append(StringBuilder.java:81) ~[?:1.8.0_392]
at grpc_shaded.com.google.protobuf.TextFormat$TextGenerator.print(TextFormat.java:856) ~[----ws_3_5--connector--connect--server--spark_connect_proto_lib_shaded-hive-2.3__hadoop-3.2_2.12--458243678--com.google.protobuf__protobuf-java__3.19.6.jar:?]
...
at org.apache.spark.sql.connect.service.ExecuteEventsManager.postStarted(ExecuteEventsManager.scala:183) ~[----ws_3_5--connector--connect--server--spark_connect_proto_lib_shaded-hive-2.3__hadoop-3.2_2.12---1628521472--connect-prepare-hive-2.3__hadoop-3.2_2.12_deploy.jar:?]
at org.apache.spark.sql.connect.service.SparkConnectExecutePlanHandler.handle(SparkConnectExecutePlanHandler.scala:32) ~[----ws_3_5--connector--connect--server--spark_connect_proto_lib_shaded-hive-2.3__hadoop-3.2_2.12---1628521472--connect-prepare-hive-2.3__hadoop-3.2_2.12_deploy.jar:?]
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1743804244
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala:
##########
@@ -240,7 +240,9 @@ private[connect] class ExecuteHolder(
def close(): Unit = synchronized {
if (closedTimeMs.isEmpty) {
// interrupt execution, if still running.
- runner.interrupt()
+ if (eventsManager.status != ExecuteStatus.Pending) {
+ runner.interrupt()
+ }
Review Comment:
Sure, that will be nicer!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1743794192
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala:
##########
@@ -240,7 +240,9 @@ private[connect] class ExecuteHolder(
def close(): Unit = synchronized {
if (closedTimeMs.isEmpty) {
// interrupt execution, if still running.
- runner.interrupt()
+ if (eventsManager.status != ExecuteStatus.Pending) {
+ runner.interrupt()
+ }
Review Comment:
```
if (!Utils.isTesting) {
runner.interrupt()
}
```
Hmm, this feels like a behaviour change that would only pop up during testing.
Could we just call `postStarted` manually in this case or somehow bypass the assert failure in testing?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on PR #47955:
URL: https://github.com/apache/spark/pull/47955#issuecomment-2327392266
> An ExecutionHolder may fail to spawn an ExecuteThreadRunner if an exception is thrown before doing so.
Is there a way we can simply handle the exception and mark the thread as interrupted/failed?
Is there a race condition here in cases where the start of the `ExecuteHolder` is delayed (GC? some blocking synchronous operation?) but the client attempts to reattach (consider the case where the initial ExecutePlan connection is terminated due to a network issue)?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on PR #47955:
URL: https://github.com/apache/spark/pull/47955#issuecomment-2327183431
@vicennial could you review this change? I believe the problem is related to *some* of known hanging issues.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1743563161
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutePlanHandler.scala:
##########
@@ -31,6 +32,18 @@ class SparkConnectExecutePlanHandler(responseObserver: StreamObserver[proto.Exec
try {
executeHolder.eventsManager.postStarted()
executeHolder.start()
+ } catch {
+ // Errors raised before the execution holder has finished spawning a thread are considered
+ // plan execution failure, and the client should not try reattaching it afterwards.
+ case s: SparkThrowable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw s
+ case t: Throwable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw SparkException.internalError(t.getMessage(), t)
Review Comment:
java.lang.OutOfMemoryError can be thrown *anywhere* heap memory is allocated (confirmed).
Ex:
```java.lang.OutOfMemoryError: null
at java.lang.AbstractStringBuilder.hugeCapacity(AbstractStringBuilder.java:161) ~[?:1.8.0_392]
at java.lang.AbstractStringBuilder.newCapacity(AbstractStringBuilder.java:155) ~[?:1.8.0_392]
at java.lang.AbstractStringBuilder.ensureCapacityInternal(AbstractStringBuilder.java:125) ~[?:1.8.0_392]
at java.lang.AbstractStringBuilder.append(AbstractStringBuilder.java:472) ~[?:1.8.0_392]
at java.lang.AbstractStringBuilder.append(AbstractStringBuilder.java:486) ~[?:1.8.0_392]
at java.lang.StringBuilder.append(StringBuilder.java:171) ~[?:1.8.0_392]
at java.lang.StringBuilder.append(StringBuilder.java:81) ~[?:1.8.0_392]
at grpc_shaded.com.google.protobuf.TextFormat$TextGenerator.print(TextFormat.java:856) ~[----ws_3_5--connector--connect--server--spark_connect_proto_lib_shaded-hive-2.3__hadoop-3.2_2.12--458243678--com.google.protobuf__protobuf-java__3.19.6.jar:?]
...
at org.apache.spark.sql.connect.service.ExecuteEventsManager.postStarted(ExecuteEventsManager.scala:183) ~[----ws_3_5--connector--connect--server--spark_connect_proto_lib_shaded-hive-2.3__hadoop-3.2_2.12---1628521472--connect-prepare-hive-2.3__hadoop-3.2_2.12_deploy.jar:?]
at org.apache.spark.sql.connect.service.SparkConnectExecutePlanHandler.handle(SparkConnectExecutePlanHandler.scala:32) ~[----ws_3_5--connector--connect--server--spark_connect_proto_lib_shaded-hive-2.3__hadoop-3.2_2.12---1628521472--connect-prepare-hive-2.3__hadoop-3.2_2.12_deploy.jar:?]
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1745046946
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ReattachableExecuteSuite.scala:
##########
@@ -429,4 +429,21 @@ class ReattachableExecuteSuite extends SparkConnectServerTest {
val abandonedExecutions = manager.listAbandonedExecutions
assert(abandonedExecutions.forall(_.operationId != dummyOpId))
}
+
+ test("SPARK-49492: reattach succeeds on an inactive execution holder") {
+ val dummyOpId = UUID.randomUUID().toString
+ val dummyRequest =
+ buildExecutePlanRequest(buildPlan("select * from range(1)"), operationId = dummyOpId)
+ val manager = SparkConnectService.executionManager
+ val holder = manager.createExecuteHolder(dummyRequest)
+ holder.eventsManager.postStarted()
+ manager.removeExecuteHolder(holder.key)
+ withRawBlockingStub { stub =>
+ val reattach = stub.reattachExecute(buildReattachExecuteRequest(dummyOpId, None))
+ val e = intercept[StatusRuntimeException] {
+ reattach.hasNext()
+ }
+ assert(e.getMessage.contains("INVALID_HANDLE.OPERATION_NOT_FOUND"))
+ }
+ }
Review Comment:
I'm not sure if this is testing the right thing here as it would pass regardless of the changes proposed in `SparkConnectExecutePlanHandler` IIUC.
Is this meant more of a general test to verify that we return OPERATION_NOT_FOUND when the execute holder is removed?
If that's the case, could we add another test which would mimic hitting an exception in the try block introduced above and verify the `OPERATION_NOT_FOUND` error?
Essentially, the test would look like:
1. Create a dummy `proto.ExecutePlanRequest`
2. Call `SparkConnectExecutePlanHandler#handle` on the request
3. Have an exception thrown in the try block (perhaps via mocking / overriding one of the operations in the block)
4. Attempt to reattach to the above execution
5. verify op_not_found error
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #47955:
URL: https://github.com/apache/spark/pull/47955#issuecomment-2325790076
cc @juliuszsompolski @vicennial FYI
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on PR #47955:
URL: https://github.com/apache/spark/pull/47955#issuecomment-2327899840
> Is there a way we can simply handle the exception and mark the thread as interrupted/failed?
Sure, that will be clearer.
> Is there a race condition here in cases where the start of the `ExecuteHolder` is delayed (GC? some blocking synchronous operation?) but the client attempts to reattach (consider the case where the initial ExecutePlan connection is terminated due to a network issue)?
Correct! I'll fix the race and upload a new commit.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1743596533
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala:
##########
@@ -240,7 +240,9 @@ private[connect] class ExecuteHolder(
def close(): Unit = synchronized {
if (closedTimeMs.isEmpty) {
// interrupt execution, if still running.
- runner.interrupt()
+ if (eventsManager.status != ExecuteStatus.Pending) {
+ runner.interrupt()
+ }
Review Comment:
hm... wait, indeed this looks like test-only issue where postStarted is not at all called. I'll fix this!
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala:
##########
@@ -240,7 +240,9 @@ private[connect] class ExecuteHolder(
def close(): Unit = synchronized {
if (closedTimeMs.isEmpty) {
// interrupt execution, if still running.
- runner.interrupt()
+ if (eventsManager.status != ExecuteStatus.Pending) {
+ runner.interrupt()
+ }
Review Comment:
hm... wait, indeed this looks like a test-only issue where postStarted is not at all called. I'll fix this!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1743592986
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala:
##########
@@ -240,7 +240,9 @@ private[connect] class ExecuteHolder(
def close(): Unit = synchronized {
if (closedTimeMs.isEmpty) {
// interrupt execution, if still running.
- runner.interrupt()
+ if (eventsManager.status != ExecuteStatus.Pending) {
+ runner.interrupt()
+ }
Review Comment:
Not exactly, if eventsManager.postStarted throws an exception, the status stays at Pending, so eventsManager.postCancelled raises another exception.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1745055938
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/execution/ReattachableExecuteSuite.scala:
##########
@@ -429,4 +429,21 @@ class ReattachableExecuteSuite extends SparkConnectServerTest {
val abandonedExecutions = manager.listAbandonedExecutions
assert(abandonedExecutions.forall(_.operationId != dummyOpId))
}
+
+ test("SPARK-49492: reattach succeeds on an inactive execution holder") {
+ val dummyOpId = UUID.randomUUID().toString
+ val dummyRequest =
+ buildExecutePlanRequest(buildPlan("select * from range(1)"), operationId = dummyOpId)
+ val manager = SparkConnectService.executionManager
+ val holder = manager.createExecuteHolder(dummyRequest)
+ holder.eventsManager.postStarted()
+ manager.removeExecuteHolder(holder.key)
+ withRawBlockingStub { stub =>
+ val reattach = stub.reattachExecute(buildReattachExecuteRequest(dummyOpId, None))
+ val e = intercept[StatusRuntimeException] {
+ reattach.hasNext()
+ }
+ assert(e.getMessage.contains("INVALID_HANDLE.OPERATION_NOT_FOUND"))
+ }
+ }
Review Comment:
Thanks, I'll try that!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1743584747
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala:
##########
@@ -240,7 +240,9 @@ private[connect] class ExecuteHolder(
def close(): Unit = synchronized {
if (closedTimeMs.isEmpty) {
// interrupt execution, if still running.
- runner.interrupt()
+ if (eventsManager.status != ExecuteStatus.Pending) {
+ runner.interrupt()
+ }
Review Comment:
I believe this is needed to prevent the assert from failing during testing, yes? If so, could we use `Utils.isTesting` in the condition?
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutePlanHandler.scala:
##########
@@ -31,6 +32,18 @@ class SparkConnectExecutePlanHandler(responseObserver: StreamObserver[proto.Exec
try {
executeHolder.eventsManager.postStarted()
executeHolder.start()
+ } catch {
+ // Errors raised before the execution holder has finished spawning a thread are considered
+ // plan execution failure, and the client should not try reattaching it afterwards.
+ case s: SparkThrowable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw s
+ case t: Throwable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw SparkException.internalError(t.getMessage(), t)
Review Comment:
Ah, I see, okay 👍
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #47955: [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder
URL: https://github.com/apache/spark/pull/47955
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-49492][CONNECT] Reattach attempted on inactive ExecutionHolder [spark]
Posted by "changgyoopark-db (via GitHub)" <gi...@apache.org>.
changgyoopark-db commented on code in PR #47955:
URL: https://github.com/apache/spark/pull/47955#discussion_r1746001339
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutePlanHandler.scala:
##########
@@ -31,6 +32,18 @@ class SparkConnectExecutePlanHandler(responseObserver: StreamObserver[proto.Exec
try {
executeHolder.eventsManager.postStarted()
executeHolder.start()
+ } catch {
+ // Errors raised before the execution holder has finished spawning a thread are considered
+ // plan execution failure, and the client should not try reattaching it afterwards.
+ case s: SparkThrowable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw s
+ case t: Throwable =>
+ SparkConnectService.executionManager.removeExecuteHolder(executeHolder.key)
+ throw SparkException.internalError(t.getMessage(), t)
Review Comment:
Hm.. good point, yeah, OOM is definitely not a spark error.
On the other hand, OOM is definitely not NonFatal, so not sure if wrapping it in a NonFatal error type is a good idea. Unfortunately, not wrapping OOM is also problematic because (EDGE) the client will retry (OOM is translated into UNKNOWN) and will eventually get OPERATION_NOT_FOUND <- this isn't good either.
I'll firstly remove the second conversion, and address the client error issue (EDGE) separately.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org