Skip to content

Commit 91b4a2c

Browse files
xborderxborder
authored
GH-1067: Close cached HDFS FileSystem instances (#1141)
## What's Changed - This PR fixes JVM shutdown hangs after reading HDFS datasets through Arrow Java. - FileSystemDatasetFactory now tracks hdfs:// URIs used to create the factory. On close(), after releasing the native dataset factory, it best-effort closes the matching Hadoop FileSystem instances. - The Hadoop cleanup is done via reflection so Arrow Java does not add a production dependency on Hadoop. Non-HDFS URIs are ignored. Closes #1067 .
1 parent 19a3ff5 commit 91b4a2c

4 files changed

Lines changed: 273 additions & 0 deletions

File tree

‎dataset/pom.xml‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,10 @@ under the License.
5454
<artifactId>arrow-c-data</artifactId>
5555
<scope>compile</scope>
5656
</dependency>
57+
<dependency>
58+
<groupId>org.slf4j</groupId>
59+
<artifactId>slf4j-api</artifactId>
60+
</dependency>
5761
<dependency>
5862
<groupId>org.immutables</groupId>
5963
<artifactId>value-annotations</artifactId>
@@ -112,6 +116,67 @@ under the License.
112116
</exclusion>
113117
</exclusions>
114118
</dependency>
119+
<dependency>
120+
<groupId>org.apache.hadoop</groupId>
121+
<artifactId>hadoop-hdfs</artifactId>
122+
<version>${dep.hadoop.version}</version>
123+
<scope>test</scope>
124+
<exclusions>
125+
<exclusion>
126+
<groupId>commons-logging</groupId>
127+
<artifactId>commons-logging</artifactId>
128+
</exclusion>
129+
<exclusion>
130+
<groupId>log4j</groupId>
131+
<artifactId>log4j</artifactId>
132+
</exclusion>
133+
<exclusion>
134+
<groupId>org.slf4j</groupId>
135+
<artifactId>slf4j-log4j12</artifactId>
136+
</exclusion>
137+
</exclusions>
138+
</dependency>
139+
<dependency>
140+
<groupId>org.apache.hadoop</groupId>
141+
<artifactId>hadoop-hdfs</artifactId>
142+
<version>${dep.hadoop.version}</version>
143+
<type>test-jar</type>
144+
<scope>test</scope>
145+
<exclusions>
146+
<exclusion>
147+
<groupId>commons-logging</groupId>
148+
<artifactId>commons-logging</artifactId>
149+
</exclusion>
150+
<exclusion>
151+
<groupId>log4j</groupId>
152+
<artifactId>log4j</artifactId>
153+
</exclusion>
154+
<exclusion>
155+
<groupId>org.slf4j</groupId>
156+
<artifactId>slf4j-log4j12</artifactId>
157+
</exclusion>
158+
</exclusions>
159+
</dependency>
160+
<dependency>
161+
<groupId>org.apache.hadoop</groupId>
162+
<artifactId>hadoop-minicluster</artifactId>
163+
<version>${dep.hadoop.version}</version>
164+
<scope>test</scope>
165+
<exclusions>
166+
<exclusion>
167+
<groupId>commons-logging</groupId>
168+
<artifactId>commons-logging</artifactId>
169+
</exclusion>
170+
<exclusion>
171+
<groupId>log4j</groupId>
172+
<artifactId>log4j</artifactId>
173+
</exclusion>
174+
<exclusion>
175+
<groupId>org.slf4j</groupId>
176+
<artifactId>slf4j-log4j12</artifactId>
177+
</exclusion>
178+
</exclusions>
179+
</dependency>
115180
<dependency>
116181
<groupId>com.google.guava</groupId>
117182
<artifactId>guava</artifactId>

‎dataset/src/main/java/module-info.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,4 +26,5 @@
2626
requires org.apache.arrow.c;
2727
requires org.apache.arrow.memory.core;
2828
requires org.apache.arrow.vector;
29+
requires org.slf4j;
2930
}

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

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,18 +16,30 @@
1616
*/
1717
package org.apache.arrow.dataset.file;
1818

19+
import java.lang.reflect.Method;
20+
import java.net.URI;
21+
import java.net.URISyntaxException;
22+
import java.util.LinkedHashSet;
1923
import java.util.Optional;
24+
import java.util.Set;
2025
import org.apache.arrow.dataset.jni.NativeDatasetFactory;
2126
import org.apache.arrow.dataset.jni.NativeMemoryPool;
2227
import org.apache.arrow.dataset.scanner.FragmentScanOptions;
2328
import org.apache.arrow.memory.BufferAllocator;
29+
import org.slf4j.Logger;
30+
import org.slf4j.LoggerFactory;
2431

