Skip to content

[SPARK-48139][CONNECT][TESTS] Try stabilising multi-thread tests in CI #48622

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
xupefei wants to merge 6 commits into from
Closed

[SPARK-48139][CONNECT][TESTS] Try stabilising multi-thread tests in CI #48622

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
63b1fbf
try stabilising threading tests
xupefei Oct 23, 2024
832c097
Trigger Build
xupefei Oct 24, 2024
73dac13
Trigger Build
xupefei Oct 25, 2024
e69bbdd
Trigger Build
xupefei Oct 25, 2024
aa75ac3
Trigger Build
xupefei Oct 28, 2024
d969c92
Trigger Build
xupefei Oct 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 3 additions & 4 deletions connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/SparkSessionE2ESuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
*/
package org.apache.spark.sql

import java.util.concurrent.ForkJoinPool
import java.util.concurrent.Executors

import scala.collection.mutable
import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
Expand Down Expand Up @@ -137,15 +137,14 @@ class SparkSessionE2ESuite extends ConnectFunSuite with RemoteSparkSession {
assert(interrupted.length == 2, s"Interrupted operations: $interrupted.")
}

// TODO(SPARK-48139): Re-enable `SparkSessionE2ESuite.interrupt tag`
ignore("interrupt tag") {
test("interrupt tag") {
Copy link
Member

@dongjoon-hyun dongjoon-hyun Oct 31, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For the record, unfortunately, it turns out this doesn't help.

[info] - interrupt tag *** FAILED *** (1 minute)
[info]   Expected exception org.apache.spark.SparkException to be thrown, but java.util.concurrent.TimeoutException was thrown (SparkSessionE2ESuite.scala:250)

val session = spark
import session.implicits._

// global ExecutionContext has only 2 threads in Apache Spark CI
// create own thread pool for four Futures used in this test
val numThreads = 4
val fpool = new ForkJoinPool(numThreads)
val fpool = Executors.newFixedThreadPool(numThreads)
val executionContext = ExecutionContext.fromExecutorService(fpool)

val q1 = Future {
Expand Down
4 changes: 2 additions & 2 deletions core/src/test/scala/org/apache/spark/JobCancellationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.spark

import java.util.concurrent.{Semaphore, TimeUnit}
import java.util.concurrent.{Executors, Semaphore, TimeUnit}
import java.util.concurrent.atomic.AtomicInteger

import scala.collection.mutable.ArrayBuffer
Expand Down Expand Up @@ -302,7 +302,7 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
// global ExecutionContext has only 2 threads in Apache Spark CI
// create own thread pool for four Futures used in this test
val numThreads = 4
val fpool = ThreadUtils.newForkJoinPool("job-tags-test-thread-pool", numThreads)
val fpool = Executors.newFixedThreadPool(numThreads)
val executionContext = ExecutionContext.fromExecutorService(fpool)

try {
Expand Down
4 changes: 2 additions & 2 deletions ...core/src/test/scala/org/apache/spark/sql/SparkSessionJobTaggingAndCancellationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.spark.sql

import java.util.concurrent.{ConcurrentHashMap, Semaphore, TimeUnit}
import java.util.concurrent.{ConcurrentHashMap, Executors, Semaphore, TimeUnit}
import java.util.concurrent.atomic.AtomicInteger

import scala.concurrent.{ExecutionContext, Future}
Expand Down Expand Up @@ -121,7 +121,7 @@ class SparkSessionJobTaggingAndCancellationSuite
// global ExecutionContext has only 2 threads in Apache Spark CI
// create own thread pool for four Futures used in this test
val numThreads = 3
val fpool = ThreadUtils.newForkJoinPool("job-tags-test-thread-pool", numThreads)
val fpool = Executors.newFixedThreadPool(numThreads)
val executionContext = ExecutionContext.fromExecutorService(fpool)

try {
Expand Down