address Lucifer's comments

Ngone51 · Jan 9, 2025 · 721e294 · 721e294
1 parent a962e83
commit 721e294
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 14 deletions.
diff --git a/core/src/main/scala/org/apache/spark/TaskContextImpl.scala b/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
@@ -326,10 +326,7 @@ private[spark] class TaskContextImpl(
 
     TaskContext.synchronized {
       interruptIfRequired()
-
-      if (_interruptible) {
-        _interruptible = false
-      }
+      _interruptible = false
     }
     try {
       val resource = resourceBuilder

diff --git a/core/src/main/scala/org/apache/spark/scheduler/Task.scala b/core/src/main/scala/org/apache/spark/scheduler/Task.scala
@@ -239,13 +239,9 @@ private[spark] abstract class Task[T](
             taskThread.interrupt()
           }
         } else {
-          val threadToInterrupt = if (interruptThread && taskThread != null) {
-            Some(taskThread)
-          } else {
-            None
-          }
           logInfo(log"Task ${MDC(LogKeys.TASK_ID, context.taskAttemptId())} " +
             log"is currently not interruptible. ")
+          val threadToInterrupt = if (interruptThread) Option(taskThread) else None
           context.pendingInterrupt(threadToInterrupt, reason)
         }
       }

diff --git a/core/src/test/scala/org/apache/spark/JobCancellationSuite.scala b/core/src/test/scala/org/apache/spark/JobCancellationSuite.scala
@@ -739,11 +739,23 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
               open()
 
               private def dumpFile(typeName: String): Unit = {
-                val file = new File(dir, s"$typeName.$fileHint")
-                val out = new FileOutputStream(file)
-                val objOut = new ObjectOutputStream(out)
-                objOut.writeBoolean(true)
-                objOut.close()
+                var fileOut: FileOutputStream = null
+                var objOut: ObjectOutputStream = null
+                try {
+                  val file = new File(dir, s"$typeName.$fileHint")
+                  fileOut = new FileOutputStream(file)
+                  objOut = new ObjectOutputStream(fileOut)
+                  objOut.writeBoolean(true)
+                  objOut.flush()
+                } finally {
+                  if (fileOut != null) {
+                    fileOut.close()
+                  }
+                  if (objOut != null) {
+                    objOut.close()
+                  }
+                }
+
               }
 
               private def open(): Unit = {