[SPARK-57504][CORE] Avoid a potential NPE in IndexShuffleBlockResolver.getChecksums when the checksum file cannot be opened

core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala

@@ -576,22 +576,21 @@ private[spark] class IndexShuffleBlockResolver(
 
   private[shuffle] def getChecksums(checksumFile: File, blockNum: Int): Array[Long] = {
     if (!checksumFile.exists()) return null
-    val checksums = new ArrayBuffer[Long]
     // Read the checksums of blocks
-    var in: DataInputStream = null
     try {
-      in = new DataInputStream(new NioBufferedFileInputStream(checksumFile))
-      while (checksums.size < blockNum) {
-        checksums += in.readLong()
+      Utils.tryWithResource {
+        new DataInputStream(new NioBufferedFileInputStream(checksumFile))
+      } { in =>
+        val checksums = new ArrayBuffer[Long]
+        while (checksums.size < blockNum) {
+          checksums += in.readLong()
+        }
+        checksums.toArray
       }
     } catch {
       case _: IOException | _: EOFException =>
-        return null
-    } finally {
-      in.close()
+        null
     }
-
-    checksums.toArray
   }
 
   /**

core/src/test/scala/org/apache/spark/shuffle/sort/IndexShuffleBlockResolverSuite.scala

@@ -278,6 +278,22 @@ class IndexShuffleBlockResolverSuite extends SparkFunSuite {
     val checksumsFromFile = resolver.getChecksums(checksumFile, 10)
     assert(checksumsInMemory === checksumsFromFile)
   }
+
+  test("SPARK-57504: getChecksums returns null instead of throwing NPE when the " +
+    "checksum file cannot be opened") {
+    val resolver = new IndexShuffleBlockResolver(conf, blockManager)
+    // If opening the checksum file fails, getChecksums is meant to return null: it
+    // already catches IOException/EOFException. But the old non-null-safe
+    // `finally { in.close() }` threw NPE when the failure came from the stream
+    // constructor (`in` was still null), masking the original error. The sibling
+    // methods checkIndexAndDataFile and getMergedBlockData already handle this
+    // correctly. Force a constructor failure with a file that appears to exist but
+    // cannot be opened.
+    val unopenableChecksumFile = new File(tempDir, "missing.checksum") {
+      override def exists(): Boolean = true
+    }
+    assert(resolver.getChecksums(unopenableChecksumFile, 10) === null)
+  }
 }
 
 class SslIndexShuffleBlockResolverSuite extends IndexShuffleBlockResolverSuite {