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 12:22:38 UTC
(opendal) branch main updated: fix(bindings/java): use defining class loader on executor threads (#8293)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 164b17524f fix(bindings/java): use defining class loader on executor threads (#8293)
164b17524f is described below
commit 164b17524fedcff7f9567a8231fceb4c1d9d2fb3
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Sun Sep 13 20:22:31 2026 +0800
fix(bindings/java): use defining class loader on executor threads (#8293)
---
bindings/java/src/async_operator.rs | 32 ++---
bindings/java/src/executor.rs | 34 ++++--
.../java/org/apache/opendal/AsyncExecutor.java | 3 +
.../java/org/apache/opendal/AsyncOperator.java | 2 +
.../org/apache/opendal/test/ClassLoaderTest.java | 134 +++++++++++++++++++++
5 files changed, 180 insertions(+), 25 deletions(-)
diff --git a/bindings/java/src/async_operator.rs b/bindings/java/src/async_operator.rs
index 43b26ada6d..cd20d1f765 100644
--- a/bindings/java/src/async_operator.rs
+++ b/bindings/java/src/async_operator.rs
@@ -67,7 +67,7 @@ fn intern_constructor(
let scheme = jstring_to_string(env, &scheme)?;
let map = jmap_to_hashmap(env, &map)?;
let op = Operator::via_iter(scheme, map)?;
- let executor = executor_or_default(executor)?;
+ let executor = executor_or_default(env, executor)?;
let ctx = op
.context()
.with_executor(opendal::Executor::with(executor));
@@ -133,7 +133,7 @@ fn intern_write(
let path = jstring_to_string(env, &path)?;
let content = env.convert_byte_array(content)?;
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.write_options(&path, content, write_opts).await;
complete_future(id, move |_env| {
result?;
@@ -173,7 +173,7 @@ fn intern_stat(
let path = jstring_to_string(env, &path)?;
let stat_opts = make_stat_options(env, &options)?;
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.stat_options(&path, stat_opts).await;
complete_future(id, move |env| make_metadata(env, result?));
});
@@ -214,7 +214,7 @@ fn intern_read(
// Clone operator handle to move into the task
let op_cloned = unsafe { &*op }.clone();
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let mut read_op = op_cloned.read_with(&path_str);
read_op = read_op.range(range);
let result = read_op.await;
@@ -253,7 +253,7 @@ fn intern_delete(
let path = jstring_to_string(env, &path)?;
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.delete(&path).await;
complete_future(id, move |_env| {
result?;
@@ -279,13 +279,13 @@ pub unsafe extern "system" fn Java_org_apache_opendal_AsyncOperator_makeBlocking
}
fn intern_make_blocking_op(
- _env: &mut Env,
+ env: &mut Env,
op: *mut Operator,
executor: *const Executor,
) -> Result<jlong> {
let op = unsafe { &mut *op };
- let op =
- executor_or_default(executor)?.enter_with(move || blocking::Operator::new(op.clone()))?;
+ let op = executor_or_default(env, executor)?
+ .enter_with(move || blocking::Operator::new(op.clone()))?;
Ok(Box::into_raw(Box::new(op)) as jlong)
}
@@ -336,7 +336,7 @@ fn intern_create_dir(
let path = jstring_to_string(env, &path)?;
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.create_dir(&path).await;
complete_future(id, move |_env| {
result?;
@@ -376,7 +376,7 @@ fn intern_copy(
let source_path = jstring_to_string(env, &source_path)?;
let target_path = jstring_to_string(env, &target_path)?;
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.copy(&source_path, &target_path).await;
complete_future(id, move |_env| {
result?;
@@ -416,7 +416,7 @@ fn intern_rename(
let source_path = jstring_to_string(env, &source_path)?;
let target_path = jstring_to_string(env, &target_path)?;
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.rename(&source_path, &target_path).await;
complete_future(id, move |_env| {
result?;
@@ -453,7 +453,7 @@ fn intern_remove_all(
let path = jstring_to_string(env, &path)?;
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.delete_with(&path).recursive(true).await;
complete_future(id, move |_env| {
result?;
@@ -492,7 +492,7 @@ fn intern_list(
let path = jstring_to_string(env, &path)?;
let list_opts = make_list_options(env, &options)?;
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.list_options(&path, list_opts).await;
complete_future(id, move |env| make_entries(env, result?));
});
@@ -544,7 +544,7 @@ fn intern_presign_read(
let path = jstring_to_string(env, &path)?;
let expire = Duration::from_nanos(expire as u64);
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.presign_read(&path, expire).await;
complete_future(id, move |env| make_presigned_request(env, result?));
});
@@ -580,7 +580,7 @@ fn intern_presign_write(
let path = jstring_to_string(env, &path)?;
let expire = Duration::from_nanos(expire as u64);
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.presign_write(&path, expire).await;
complete_future(id, move |env| make_presigned_request(env, result?));
});
@@ -617,7 +617,7 @@ fn intern_presign_stat(
let path = jstring_to_string(env, &path)?;
let expire = Duration::from_nanos(expire as u64);
- executor_or_default(executor)?.spawn(async move {
+ executor_or_default(env, executor)?.spawn(async move {
let result = op.presign_stat(&path, expire).await;
complete_future(id, move |env| make_presigned_request(env, result?));
});
diff --git a/bindings/java/src/executor.rs b/bindings/java/src/executor.rs
index a0feee9ad4..ef4a863915 100644
--- a/bindings/java/src/executor.rs
+++ b/bindings/java/src/executor.rs
@@ -28,6 +28,7 @@ use jni::JavaVM;
use jni::jni_sig;
use jni::jni_str;
use jni::objects::JClass;
+use jni::objects::JClassLoader;
use jni::objects::JObject;
use jni::objects::JValue;
use jni::sys::{jint, jlong};
@@ -113,8 +114,8 @@ pub extern "system" fn Java_org_apache_opendal_AsyncExecutor_makeTokioExecutor<'
_: JClass<'local>,
cores: usize,
) -> jlong {
- env.with_env(|_env| -> Result<jlong> {
- let executor = make_tokio_executor(cores)?;
+ env.with_env(|env| -> Result<jlong> {
+ let executor = make_tokio_executor(env, cores)?;
Ok(Box::into_raw(Box::new(executor)) as jlong)
})
.resolve::<ThrowException>()
@@ -134,7 +135,13 @@ pub unsafe extern "system" fn Java_org_apache_opendal_AsyncExecutor_disposeInter
}
}
-pub(crate) fn make_tokio_executor(cores: usize) -> Result<Executor> {
+fn make_tokio_executor(env: &mut Env, cores: usize) -> Result<Executor> {
+ // FindClass uses the declaring native method's loader on this Java caller.
+ // Retain that loader in the runtime's startup hook so native workers can
+ // resolve OpenDAL classes even when the system loader cannot see them.
+ let class = env.find_class(jni_str!("org/apache/opendal/AsyncExecutor"))?;
+ let class_loader = class.get_class_loader(env)?;
+ let class_loader = env.new_global_ref(class_loader)?;
let counter = AtomicUsize::new(0);
let executor = tokio::runtime::Builder::new_multi_thread()
.worker_threads(cores)
@@ -142,11 +149,11 @@ 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(|| {
+ .on_thread_start(move || {
// `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)
+ vm.attach_current_thread(|env| initialize_current_thread(env, &class_loader))
.expect("attach current thread must succeed");
})
.on_thread_stop(|| {
@@ -176,7 +183,7 @@ pub(crate) fn make_tokio_executor(cores: usize) -> Result<Executor> {
})
}
-fn set_current_thread_name(env: &mut Env) -> Result<()> {
+fn initialize_current_thread(env: &mut Env, class_loader: &JClassLoader) -> Result<()> {
let current_thread = env
.call_static_method(
jni_str!("java/lang/Thread"),
@@ -185,6 +192,13 @@ fn set_current_thread_name(env: &mut Env) -> Result<()> {
&[],
)?
.l()?;
+ // jni-rs consults the context class loader before falling back to FindClass.
+ env.call_method(
+ ¤t_thread,
+ jni_str!("setContextClassLoader"),
+ jni_sig!("(Ljava/lang/ClassLoader;)V"),
+ &[JValue::Object(class_loader)],
+ )?;
let thread_name = match std::thread::current().name() {
Some(thread_name) => env.new_string(thread_name)?,
None => unreachable!("thread name must be set"),
@@ -200,7 +214,7 @@ fn set_current_thread_name(env: &mut Env) -> Result<()> {
/// Clone the supplied executor, or share the runtime owned by default operators.
#[inline]
-pub(crate) fn executor_or_default(executor: *const Executor) -> Result<Executor> {
+pub(crate) fn executor_or_default(env: &mut Env, executor: *const Executor) -> Result<Executor> {
if !executor.is_null() {
// SAFETY: The caller must keep a supplied executor alive until its operators close.
return Ok(unsafe { &*executor }.clone());
@@ -213,8 +227,10 @@ pub(crate) fn executor_or_default(executor: *const Executor) -> Result<Executor>
return Ok(Executor { runtime });
}
- let executor =
- make_tokio_executor(available_parallelism().map(NonZeroUsize::get).unwrap_or(1))?;
+ let executor = make_tokio_executor(
+ env,
+ available_parallelism().map(NonZeroUsize::get).unwrap_or(1),
+ )?;
*cached = Arc::downgrade(&executor.runtime);
Ok(executor)
}
diff --git a/bindings/java/src/main/java/org/apache/opendal/AsyncExecutor.java b/bindings/java/src/main/java/org/apache/opendal/AsyncExecutor.java
index 227cb32337..797280b99b 100644
--- a/bindings/java/src/main/java/org/apache/opendal/AsyncExecutor.java
+++ b/bindings/java/src/main/java/org/apache/opendal/AsyncExecutor.java
@@ -23,6 +23,9 @@ package org.apache.opendal;
* AsyncExecutor represents an underneath OpenDAL executor that runs async tasks spawned in the Rust world.
*
* <p>If the executor is passed to construct operators, the executor must outlive the operators.</p>
+ *
+ * <p>Worker threads use the class loader that defines OpenDAL as their context class loader.
+ * Completion callbacks that change it must restore it before returning.</p>
*/
public class AsyncExecutor extends NativeObject {
public static AsyncExecutor createTokioExecutor(int cores) {
diff --git a/bindings/java/src/main/java/org/apache/opendal/AsyncOperator.java b/bindings/java/src/main/java/org/apache/opendal/AsyncOperator.java
index c428dbe2f0..c6bb4d0e79 100644
--- a/bindings/java/src/main/java/org/apache/opendal/AsyncOperator.java
+++ b/bindings/java/src/main/java/org/apache/opendal/AsyncOperator.java
@@ -37,6 +37,8 @@ import java.util.concurrent.ConcurrentHashMap;
* derived streams, and outstanding operations keep it alive. Closing the last
* operator and stream releases it after outstanding operations complete; a later
* operator creates a new default executor.</p>
+ *
+ * <p>The default executor uses the same worker context class loader behavior as {@link AsyncExecutor}.</p>
*/
public class AsyncOperator extends NativeObject {
diff --git a/bindings/java/src/test/java/org/apache/opendal/test/ClassLoaderTest.java b/bindings/java/src/test/java/org/apache/opendal/test/ClassLoaderTest.java
new file mode 100644
index 0000000000..8eaabd7d1b
--- /dev/null
+++ b/bindings/java/src/test/java/org/apache/opendal/test/ClassLoaderTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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 java.net.URL;
+import java.net.URLClassLoader;
+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.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.apache.opendal.AsyncExecutor;
+import org.apache.opendal.AsyncOperator;
+import org.apache.opendal.Entry;
+import org.apache.opendal.OpenDALException;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+public class ClassLoaderTest {
+ @TempDir
+ Path tempDir;
+
+ @ParameterizedTest
+ @ValueSource(strings = {"default", "explicit"})
+ void testAsyncOperationsWithIsolatedClassLoader(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 String tests = Paths.get(ClassLoaderTest.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI())
+ .toString();
+ final String classes = AsyncOperator.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toExternalForm();
+ final Path log = tempDir.resolve(mode + ".log");
+ // The system loader sees only test classes; OpenDAL is exclusively in the child loader.
+ final Process process = new ProcessBuilder(
+ java.toString(),
+ "-ea",
+ "-Xcheck:jni",
+ "-Djava.library.path=" + System.getProperty("java.library.path"),
+ "-cp",
+ tests,
+ Application.class.getName(),
+ classes,
+ mode)
+ .redirectErrorStream(true)
+ .redirectOutput(log.toFile())
+ .start();
+ final boolean exited;
+ try {
+ exited = process.waitFor(20, 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("ASYNC_OPERATIONS_COMPLETED");
+ assertThat(exited)
+ .as("JVM must exit after releasing the executor: %s", output)
+ .isTrue();
+ assertThat(process.exitValue()).as(output).isZero();
+ }
+
+ public static class Application {
+ public static void main(String[] args) throws Exception {
+ final URL tests =
+ Application.class.getProtectionDomain().getCodeSource().getLocation();
+ try (final URLClassLoader loader = new URLClassLoader(new URL[] {tests, new URL(args[0])}, null)) {
+ // Keep the caller's context loader unable to see OpenDAL. Native initialization
+ // must use the defining loader rather than inheriting the caller's context loader.
+ Class.forName("org.apache.opendal.test.ClassLoaderTest$Operations", true, loader)
+ .getMethod("run", String.class)
+ .invoke(null, args[1]);
+ }
+ System.out.println("ASYNC_OPERATIONS_COMPLETED");
+ }
+ }
+
+ public static class Operations {
+ public static void run(String mode) throws Exception {
+ for (int round = 0; round < 3; round++) {
+ try (final AsyncExecutor executor =
+ mode.equals("explicit") ? AsyncExecutor.createTokioExecutor(1) : null;
+ final AsyncOperator op = AsyncOperator.of("memory", Collections.emptyMap(), executor)) {
+ final byte[] content = "hello".getBytes(StandardCharsets.UTF_8);
+ op.write("example.txt", content).get(5, TimeUnit.SECONDS);
+ assert java.util.Arrays.equals(op.read("example.txt").get(5, TimeUnit.SECONDS), content);
+ assert op.stat("example.txt").get(5, TimeUnit.SECONDS).getContentLength() == content.length;
+ final List<Entry> entries = op.list("/").get(5, TimeUnit.SECONDS);
+ assert entries.size() == 1;
+ assert entries.get(0).getPath().equals("example.txt");
+ try {
+ op.read("missing.txt").get(5, TimeUnit.SECONDS);
+ throw new AssertionError("Reading a missing file must fail");
+ } catch (ExecutionException e) {
+ assert e.getCause() instanceof OpenDALException : e;
+ assert ((OpenDALException) e.getCause()).getCode() == OpenDALException.Code.NotFound;
+ }
+ // A failed operation must not poison later callbacks on the executor.
+ op.delete("example.txt").get(5, TimeUnit.SECONDS);
+ }
+ }
+ }
+ }
+}