You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/09/13 06:16:06 UTC
(opendal) 01/01: fix(bindings/java): detach idle executor threads from the JVM
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/java-jvm-shutdown
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit defbfbff69c791905fa09c2584ee55d502926a9f
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Sun Sep 13 14:15:36 2026 +0800
fix(bindings/java): detach idle executor threads from the JVM
---
bindings/java/src/async_operator.rs | 63 ++++++++-------
bindings/java/src/executor.rs | 44 -----------
.../apache/opendal/test/ExecutorShutdownTest.java | 92 ++++++++++++++++++++++
3 files changed, 127 insertions(+), 72 deletions(-)
diff --git a/bindings/java/src/async_operator.rs b/bindings/java/src/async_operator.rs
index 5d9417277..9306d30f0 100644
--- a/bindings/java/src/async_operator.rs
+++ b/bindings/java/src/async_operator.rs
@@ -17,6 +17,8 @@
use std::time::Duration;
+use jni::AttachConfig;
+use jni::DEFAULT_LOCAL_FRAME_CAPACITY;
use jni::Env;
use jni::EnvUnowned;
use jni::JavaVM;
@@ -27,6 +29,7 @@ use jni::objects::JClass;
use jni::objects::JObject;
use jni::objects::JString;
use jni::objects::JValue;
+use jni::strings::JNIString;
use jni::sys::jlong;
use jni::sys::jsize;
use opendal::Entry;
@@ -617,40 +620,44 @@ fn intern_presign_stat(
/// Complete the Java `CompletableFuture` identified by `id`.
///
-/// This runs on a `tokio` worker thread that was permanently attached to the
-/// JVM in `on_thread_start`, so `attach_current_thread` only pushes a scoped
-/// JNI frame here. The `build` closure turns the awaited operation result into a
-/// Java object within that frame; on error the future is completed
-/// exceptionally. All local references created during completion live and die
-/// within the frame.
+/// Attach the worker only while completing the future, so idle runtime threads
+/// do not prevent the JVM from exiting. The `build` closure turns the awaited
+/// operation result into a Java object; on error the future is completed
+/// exceptionally. All local references are released before the worker detaches.
fn complete_future<F>(id: jlong, build: F)
where
F: for<'a> FnOnce(&mut Env<'a>) -> Result<JObject<'a>>,
{
let vm = JavaVM::singleton().expect("JavaVM singleton must be initialized");
- vm.attach_current_thread(|env| -> Result<()> {
- let future = get_future(env, id)?;
- match build(env) {
- Ok(object) => {
- env.call_method(
- &future,
- jni_str!("complete"),
- jni_sig!("(Ljava/lang/Object;)Z"),
- &[JValue::Object(&object)],
- )?;
+ let thread = std::thread::current();
+ let thread_name = JNIString::from(thread.name().expect("executor thread name must be set"));
+ vm.attach_current_thread_with_config(
+ || AttachConfig::new().scoped(true).thread_name(&thread_name),
+ Some(DEFAULT_LOCAL_FRAME_CAPACITY),
+ |env| -> Result<()> {
+ let future = get_future(env, id)?;
+ match build(env) {
+ Ok(object) => {
+ env.call_method(
+ &future,
+ jni_str!("complete"),
+ jni_sig!("(Ljava/lang/Object;)Z"),
+ &[JValue::Object(&object)],
+ )?;
+ }
+ Err(err) => {
+ let exception = err.to_exception(env)?;
+ env.call_method(
+ &future,
+ jni_str!("completeExceptionally"),
+ jni_sig!("(Ljava/lang/Throwable;)Z"),
+ &[JValue::Object(&exception)],
+ )?;
+ }
}
- Err(err) => {
- let exception = err.to_exception(env)?;
- env.call_method(
- &future,
- jni_str!("completeExceptionally"),
- jni_sig!("(Ljava/lang/Throwable;)Z"),
- &[JValue::Object(&exception)],
- )?;
- }
- }
- Ok(())
- })
+ Ok(())
+ },
+ )
.expect("complete future must succeed");
}
diff --git a/bindings/java/src/executor.rs b/bindings/java/src/executor.rs
index 3f15688e7..3d3bfe55b 100644
--- a/bindings/java/src/executor.rs
+++ b/bindings/java/src/executor.rs
@@ -22,14 +22,10 @@ use std::sync::OnceLock;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread::available_parallelism;
-use jni::Env;
use jni::EnvUnowned;
use jni::JavaVM;
-use jni::jni_sig;
-use jni::jni_str;
use jni::objects::JClass;
use jni::objects::JObject;
-use jni::objects::JValue;
use jni::sys::{jint, jlong};
use tokio::task::JoinHandle;
@@ -121,24 +117,6 @@ pub(crate) fn make_tokio_executor(cores: usize) -> Result<Executor> {
let id = counter.fetch_add(1, Ordering::SeqCst);
format!("opendal-tokio-worker-{id}")
})
- .on_thread_start(|| {
- // `attach_current_thread` creates a permanent attachment; the thread
- // is detached automatically when it exits.
- let vm = JavaVM::singleton().expect("JavaVM singleton must be initialized");
- vm.attach_current_thread(set_current_thread_name)
- .expect("attach current thread must succeed");
- })
- .on_thread_stop(|| {
- // Typically, the thread attached to the JVM will be detached automatically
- // when the thread exits. However, there are some edge cases on Windows that
- // may lead to deadlocks. To mitigate this, we explicitly detach the thread here.
- //
- // See https://github.com/apache/opendal/issues/6869 and
- // https://github.com/jni-rs/jni-rs/issues/701 for more details.
- if let Ok(vm) = JavaVM::singleton() {
- let _ = vm.detach_current_thread();
- }
- })
.enable_all()
.build()
.map_err(|e| {
@@ -151,28 +129,6 @@ pub(crate) fn make_tokio_executor(cores: usize) -> Result<Executor> {
Ok(Executor::Tokio(executor))
}
-fn set_current_thread_name(env: &mut Env) -> Result<()> {
- let current_thread = env
- .call_static_method(
- jni_str!("java/lang/Thread"),
- jni_str!("currentThread"),
- jni_sig!("()Ljava/lang/Thread;"),
- &[],
- )?
- .l()?;
- let thread_name = match std::thread::current().name() {
- Some(thread_name) => env.new_string(thread_name)?,
- None => unreachable!("thread name must be set"),
- };
- env.call_method(
- ¤t_thread,
- jni_str!("setName"),
- jni_sig!("(Ljava/lang/String;)V"),
- &[JValue::Object(&thread_name)],
- )?;
- Ok(())
-}
-
/// # Panic
///
/// Crash if the executor is disposed.
diff --git a/bindings/java/src/test/java/org/apache/opendal/test/ExecutorShutdownTest.java b/bindings/java/src/test/java/org/apache/opendal/test/ExecutorShutdownTest.java
new file mode 100644
index 000000000..a5057d06c
--- /dev/null
+++ b/bindings/java/src/test/java/org/apache/opendal/test/ExecutorShutdownTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.opendal.test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+import org.apache.opendal.AsyncOperator;
+import org.apache.opendal.OpenDALException;
+import org.apache.opendal.Operator;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+public class ExecutorShutdownTest {
+ @TempDir
+ Path tempDir;
+
+ @ParameterizedTest
+ @ValueSource(strings = {"blocking", "async"})
+ void testJvmExitsAfterClosingOperator(String mode) throws Exception {
+ final String executable = System.getProperty("os.name").startsWith("Windows") ? "java.exe" : "java";
+ final Path java = Paths.get(System.getProperty("java.home"), "bin", executable);
+ final Path log = tempDir.resolve(mode + ".log");
+ final Process process = new ProcessBuilder(
+ java.toString(),
+ "-Xcheck:jni",
+ "-Djava.library.path=" + System.getProperty("java.library.path"),
+ "-cp",
+ System.getProperty("surefire.test.class.path", System.getProperty("java.class.path")),
+ Application.class.getName(),
+ mode)
+ .redirectErrorStream(true)
+ .redirectOutput(log.toFile())
+ .start();
+ final boolean exited;
+ try {
+ exited = process.waitFor(15, TimeUnit.SECONDS);
+ } finally {
+ if (process.isAlive()) {
+ process.destroyForcibly();
+ process.waitFor(5, TimeUnit.SECONDS);
+ }
+ }
+ final String output = new String(Files.readAllBytes(log), StandardCharsets.UTF_8);
+ assertThat(output).contains("OPERATORS_CLOSED");
+ assertThat(exited).as("JVM must exit after main returns: %s", output).isTrue();
+ assertThat(process.exitValue()).as(output).isZero();
+ }
+
+ public static class Application {
+ public static void main(String[] args) {
+ final byte[] content = "Hello, OpenDAL!".getBytes(StandardCharsets.UTF_8);
+ if (args[0].equals("blocking")) {
+ try (final Operator op = Operator.of("memory", Collections.emptyMap())) {
+ op.write("example.txt", content);
+ assertThat(op.read("example.txt")).isEqualTo(content);
+ }
+ } else {
+ try (final AsyncOperator op = AsyncOperator.of("memory", Collections.emptyMap())) {
+ op.write("example.txt", content).join();
+ assertThat(op.read("example.txt").join()).isEqualTo(content);
+ assertThatThrownBy(() -> op.read("missing.txt").join()).hasCauseInstanceOf(OpenDALException.class);
+ assertThat(op.read("example.txt").join()).isEqualTo(content);
+ }
+ }
+ System.out.println("OPERATORS_CLOSED");
+ }
+ }
+}