[SPARK-53673][CONNECT][TESTS][3.5] Fix a flaky test failure in SparkSessionE2ESuite - interrupt tag caused by the usage of ForkJoinPool

[sarutak]

What changes were proposed in this pull request?

This PR backports #52417 to branch-3.5.

Different from master and branch-4.0, the SPARK-53673 doesn't seem to affect the branch-3.5 at this time because the implementation of ClientCalls#waitForNext in gRPC 1.56.0 which branch-3.5 depends on is different from the one in gRPC 1.67.1 which master and branch-4.0 depend on.
More specifically, the test doesn't go through the pass which calls ArrayBlockingQueue#take but go through this else block.
But I think it's better to backport #52417 to branch-3.5 to prevent future changes from causing that issue.

Why are the changes needed?

Just in case to prevent future changes from causing that issue.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

1. Temporarily enable the test and insert a sleep into the test like as follows

   // TODO(SPARK-48139): Re-enable `SparkSessionE2ESuite.interrupt tag`
-  ignore("interrupt tag") {
+  test("interrupt tag") {
     val session = spark
     import session.implicits._
 
@@ -204,6 +204,7 @@ class SparkSessionE2ESuite extends ConnectFunSuite with RemoteSparkSession {
         spark.clearTags() // clear for the case of thread reuse by another Future
       }
     }(executionContext)
+    Thread.sleep(1000)
     val q4 = Future {
       assert(spark.getTags() == Set())
       spark.addTag("one")

2. Run the test and confirm it passes

$ build/sbt 'connect-client-jvm/testOnly org.apache.spark.sql.SparkSessionE2ESuite -- -z "interrupt tag"'

Was this patch authored or co-authored using generative AI tooling?

No.

[dongjoon-hyun]

Thank you, @sarutak . Merged to branch-3.5.

[dongjoon-hyun]

+1, LGTM.

[sarutak]

Thank you, @dongjoon-hyun !