This is an automated email from the ASF dual-hosted git repository.
voonhous 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 dd23feef9c4c fix: close the filesystem's stream on rewrap (#19935)
dd23feef9c4c is described below
commit dd23feef9c4c7a507da90110c53955fa34fe6869
Author: voonhous <[email protected]>
AuthorDate: Mon Sep 14 16:37:53 2026 +0800
fix: close the filesystem's stream on rewrap (#19935)
HadoopFSUtils.getFSDataInputStream unwraps the FSDataInputStream a
filesystem returns, buffers its inner FSInputStream and hands out
the buffered stream, so closing what Hudi returns never closes the
object the filesystem created. Filesystems that track the streams
they hand out, such as Spark's test DebugFilesystem, then report
every Hudi log file read as leaked although the OS stream is
closed. TestStreamingSource aborted on exactly that ("26 possibly
leaked file streams") once it ran in a JVM where DebugFilesystem
was the cached local filesystem; in the old serial job a plain
LocalFileSystem got cached first and the check was vacuous.
The returned stream now closes the original FSDataInputStream
after the buffered wrapper; the inner stream's close is idempotent
by the Closeable contract. A unit test checks the outer object is
closed once.
---
.../org/apache/hudi/hadoop/fs/HadoopFSUtils.java | 15 +++++++-
.../apache/hudi/hadoop/fs/TestHadoopFSUtils.java | 45 ++++++++++++++++++++++
2 files changed, 59 insertions(+), 1 deletion(-)
diff --git
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
index 8c0293dbe795..251a47db4055 100644
---
a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
+++
b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
@@ -231,8 +231,21 @@ public class HadoopFSUtils {
}
if (fsDataInputStream.getWrappedStream() instanceof FSInputStream) {
+ // The buffered stream wraps the inner FSInputStream, not the
FSDataInputStream the filesystem
+ // returned. Filesystems that track the streams they hand out (Spark's
DebugFilesystem in tests)
+ // see a leak unless that outer object is closed too; the inner stream's
close is idempotent.
+ final FSDataInputStream original = fsDataInputStream;
return new TimedFSDataInputStream(convertToHadoopPath(filePath), new
FSDataInputStream(
- new BufferedFSInputStream((FSInputStream)
fsDataInputStream.getWrappedStream(), bufferSize)));
+ new BufferedFSInputStream((FSInputStream)
original.getWrappedStream(), bufferSize)) {
+ @Override
+ public void close() throws IOException {
+ try {
+ super.close();
+ } finally {
+ original.close();
+ }
+ }
+ });
}
// fsDataInputStream.getWrappedStream() maybe a BufferedFSInputStream
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
index 92e7234eab7d..901d803d2375 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/hadoop/fs/TestHadoopFSUtils.java
@@ -31,6 +31,7 @@ import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FilterFileSystem;
@@ -181,6 +182,50 @@ public class TestHadoopFSUtils {
}
}
+ /**
+ * The buffered stream built for the {@code FSInputStream} branch wraps the
inner stream, not the
+ * {@link FSDataInputStream} the filesystem returned, so closing it used to
leave that outer object open.
+ * Filesystems that track the streams they hand out - Spark's {@code
DebugFilesystem} in tests - reported
+ * every log file read as a leaked stream.
+ */
+ @Test
+ public void testGetFSDataInputStreamClosesTheFileSystemStream(@TempDir
java.nio.file.Path tempDir) throws IOException {
+ byte[] contents = new byte[] {7, 8, 9};
+ try (TrackingLocalFileSystem fs = new TrackingLocalFileSystem()) {
+ fs.initialize(URI.create("file:///"), new Configuration());
+ Path file = new Path(tempDir.resolve("tracked.log").toUri());
+ try (FSDataOutputStream out = fs.create(file)) {
+ out.write(contents);
+ }
+
+ FSDataInputStream stream =
+ HadoopFSUtils.getFSDataInputStream(fs, new
StoragePath(file.toUri()), 4096, true);
+ assertEquals(7, stream.read());
+ stream.close();
+ assertEquals(1, fs.closeCount.get(),
+ "closing the returned stream must also close the stream the
filesystem handed out");
+ }
+ }
+
+ /** A {@link LocalFileSystem} that counts the closes of the streams it hands
out. */
+ private static class TrackingLocalFileSystem extends LocalFileSystem {
+ private final AtomicInteger closeCount = new AtomicInteger();
+
+ @Override
+ public FSDataInputStream open(Path f, int bufferSize) throws IOException {
+ return new FSDataInputStream(super.open(f,
bufferSize).getWrappedStream()) {
+ @Override
+ public void close() throws IOException {
+ try {
+ super.close();
+ } finally {
+ closeCount.incrementAndGet();
+ }
+ }
+ };
+ }
+ }
+
/** A FileSystem with the reported shape: {@code getUri()} works, {@code
getScheme()} throws. */
private static FileSystem newFsWithoutGetScheme(FileSystem delegate) {
FileSystem fs = new FilterFileSystem(delegate);
