[SPARK-57803][CORE] Delete the temp file when closeAndRead fails on the fetch-to-disk path

### What changes were proposed in this pull request?

`OneForOneBlockFetcher.DownloadCallback.onComplete` (the fetch-to-disk path) calls `channel.closeAndRead()` to close the download channel and obtain a `ManagedBuffer` over the temp file, then either registers the temp file for later cleanup or deletes it. On an `IOException` from `closeAndRead()` it rethrew without any cleanup, leaking the temp file (which is registered for cleanup only after a successful read) and possibly an open channel. The sibling `onFailure` already closes the channel and deletes the temp file; the success path did not.

This wraps `closeAndRead()` so a failure deletes the temp file and closes the channel before rethrowing, matching `onFailure`. The happy path is unchanged: the buffer is handed to the listener only after `closeAndRead()` succeeds.

### Why are the changes needed?

The temp file is created on the fetching executor when a remote block exceeds `spark.maxRemoteBlockSizeFetchToMem` and is streamed to disk. Executors are long-lived, so the orphaned temp files accumulate under the local directories until the executor exits.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

New `OneForOneBlockFetcherSuite.testFetchToDiskClosesAndDeletesTempFileWhenCloseAndReadFails` stubs a `closeAndRead()` that throws and verifies the channel is closed, the temp file is deleted, and it is never registered for cleanup. It fails without the fix and passes with it. `build/sbt 'network-shuffle/testOnly *OneForOneBlockFetcherSuite'` -> 14 tests pass.

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

Generated-by: Claude Code

Closes #56920 from LuciferYang/worktree-spark-fetch-tmpfile-leak.

Authored-by: YangJie <yangjie01@baidu.com>
Signed-off-by: yangjie01 <yangjie01@baidu.com>
(cherry picked from commit 62fd9cb)
Signed-off-by: yangjie01 <yangjie01@baidu.com>

Author: LuciferYang

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockFetcher.java

@@ -364,7 +364,18 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
 
     @Override
     public void onComplete(String streamId) throws IOException {
-      listener.onBlockFetchSuccess(blockIds[chunkIndex], channel.closeAndRead());
+      ManagedBuffer buffer;
+      try {
+        buffer = channel.closeAndRead();
+      } catch (IOException e) {
+        // closeAndRead() failed (typically from the channel close) before the buffer was handed
+        // off, so the temp file was never registered for cleanup. Delete it, and close the
+        // channel in case it was left open, mirroring onFailure, instead of leaking them.
+        channel.close();
+        targetFile.delete();
+        throw e;
+      }
+      listener.onBlockFetchSuccess(blockIds[chunkIndex], buffer);
       if (!downloadFileManager.registerTempFileToClean(targetFile)) {
         targetFile.delete();
       }

common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/OneForOneBlockFetcherSuite.java

@@ -17,6 +17,7 @@
 
 package org.apache.spark.network.shuffle;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -34,14 +35,17 @@
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import org.apache.spark.network.buffer.ManagedBuffer;
 import org.apache.spark.network.buffer.NettyManagedBuffer;
 import org.apache.spark.network.buffer.NioManagedBuffer;
 import org.apache.spark.network.client.ChunkReceivedCallback;
 import org.apache.spark.network.client.RpcResponseCallback;
+import org.apache.spark.network.client.StreamCallback;
 import org.apache.spark.network.client.TransportClient;
 import org.apache.spark.network.shuffle.protocol.BlockTransferMessage;
 import org.apache.spark.network.shuffle.protocol.FetchShuffleBlocks;
@@ -284,6 +288,52 @@ public void testInvalidShuffleBlockIds() {
         new int[][] {{ 0 }}), conf));
   }
 
+  /**
+   * Verifies that when fetching to disk, a failure of
+   * {@link DownloadFileWritableChannel#closeAndRead} does not leak the temp file: the channel is
+   * closed and the temp file is deleted, and it is never registered for later cleanup.
+   */
+  @Test
+  public void testFetchToDiskClosesAndDeletesTempFileWhenCloseAndReadFails() throws IOException {
+    String blockId = "shuffle_0_0_0";
+    String[] blockIds = new String[] { blockId };
+
+    TransportClient client = mock(TransportClient.class);
+    BlockFetchingListener listener = mock(BlockFetchingListener.class);
+
+    DownloadFileWritableChannel channel = mock(DownloadFileWritableChannel.class);
+    when(channel.closeAndRead()).thenThrow(new IOException("boom"));
+    DownloadFile downloadFile = mock(DownloadFile.class);
+    when(downloadFile.openForWriting()).thenReturn(channel);
+    DownloadFileManager downloadFileManager = mock(DownloadFileManager.class);
+    when(downloadFileManager.createTempFile(any())).thenReturn(downloadFile);
+
+    OneForOneBlockFetcher fetcher = new OneForOneBlockFetcher(
+      client, "app-id", "exec-id", blockIds, listener, conf, downloadFileManager);
+
+    // Respond to the "openBlocks" RPC with a StreamHandle of a single chunk.
+    doAnswer(invocation -> {
+      RpcResponseCallback callback = (RpcResponseCallback) invocation.getArguments()[1];
+      callback.onSuccess(new StreamHandle(123, 1).toByteBuffer());
+      return null;
+    }).when(client).sendRpc(any(ByteBuffer.class), any(RpcResponseCallback.class));
+
+    // Drive the StreamCallback's onComplete, which triggers the failing closeAndRead().
+    doAnswer(invocation -> {
+      StreamCallback callback = (StreamCallback) invocation.getArguments()[1];
+      assertThrows(IOException.class, () -> callback.onComplete("stream"));
+      return null;
+    }).when(client).stream(any(), any(StreamCallback.class));
+
+    fetcher.start();
+
+    verify(channel).close();
+    verify(downloadFile).delete();
+    verify(downloadFileManager, never()).registerTempFileToClean(any());
+    // closeAndRead() failed, so the buffer must not have been handed off to the listener.
+    verify(listener, never()).onBlockFetchSuccess(any(), any());
+  }
+
   /**
    * Begins a fetch on the given set of blocks by mocking out the server side of the RPC which
    * simply returns the given (BlockId, Block) pairs.