This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 62fd9cb420db [SPARK-57803][CORE] Delete the temp file when
closeAndRead fails on the fetch-to-disk path
62fd9cb420db is described below
commit 62fd9cb420db48a3ff7edd0f18b41da6e6a0e3a5
Author: YangJie <[email protected]>
AuthorDate: Thu Jul 2 10:47:56 2026 +0800
[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 c [...]
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 <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
.../network/shuffle/OneForOneBlockFetcher.java | 13 +++++-
.../shuffle/OneForOneBlockFetcherSuite.java | 50 ++++++++++++++++++++++
2 files changed, 62 insertions(+), 1 deletion(-)
diff --git
a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockFetcher.java
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockFetcher.java
index c5c6ab313e19..5c98ae7a5b1e 100644
---
a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockFetcher.java
+++
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockFetcher.java
@@ -364,7 +364,18 @@ public class OneForOneBlockFetcher {
@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();
}
diff --git
a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/OneForOneBlockFetcherSuite.java
b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/OneForOneBlockFetcherSuite.java
index f127568c8a33..9a58ff7eaee9 100644
---
a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/OneForOneBlockFetcherSuite.java
+++
b/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;
@@ -33,14 +34,17 @@ import static org.mockito.ArgumentMatchers.anyLong;
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;
@@ -283,6 +287,52 @@ public class OneForOneBlockFetcherSuite {
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.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
