@@ -229,27 +229,33 @@ class CometExecIterator(
     if (!closed) {
       closed = true
 
+      // Attempt every resource's cleanup independently, so that one failure does not skip the
+      // remaining resources: this close() is the only chance to release them, since `closed` is
+      // already set and the task-completion retry is a no-op. The first failure is rethrown with
+      // any later ones attached as suppressed exceptions.
       var failure: Throwable = null
-      try {
+      def attempt(cleanup: => Unit): Unit = {
+        try {
+          cleanup
+        } catch {
+          case t: Throwable =>
+            if (failure == null) failure = t else failure.addSuppressed(t)
+        }
+      }
+
+      attempt {
         if (currentBatch != null) {
           currentBatch.close()
           currentBatch = null
         }
-        nativeUtil.close()
-        shuffleBlockIterators.values.foreach(_.close())
-      } catch {
-        case t: Throwable => failure = t
       }
+      attempt(nativeUtil.close())
+      shuffleBlockIterators.values.foreach(it => attempt(it.close()))
 
-      // Release the native plan even if the teardown above failed. Since this is the only chance
-      // to do so, skipping it would strand the native execution context, which owns this plan's
-      // task-shared memory pool reference and several JNI global refs.
-      try {
-        nativeLib.releasePlan(plan)
-      } catch {
-        case t: Throwable =>
-          if (failure == null) failure = t else failure.addSuppressed(t)
-      }
+      // Released last and exactly once, even if the teardown above failed: dropping the native
+      // execution context frees this plan's task-shared memory pool reference and several JNI
+      // global refs.
+      attempt(nativeLib.releasePlan(plan))
 
       if (failure != null) {
         throw failure
 
@@ -141,6 +141,14 @@ class CometExecIteratorLifecycleSuite extends CometTestBase {
         new CometShuffleBlockIterator(new ByteArrayInputStream(Array.emptyByteArray)) {
           override def close(): Unit = throw boom
         }
+      @volatile var laterInputClosed = false
+      val trackingBlockIter =
+        new CometShuffleBlockIterator(new ByteArrayInputStream(Array.emptyByteArray)) {
+          override def close(): Unit = {
+            laterInputClosed = true
+            super.close()
+          }
+        }
       val limitOp =
         CometExecUtils.getLimitNativePlan(Seq(PrettyAttribute("test", LongType)), 100).get
       val iter = new CometExecIterator(
@@ -151,10 +159,13 @@ class CometExecIteratorLifecycleSuite extends CometTestBase {
         nativeMetrics = CometMetricNode(Map.empty),
         numParts = 1,
         partitionIndex = 0,
-        shuffleBlockIterators = Map(0 -> throwingBlockIter))
+        shuffleBlockIterators = Map(0 -> throwingBlockIter, 1 -> trackingBlockIter))
 
       val thrown = intercept[java.io.IOException](iter.close())
       assert(thrown eq boom)
+      // One input's close failure must not skip the remaining resources: this close() is the only
+      // chance to release them, since the task-completion retry is a no-op once `closed` is set.
+      assert(laterInputClosed, "a later shuffle input was not closed after an earlier one threw")
       // The first close() must have marked the iterator closed and released the plan despite the
       // teardown failure: a second close() re-running releasePlan would free the native
       // execution context twice, and skipping the release would strand it.