You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/11/22 23:57:18 UTC
(spark) branch branch-4.1 updated: [SPARK-54363][REPL][TESTS] Access to buffer in SparkShellSuite should be synchronized between threads
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 9bf0ed144f6f [SPARK-54363][REPL][TESTS] Access to buffer in SparkShellSuite should be synchronized between threads
9bf0ed144f6f is described below
commit 9bf0ed144f6f174b2f66deec67e38556e7e97687
Author: Vlad Rozov <vr...@amazon.com>
AuthorDate: Sat Nov 22 15:57:02 2025 -0800
[SPARK-54363][REPL][TESTS] Access to buffer in SparkShellSuite should be synchronized between threads
### What changes were proposed in this pull request?
Synchronize access to `buffer` in `SparkShellSuite` using existing `lock`.
### Why are the changes needed?
`buffer` is updated on `ProcessOutputCapturer` threads and when it is access during exception handling, access to the buffer needs to be locked to avoid using partially updated buffer.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
running test
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #53074 from vrozov/SPARK-54363.
Authored-by: Vlad Rozov <vr...@amazon.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit f5a55406741304d05629fe5dc8e3d039551e5744)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../org/apache/spark/repl/SparkShellSuite.scala | 38 ++++++++++++----------
1 file changed, 21 insertions(+), 17 deletions(-)
diff --git a/repl/src/test/scala/org/apache/spark/repl/SparkShellSuite.scala b/repl/src/test/scala/org/apache/spark/repl/SparkShellSuite.scala
index 067f08cb6752..1aad74c8aefc 100644
--- a/repl/src/test/scala/org/apache/spark/repl/SparkShellSuite.scala
+++ b/repl/src/test/scala/org/apache/spark/repl/SparkShellSuite.scala
@@ -95,6 +95,25 @@ class SparkShellSuite extends SparkFunSuite {
}
}
+ def handleException(cause: Throwable): Unit = lock.synchronized {
+ val message =
+ s"""
+ |=======================
+ |SparkShellSuite failure output
+ |=======================
+ |Spark Shell command line: ${command.mkString(" ")}
+ |Exception: $cause
+ |Failed to capture next expected output "${expectedAnswers(next)}" within $timeout.
+ |
+ |${buffer.mkString("\n")}
+ |===========================
+ |End SparkShellSuite failure output
+ |===========================
+ """.stripMargin
+ logError(message, cause)
+ fail(message, cause)
+ }
+
val process = new ProcessBuilder(command: _*).start()
val stdinWriter = new OutputStreamWriter(process.getOutputStream, StandardCharsets.UTF_8)
@@ -119,23 +138,8 @@ class SparkShellSuite extends SparkFunSuite {
}
ThreadUtils.awaitResult(foundAllExpectedAnswers.future, timeoutForQuery)
log.info("Found all expected output.")
- } catch { case cause: Throwable =>
- val message =
- s"""
- |=======================
- |SparkShellSuite failure output
- |=======================
- |Spark Shell command line: ${command.mkString(" ")}
- |Exception: $cause
- |Failed to capture next expected output "${expectedAnswers(next)}" within $timeout.
- |
- |${buffer.mkString("\n")}
- |===========================
- |End SparkShellSuite failure output
- |===========================
- """.stripMargin
- logError(message, cause)
- fail(message, cause)
+ } catch {
+ case cause: Throwable => handleException(cause)
} finally {
if (!process.waitFor(1, MINUTES)) {
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org