This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 1d74ee5f8c6d fix: close the outer file stream of inline reads (#19936)
1d74ee5f8c6d is described below
commit 1d74ee5f8c6d09660d2f163b0e00f7754f0d074f
Author: voonhous <[email protected]>
AuthorDate: Mon Sep 14 14:47:23 2026 +0800
fix: close the outer file stream of inline reads (#19936)
InLineFsDataInputStream wraps the outer file's FSDataInputStream in
an offset-adjusting InputStream and passes that to its super
constructor. Neither class overrides close(), so closing an inline
stream ends at InputStream.close(), a no-op, and the outer file's
stream stays open until garbage collection. Every inline HFile
read (metadata table log blocks) leaked a file handle this way.
Spark's DebugFilesystem, which tracks the streams it hands out,
reports it in TestStreamingSource as "possibly leaked file
streams" whenever it is the cached local filesystem of the JVM.
close() now closes the wrapped stream and then the outer one. A
unit test checks the outer stream is closed exactly once.
---
.../hadoop/fs/inline/InLineFsDataInputStream.java | 11 ++++++++++
.../common/fs/inline/TestInLineFileSystem.java | 25 ++++++++++++++++++++++
2 files changed, 36 insertions(+)
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFsDataInputStream.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFsDataInputStream.java
index b10b55d8de23..f74e6d693866 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFsDataInputStream.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFsDataInputStream.java
@@ -55,6 +55,17 @@ public class InLineFsDataInputStream extends
FSDataInputStream {
outerStream.seek(startOffset);
}
+ // The offset-adjusting wrapper does not own the outer stream, so the
inherited close() only
+ // closes the wrapper; the outer FSDataInputStream must be closed here or
its file handle leaks.
+ @Override
+ public void close() throws IOException {
+ try {
+ super.close();
+ } finally {
+ outerStream.close();
+ }
+ }
+
@Override
public void seek(long desired) throws IOException {
if (desired > length) {
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java
index d4ae0912256c..6d8a42fb91cd 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java
@@ -21,6 +21,7 @@ package org.apache.hudi.common.fs.inline;
import org.apache.hudi.common.testutils.FileSystemTestUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.hadoop.fs.inline.InLineFileSystem;
+import org.apache.hudi.hadoop.fs.inline.InLineFsDataInputStream;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.inline.InLineFSUtils;
@@ -40,6 +41,7 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.hudi.common.testutils.FileSystemTestUtils.RANDOM;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
@@ -221,6 +223,29 @@ public class TestInLineFileSystem {
fsDataInputStream.close();
}
+ @Test
+ public void testCloseClosesOuterStream() throws IOException {
+ OuterPathInfo outerPathInfo = generateOuterFileAndGetInfo(1000);
+ AtomicInteger outerCloseCount = new AtomicInteger(0);
+ FSDataInputStream outerStream =
outerPathInfo.outerPath.getFileSystem(conf).open(outerPathInfo.outerPath);
+ FSDataInputStream trackingOuterStream = new FSDataInputStream(outerStream)
{
+ @Override
+ public void close() throws IOException {
+ super.close();
+ outerCloseCount.incrementAndGet();
+ }
+ };
+
+ InLineFsDataInputStream inlineStream =
+ new InLineFsDataInputStream(outerPathInfo.startOffset,
trackingOuterStream, outerPathInfo.length);
+ assertEquals(outerPathInfo.expectedBytes[0] & 0xff, inlineStream.read());
+ assertEquals(0, outerCloseCount.get());
+
+ inlineStream.close();
+ // closing the inline stream must close the outer file handle, otherwise
it leaks until GC
+ assertEquals(1, outerCloseCount.get());
+ }
+
private void verifyArrayEquality(byte[] expected, int expectedOffset, int
expectedLength,
byte[] actual, int actualOffset, int
actualLength) {
assertArrayEquals(Arrays.copyOfRange(expected, expectedOffset,
expectedOffset + expectedLength),
