lidavidm commented on code in PR #1141:
URL: https://github.com/apache/arrow-java/pull/1141#discussion_r3852955260
##########
dataset/src/main/java/org/apache/arrow/dataset/file/FileSystemDatasetFactory.java:
##########
@@ -51,6 +60,65 @@ public FileSystemDatasetFactory(
String[] uris,
Optional<FragmentScanOptions> fragmentScanOptions) {
super(allocator, memoryPool, createNative(format, uris,
fragmentScanOptions));
+ this.hdfsFileSystems = toHdfsFileSystems(uris);
+ }
+
+ /**
+ * Close this factory and release the native instance. For HDFS URIs, also
closes the cached
+ * Hadoop FileSystem to release non-daemon threads that would otherwise
prevent JVM exit. See <a
+ * href="https://github.com/apache/arrow-java/issues/1067";>#1067</a>.
+ */
+ @Override
+ public synchronized void close() {
+ try {
+ super.close();
+ } finally {
+ hdfsFileSystems.forEach(FileSystemDatasetFactory::closeHadoopFileSystem);
+ }
Review Comment:
nit: could we use AutoCloseables here?
##########
dataset/src/main/java/org/apache/arrow/dataset/file/FileSystemDatasetFactory.java:
##########
@@ -51,6 +60,65 @@ public FileSystemDatasetFactory(
String[] uris,
Optional<FragmentScanOptions> fragmentScanOptions) {
super(allocator, memoryPool, createNative(format, uris,
fragmentScanOptions));
+ this.hdfsFileSystems = toHdfsFileSystems(uris);
+ }
+
+ /**
+ * Close this factory and release the native instance. For HDFS URIs, also
closes the cached
+ * Hadoop FileSystem to release non-daemon threads that would otherwise
prevent JVM exit. See <a
+ * href="https://github.com/apache/arrow-java/issues/1067";>#1067</a>.
+ */
+ @Override
+ public synchronized void close() {
+ try {
+ super.close();
+ } finally {
+ hdfsFileSystems.forEach(FileSystemDatasetFactory::closeHadoopFileSystem);
+ }
+ }
+
+ /**
+ * For each {@code hdfs://} URI, close the cached Hadoop FileSystem.
+ * When Arrow C++ accesses HDFS via libhdfs, the Hadoop Java client creates
cached FileSystem
+ * instances with non-daemon threads (IPC connections, lease renewers) that
prevent JVM exit.
+ * Closing the FileSystem terminates these connections. Uses reflection to
avoid a compile-time
+ * dependency on hadoop-common.
+ */
+ static void closeHadoopFileSystemsIfHdfs(String... uris) {
+
toHdfsFileSystems(uris).forEach(FileSystemDatasetFactory::closeHadoopFileSystem);
+ }
+
+ private static Set<URI> toHdfsFileSystems(String... uris) {
+ Set<URI> hdfsFileSystems = new LinkedHashSet<>();
+ if (uris == null) {
+ return hdfsFileSystems;
+ }
+ for (String uri : uris) {
+ try {
+ URI parsedUri = new URI(uri);
+ if ("hdfs".equalsIgnoreCase(parsedUri.getScheme())) {
+ hdfsFileSystems.add(
+ new URI(parsedUri.getScheme(), parsedUri.getAuthority(), null,
null, null));
+ }
+ } catch (Exception e) {
+ // Ignore here; native factory creation reports invalid user URIs.
Review Comment:
nit: could we catch a more specific exception class for invalid URIs?
##########
dataset/src/main/java/org/apache/arrow/dataset/file/FileSystemDatasetFactory.java:
##########
@@ -51,6 +60,65 @@ public FileSystemDatasetFactory(
String[] uris,
Optional<FragmentScanOptions> fragmentScanOptions) {
super(allocator, memoryPool, createNative(format, uris,
fragmentScanOptions));
+ this.hdfsFileSystems = toHdfsFileSystems(uris);
+ }
+
+ /**
+ * Close this factory and release the native instance. For HDFS URIs, also
closes the cached
+ * Hadoop FileSystem to release non-daemon threads that would otherwise
prevent JVM exit. See <a
+ * href="https://github.com/apache/arrow-java/issues/1067";>#1067</a>.
+ */
+ @Override
+ public synchronized void close() {
+ try {
+ super.close();
+ } finally {
+ hdfsFileSystems.forEach(FileSystemDatasetFactory::closeHadoopFileSystem);
+ }
+ }
+
+ /**
+ * For each {@code hdfs://} URI, close the cached Hadoop FileSystem.
+ * When Arrow C++ accesses HDFS via libhdfs, the Hadoop Java client creates
cached FileSystem
+ * instances with non-daemon threads (IPC connections, lease renewers) that
prevent JVM exit.
+ * Closing the FileSystem terminates these connections. Uses reflection to
avoid a compile-time
+ * dependency on hadoop-common.
+ */
+ static void closeHadoopFileSystemsIfHdfs(String... uris) {
+
toHdfsFileSystems(uris).forEach(FileSystemDatasetFactory::closeHadoopFileSystem);
+ }
+
+ private static Set<URI> toHdfsFileSystems(String... uris) {
+ Set<URI> hdfsFileSystems = new LinkedHashSet<>();
+ if (uris == null) {
+ return hdfsFileSystems;
+ }
+ for (String uri : uris) {
+ try {
+ URI parsedUri = new URI(uri);
+ if ("hdfs".equalsIgnoreCase(parsedUri.getScheme())) {
+ hdfsFileSystems.add(
+ new URI(parsedUri.getScheme(), parsedUri.getAuthority(), null,
null, null));
+ }
+ } catch (Exception e) {
+ // Ignore here; native factory creation reports invalid user URIs.
+ }
+ }
+ return hdfsFileSystems;
+ }
+
+ private static void closeHadoopFileSystem(URI hdfsUri) {
+ try {
+ Class<?> confClass =
Class.forName("org.apache.hadoop.conf.Configuration");
+ Object conf = confClass.getDeclaredConstructor().newInstance();
+ Class<?> fsClass = Class.forName("org.apache.hadoop.fs.FileSystem");
+ Method getMethod = fsClass.getMethod("get", URI.class, confClass);
+ Object fs = getMethod.invoke(null, hdfsUri, conf);
+ Method closeMethod = fsClass.getMethod("close");
+ closeMethod.invoke(fs);
+ } catch (Exception e) {
+ // Best-effort cleanup; Hadoop may not be on classpath or FileSystem
already closed
Review Comment:
nit: worth perhaps logging this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
