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:05 UTC
(opendal) branch xuanwo/java-jvm-shutdown created (now defbfbff6)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a change to branch xuanwo/java-jvm-shutdown
in repository https://gitbox.apache.org/repos/asf/opendal.git
at defbfbff6 fix(bindings/java): detach idle executor threads from the JVM
This branch includes the following new commits:
new defbfbff6 fix(bindings/java): detach idle executor threads from the JVM
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/01: fix(bindings/java): detach idle executor threads from the JVM
Posted by xu...@apache.org.
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");
+ }
+ }
+}