2532
/** Java binding of the C++ FileSystemDatasetFactory. */
2633
public class FileSystemDatasetFactory extends NativeDatasetFactory {
2734

35+
private static final Logger LOGGER = LoggerFactory.getLogger(FileSystemDatasetFactory.class);
36+
37+
private final Set<URI> hdfsFileSystems;
38+
2839
public FileSystemDatasetFactory(
2940
BufferAllocator allocator, NativeMemoryPool memoryPool, FileFormat format, String uri) {
3041
super(allocator, memoryPool, createNative(format, uri, Optional.empty()));
42+
this.hdfsFileSystems = toHdfsFileSystems(uri);
3143
}
3244

3345
public FileSystemDatasetFactory(
@@ -37,11 +49,13 @@ public FileSystemDatasetFactory(
3749
String uri,
3850
Optional<FragmentScanOptions> fragmentScanOptions) {
3951
super(allocator, memoryPool, createNative(format, uri, fragmentScanOptions));
52+
this.hdfsFileSystems = toHdfsFileSystems(uri);
4053
}
4154

4255
public FileSystemDatasetFactory(
4356
BufferAllocator allocator, NativeMemoryPool memoryPool, FileFormat format, String[] uris) {
4457
super(allocator, memoryPool, createNative(format, uris, Optional.empty()));
58+
this.hdfsFileSystems = toHdfsFileSystems(uris);
4559
}
4660

4761
public FileSystemDatasetFactory(
@@ -51,6 +65,68 @@ public FileSystemDatasetFactory(
5165
String[] uris,
5266
Optional<FragmentScanOptions> fragmentScanOptions) {
5367
super(allocator, memoryPool, createNative(format, uris, fragmentScanOptions));
68+
this.hdfsFileSystems = toHdfsFileSystems(uris);
69+
}
70+
71+
/**
72+
* Close this factory and release the native instance. For HDFS URIs, also closes the cached
73+
* Hadoop FileSystem to release non-daemon threads that would otherwise prevent JVM exit. See <a
74+
* href="https://github.com/apache/arrow-java/issues/1067">#1067</a>.
75+
*/
76+
@Override
77+
public synchronized void close() {
78+
try {
79+
super.close();
80+
} finally {
81+
hdfsFileSystems.forEach(FileSystemDatasetFactory::closeHadoopFileSystem);
82+
}
83+
}
84+
85+
/**
86+
* For each {@code hdfs://} URI, close the cached Hadoop FileSystem. When Arrow C++ accesses HDFS
87+
* via libhdfs, the Hadoop Java client creates cached FileSystem instances with non-daemon threads
88+
* (IPC connections, lease renewers) that prevent JVM exit. Closing the FileSystem terminates
89+
* these connections. Uses reflection to avoid a compile-time dependency on hadoop-common.
90+
*/
91+
static void closeHadoopFileSystemsIfHdfs(String... uris) {
92+
toHdfsFileSystems(uris).forEach(FileSystemDatasetFactory::closeHadoopFileSystem);
93+
}
94+
95+
private static Set<URI> toHdfsFileSystems(String... uris) {
96+
Set<URI> hdfsFileSystems = new LinkedHashSet<>();
97+
if (uris == null) {
98+
return hdfsFileSystems;
99+
}
100+
for (String uri : uris) {
101+
if (uri == null) {
102+
continue;
103+
}
104+
try {
105+
URI parsedUri = new URI(uri);
106+
if ("hdfs".equalsIgnoreCase(parsedUri.getScheme())) {
107+
hdfsFileSystems.add(
108+
new URI(parsedUri.getScheme(), parsedUri.getAuthority(), null, null, null));
109+
}
110+
} catch (URISyntaxException e) {
111+
// Ignore here; native factory creation reports invalid user URIs.
112+
}
113+
}
114+
return hdfsFileSystems;
115+
}
116+
117+
private static void closeHadoopFileSystem(URI hdfsUri) {
118+
try {
119+
Class<?> confClass = Class.forName("org.apache.hadoop.conf.Configuration");
120+
Object conf = confClass.getDeclaredConstructor().newInstance();
121+
Class<?> fsClass = Class.forName("org.apache.hadoop.fs.FileSystem");
122+
Method getMethod = fsClass.getMethod("get", URI.class, confClass);
123+
Object fs = getMethod.invoke(null, hdfsUri, conf);
124+
Method closeMethod = fsClass.getMethod("close");
125+
closeMethod.invoke(fs);
126+
} catch (Exception e) {
127+
// Best-effort cleanup; Hadoop may not be on classpath or FileSystem already closed.
128+
LOGGER.debug("Failed to close Hadoop FileSystem for {}", hdfsUri, e);
129+
}
54130
}
55131

56132
private static long createNative(

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

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,131 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.arrow.dataset.file;
18+
19+
import static org.junit.jupiter.api.Assertions.assertEquals;
20+
import static org.junit.jupiter.api.Assertions.assertFalse;
21+
import static org.junit.jupiter.api.Assertions.assertTrue;
22+
23+
import java.io.File;
24+
import java.io.IOException;
25+
import java.util.concurrent.TimeUnit;
26+
import org.apache.hadoop.conf.Configuration;
27+
import org.apache.hadoop.fs.FileSystem;
28+
import org.apache.hadoop.fs.Path;
29+
import org.apache.hadoop.hdfs.MiniDFSCluster;
30+
import org.junit.jupiter.api.AfterAll;
31+
import org.junit.jupiter.api.BeforeAll;
32+
import org.junit.jupiter.api.Test;
33+
import org.junit.jupiter.api.io.TempDir;
34+
35+
/** Regression test for <a href="https://github.com/apache/arrow-java/issues/1067">#1067</a>. */
36+
public class TestHdfsFileSystemCleanup {
37+
38+
private static final int CHILD_TIMEOUT_SECONDS = 10;
39+
40+
private static MiniDFSCluster cluster;
41+
42+
@TempDir static File clusterDir;
43+
44+
@BeforeAll
45+
static void startCluster() throws IOException {
46+
Configuration conf = new Configuration();
47+
conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, clusterDir.getAbsolutePath());
48+
cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build();
49+
cluster.waitActive();
50+
}
51+
52+
@AfterAll
53+
static void stopCluster() {
54+
if (cluster != null) {
55+
cluster.shutdown();
56+
}
57+
}
58+
59+
@Test
60+
void testJvmHangsWithoutCleanup() throws Exception {
61+
Process child = forkChildProcess(false);
62+
assertFalse(waitForExit(child), "JVM should hang when HDFS is not cleaned up");
63+
}
64+
65+
@Test
66+
void testJvmExitsWithCleanup() throws Exception {
67+
Process child = forkChildProcess(true);
68+
assertTrue(waitForExit(child), "JVM should exit when FileSystemDatasetFactory cleanup runs");
69+
assertEquals(0, child.exitValue(), "Child process should exit cleanly (exit code 0)");
70+
}
71+
72+
private boolean waitForExit(Process child) throws InterruptedException {
73+
boolean exited = child.waitFor(CHILD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
74+
if (!exited) {
75+
child.destroyForcibly();
76+
}
77+
return exited;
78+
}
79+
80+
private Process forkChildProcess(boolean withCleanup) throws IOException {
81+
String classpath = System.getProperty("java.class.path");
82+
int port = cluster.getNameNodePort();
83+
ProcessBuilder pb =
84+
new ProcessBuilder(
85+
ProcessHandle.current().info().command().orElse("java"),
86+
"-cp",
87+
classpath,
88+
HdfsClientSimulator.class.getName(),
89+
String.valueOf(port),
90+
String.valueOf(withCleanup));
91+
pb.redirectError(ProcessBuilder.Redirect.DISCARD);
92+
pb.redirectOutput(ProcessBuilder.Redirect.DISCARD);
93+
return pb.start();
94+
}
95+
96+
/** Simulates libhdfs leaving a non-daemon thread attached to an HDFS connection. */
97+
public static class HdfsClientSimulator {
98+
public static void main(String[] args) throws Exception {
99+
int port = Integer.parseInt(args[0]);
100+
boolean withCleanup = Boolean.parseBoolean(args[1]);
101+
102+
Configuration conf = new Configuration();
103+
String hdfsUri = "hdfs://localhost:" + port;
104+
conf.set("fs.defaultFS", hdfsUri);
105+
106+
FileSystem fs = FileSystem.get(conf);
107+
fs.exists(new Path("/"));
108+
109+
Thread connectionThread =
110+
new Thread(
111+
() -> {
112+
while (true) {
113+
try {
114+
fs.getFileStatus(new Path("/"));
115+
Thread.sleep(500);
116+
} catch (Exception e) {
117+
break;
118+
}
119+
}
120+
},
121+
"simulated-libhdfs-ipc-thread");
122+
connectionThread.setDaemon(false);
123+
connectionThread.start();
124+
125+
if (withCleanup) {
126+
FileSystemDatasetFactory.closeHadoopFileSystemsIfHdfs(hdfsUri);
127+
connectionThread.join(5000);
128+
}
129+
}
130+
}
131+
}

0 commit comments

Comments
 (0)