[SPARK-50853][CORE] Close temp shuffle file writable channel

[ChenMichael]

What changes were proposed in this pull request?

Currently, there are two implementations of DownloadFileWritableChannel (which is used for writing data fetched to disk), SimpleDownloadWritableChannel and EncryptedDownloadWritableChannel. The latter closes the writable channel in it's implementation of closeAndRead method while the former does not. As a result, SimpleDownloadWritableChannel channel is never closed and is relying on either the finalizer in FileOutputStream or the phantom cleanable in FileDescriptor to close the file descriptor. The change in this PR is to close the channel in SimpleDownloadWritableChannel closeAndRead method.

Why are the changes needed?

Should be closing file handles when they are not needed anymore instead of relying on finalizer/cleanables to do it.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing spark tests.

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

No

[dongjoon-hyun]

Thank you for making a PR. This sounds like a bug fix. Do you think if we can make a unit test for your claims, @ChenMichael ?

Should be closing file handles when they are not needed anymore instead of relying on finalizer/cleanables to do it.

[dongjoon-hyun]

cc @mridulm

[LuciferYang]

It seems that this code path does not have sufficient test coverage. I added a log statement in the closeAndRead method and then conducted tests on the network-shuffle and core modules locally. I found that:

1. The tests in the network-shuffle module do not cover this code path
2. There are a few test cases in the core module that indirectly cover this path (such as some test cases in DistributedSuite), but it is not easy to add new assertions.

So, could you add a targeted test case in this pull request? @ChenMichael Thanks ~

[ChenMichael]

Yea, I was finding it difficult to write a unit test for this, so the way I tested was by launching spark shell, running some queries that would force shuffles and looking at the number of open file handles for the executors through lsof. I'll spend some time trying to come up with test cases for it.

[ChenMichael]

Not sure if there's a better way to demonstrate the problem, but I added a test case that is kind of showing how the channel isn't closed through fetchBlocks flow at least. I had to make a bunch of things public though. Let me know if there's a better way to test this.

[LuciferYang]

Personally, I think this test case is a bit heavy. According to the pr description, perhaps the new case just needs to ensure that after calling the closeAndRead() method on SimpleDownloadWritableChannel, the isOpen() status should return false (Before this PR, this assertion would fail.).

For example, maybe we could add a very simple case in the network-shuffle module:

@Test
    public void testIsOpenAfterCallCloseAndRead() throws IOException {
        File tempFile = File.createTempFile("test", ".tmp");
        tempFile.deleteOnExit();
        Map<String, String> confMap = new HashMap<>();
        // Perhaps some additional configuration options need to be put into the confMap.
        TransportConf shuffle = new TransportConf("shuffle", new MapConfigProvider(confMap));

        DownloadFile downloadFile = null;
        try {
            downloadFile = new SimpleDownloadFile(tempFile, shuffle);

            DownloadFileWritableChannel channel = downloadFile.openForWriting();

            // ...
            // ... Perhaps some other operations are needed.
            channel.closeAndRead();

            Assertions.assertFalse(channel.isOpen(), "Channel should be closed after closeAndRead.");
            // ... Perhaps some additional assertions need to be added.
        } finally {
            if (downloadFile != null) {
                downloadFile.delete();
            }
        }
    }

[ChenMichael]

Agreed. This is much simpler. Done

[LuciferYang]

Thank you for updating. Could you make CI happy @ChenMichael

[ChenMichael]

I looked at the failing tests and they aren't related to the changes here. They are also passing in the rerun

[LuciferYang]

+1, LGTM

[dongjoon-hyun]

+1, LGTM. Thank you, @ChenMichael and @LuciferYang .

[dongjoon-hyun]

Merged to master/4.0.

Please make a backporting PR to branch-3.5, @ChenMichael .