FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

GH-1067: Simplify HDFS cleanup handling · apache/arrow-java@0c44df3 · GitHub

/ arrow-java Public

Commit 0c44df3

Browse files
xborder
committed
GH-1067: Simplify HDFS cleanup handling
1 parent d8087ea commit 0c44df3

2 files changed

Lines changed: 34 additions & 83 deletions

File tree

‎dataset/src/main/java/org/apache/arrow/dataset/file/FileSystemDatasetFactory.java‎

Lines changed: 21 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -29,12 +29,12 @@
2929
/** Java binding of the C++ FileSystemDatasetFactory. */
3030
public class FileSystemDatasetFactory extends NativeDatasetFactory {
3131

32-
private final String[] uris;
32+
private final Set<URI> hdfsFileSystems;
3333

3434
public FileSystemDatasetFactory(
3535
BufferAllocator allocator, NativeMemoryPool memoryPool, FileFormat format, String uri) {
3636
super(allocator, memoryPool, createNative(format, uri, Optional.empty()));
37-
this.uris = uri == null ? new String[0] : new String[] {uri};
37+
this.hdfsFileSystems = toHdfsFileSystems(uri);
3838
}
3939

4040
public FileSystemDatasetFactory(
@@ -44,13 +44,13 @@ public FileSystemDatasetFactory(
4444
String uri,
4545
Optional<FragmentScanOptions> fragmentScanOptions) {
4646
super(allocator, memoryPool, createNative(format, uri, fragmentScanOptions));
47-
this.uris = uri == null ? new String[0] : new String[] {uri};
47+
this.hdfsFileSystems = toHdfsFileSystems(uri);
4848
}
4949

5050
public FileSystemDatasetFactory(
5151
BufferAllocator allocator, NativeMemoryPool memoryPool, FileFormat format, String[] uris) {
5252
super(allocator, memoryPool, createNative(format, uris, Optional.empty()));
53-
this.uris = uris == null ? new String[0] : uris.clone();
53+
this.hdfsFileSystems = toHdfsFileSystems(uris);
5454
}
5555

5656
public FileSystemDatasetFactory(
@@ -60,7 +60,7 @@ public FileSystemDatasetFactory(
6060
String[] uris,
6161
Optional<FragmentScanOptions> fragmentScanOptions) {
6262
super(allocator, memoryPool, createNative(format, uris, fragmentScanOptions));
63-
this.uris = uris == null ? new String[0] : uris.clone();
63+
this.hdfsFileSystems = toHdfsFileSystems(uris);
6464
}
6565

6666
/**
@@ -73,7 +73,7 @@ public synchronized void close() {
7373
try {
7474
super.close();
7575
} finally {
76-
closeHadoopFileSystemsIfHdfs(uris);
76+
hdfsFileSystems.forEach(FileSystemDatasetFactory::closeHadoopFileSystem);
7777
}
7878
}
7979

@@ -85,34 +85,26 @@ public synchronized void close() {
8585
* dependency on hadoop-common.
8686
*/
8787
static void closeHadoopFileSystemsIfHdfs(String... uris) {
88-
if (uris == null || uris.length == 0) {
89-
return;
90-
}
91-
Set<URI> hdfsFileSystems = new LinkedHashSet<>();
92-
for (String uri : uris) {
93-
URI hdfsUri = toHdfsFileSystemUri(uri);
94-
if (hdfsUri != null) {
95-
hdfsFileSystems.add(hdfsUri);
96-
}
97-
}
98-
for (URI hdfsUri : hdfsFileSystems) {
99-
closeHadoopFileSystem(hdfsUri);
100-
}
88+
toHdfsFileSystems(uris).forEach(FileSystemDatasetFactory::closeHadoopFileSystem);
10189
}
10290

103-
private static URI toHdfsFileSystemUri(String uri) {
104-
if (uri == null) {
105-
return null;
91+
private static Set<URI> toHdfsFileSystems(String... uris) {
92+
Set<URI> hdfsFileSystems = new LinkedHashSet<>();
93+
if (uris == null) {
94+
return hdfsFileSystems;
10695
}
107-
try {
108-
URI parsedUri = new URI(uri);
109-
if (!"hdfs".equalsIgnoreCase(parsedUri.getScheme())) {
110-
return null;
96+
for (String uri : uris) {
97+
try {
98+
URI parsedUri = new URI(uri);
99+
if ("hdfs".equalsIgnoreCase(parsedUri.getScheme())) {
100+
hdfsFileSystems.add(
101+
new URI(parsedUri.getScheme(), parsedUri.getAuthority(), null, null, null));
102+
}
103+
} catch (Exception e) {
104+
// Ignore here; native factory creation reports invalid user URIs.
111105
}
112-
return new URI(parsedUri.getScheme(), parsedUri.getAuthority(), null, null, null);
113-
} catch (Exception e) {
114-
return null;
115106
}
107+
return hdfsFileSystems;
116108
}
117109

118110
private static void closeHadoopFileSystem(URI hdfsUri) {

‎dataset/src/test/java/org/apache/arrow/dataset/file/TestHdfsFileSystemCleanup.java‎

Lines changed: 13 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -32,32 +32,18 @@
3232
import org.junit.jupiter.api.Test;
3333
import org.junit.jupiter.api.io.TempDir;
3434

35-
/**
36-
* Validates the fix for <a href="https://github.com/apache/arrow-java/issues/1067">#1067</a>: JVM
37-
* hangs after reading HDFS files via Arrow Dataset API due to non-daemon native threads.
38-
*
39-
* <p>When Arrow C++ accesses HDFS via libhdfs, the C library attaches native threads to the JVM via
40-
* {@code AttachCurrentThread}. These threads are non-daemon by default. They block on HDFS IPC
41-
* connections managed by cached {@code FileSystem} instances, preventing the JVM from exiting after
42-
* {@code main()} returns.
43-
*
44-
* <p>These tests fork a child JVM that simulates this behavior: a non-daemon thread holds an HDFS
45-
* connection. Without cleanup, the JVM hangs. With {@code FileSystem.close()} on the cached
46-
* instance (the same mechanism used by {@link FileSystemDatasetFactory#close()}), the connection is
47-
* closed and the thread exits, allowing the JVM to terminate.
48-
*/
35+
/** Regression test for <a href="https://github.com/apache/arrow-java/issues/1067">#1067</a>. */
4936
public class TestHdfsFileSystemCleanup {
5037

5138
private static final int CHILD_TIMEOUT_SECONDS = 10;
5239

5340
private static MiniDFSCluster cluster;
54-
private static Configuration conf;
5541

5642
@TempDir static File clusterDir;
5743

5844
@BeforeAll
5945
static void startCluster() throws IOException {
60-
conf = new Configuration();
46+
Configuration conf = new Configuration();
6147
conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, clusterDir.getAbsolutePath());
6248
cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build();
6349
cluster.waitActive();
@@ -70,35 +56,26 @@ static void stopCluster() {
7056
}
7157
}
7258

73-
/**
74-
* Without cleanup: a child JVM with a non-daemon thread holding an HDFS connection hangs after
75-
* main() returns because the non-daemon thread keeps the JVM alive.
76-
*/
7759
@Test
7860
void testJvmHangsWithoutCleanup() throws Exception {
7961
Process child = forkChildProcess(false);
80-
boolean exited = child.waitFor(CHILD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
81-
if (!exited) {
82-
child.destroyForcibly();
83-
}
84-
assertFalse(exited, "JVM should hang when non-daemon HDFS thread is not cleaned up");
62+
assertFalse(waitForExit(child), "JVM should hang when HDFS is not cleaned up");
8563
}
8664

87-
/**
88-
* With cleanup: closing the specific cached Hadoop FileSystem (the same mechanism used by {@link
89-
* FileSystemDatasetFactory#close()}) closes the HDFS connections, causing the non-daemon thread
90-
* to exit and allowing the JVM to terminate normally.
91-
*/
9265
@Test
9366
void testJvmExitsWithCleanup() throws Exception {
9467
Process child = forkChildProcess(true);
68+
assertTrue(
69+
waitForExit(child), "JVM should exit when FileSystemDatasetFactory cleanup runs");
70+
assertEquals(0, child.exitValue(), "Child process should exit cleanly (exit code 0)");
71+
}
72+
73+
private boolean waitForExit(Process child) throws InterruptedException {
9574
boolean exited = child.waitFor(CHILD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
9675
if (!exited) {
9776
child.destroyForcibly();
9877
}
99-
assertTrue(
100-
exited, "JVM should exit when FileSystemDatasetFactory cleanup runs before return");
101-
assertEquals(0, child.exitValue(), "Child process should exit cleanly (exit code 0)");
78+
return exited;
10279
}
10380

10481
private Process forkChildProcess(boolean withCleanup) throws IOException {
@@ -117,15 +94,7 @@ private Process forkChildProcess(boolean withCleanup) throws IOException {
11794
return pb.start();
11895
}
11996

120-
/**
121-
* Simulates the libhdfs behavior in a standalone JVM.
122-
*
123-
* <p>When Arrow C++ uses libhdfs to access HDFS, native threads are attached to the JVM via
124-
* {@code AttachCurrentThread}. These threads are non-daemon by default and block on HDFS IPC
125-
* connections. This class simulates that by creating a non-daemon thread that periodically polls
126-
* the HDFS filesystem. Without closing the cached FileSystem, the thread keeps the JVM alive
127-
* indefinitely after {@code main()} returns.
128-
*/
97+
/** Simulates libhdfs leaving a non-daemon thread attached to an HDFS connection. */
12998
public static class HdfsClientSimulator {
13099
public static void main(String[] args) throws Exception {
131100
int port = Integer.parseInt(args[0]);
@@ -135,13 +104,9 @@ public static void main(String[] args) throws Exception {
135104
String hdfsUri = "hdfs://localhost:" + port;
136105
conf.set("fs.defaultFS", hdfsUri);
137106

138-
// Get a cached FileSystem (same as libhdfs does internally)
139107
FileSystem fs = FileSystem.get(conf);
140108
fs.exists(new Path("/"));
141109

142-
// Simulate a non-daemon thread attached via AttachCurrentThread that
143-
// blocks on the HDFS IPC connection. In the real libhdfs scenario,
144-
// these are native threads that process RPC responses.
145110
Thread connectionThread =
146111
new Thread(
147112
() -> {
@@ -150,24 +115,18 @@ public static void main(String[] args) throws Exception {
150115
fs.getFileStatus(new Path("/"));
151116
Thread.sleep(500);
152117
} catch (Exception e) {
153-
// FileSystem was closed or thread interrupted — exit
154118
break;
155119
}
156120
}
157-
});
121+
},
122+
"simulated-libhdfs-ipc-thread");
158123
connectionThread.setDaemon(false);
159-
connectionThread.setName("simulated-libhdfs-ipc-thread");
160124
connectionThread.start();
161125

162126
if (withCleanup) {
163-
// Use the same helper invoked by FileSystemDatasetFactory.close().
164-
// Closing the cached FileSystem terminates IPC connections, causing
165-
// the non-daemon thread to get an IOException and exit.
166127
FileSystemDatasetFactory.closeHadoopFileSystemsIfHdfs(hdfsUri);
167128
connectionThread.join(5000);
168129
}
169-
170-
// main() returns. Without cleanup, the non-daemon thread keeps the JVM alive.
171130
}
172131
}
173132
}

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL