diff --git a/bindings/java/src/executor.rs b/bindings/java/src/executor.rs index ef4a863915d6..36730dfee9f5 100644 --- a/bindings/java/src/executor.rs +++ b/bindings/java/src/executor.rs @@ -112,10 +112,17 @@ impl Executor { pub extern "system" fn Java_org_apache_opendal_AsyncExecutor_makeTokioExecutor<'local>( mut env: EnvUnowned<'local>, _: JClass<'local>, - cores: usize, + cores: jint, ) -> jlong { env.with_env(|env| -> Result { - let executor = make_tokio_executor(env, cores)?; + if cores <= 0 { + env.throw_new( + jni_str!("java/lang/IllegalArgumentException"), + jni_str!("cores must be positive"), + )?; + return Ok(0); + } + let executor = make_tokio_executor(env, cores as usize)?; Ok(Box::into_raw(Box::new(executor)) as jlong) }) .resolve::() 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 797280b99bb1..c38e57a7653f 100644 --- a/bindings/java/src/main/java/org/apache/opendal/AsyncExecutor.java +++ b/bindings/java/src/main/java/org/apache/opendal/AsyncExecutor.java @@ -28,6 +28,13 @@ * Completion callbacks that change it must restore it before returning.

*/ public class AsyncExecutor extends NativeObject { + /** + * Create a Tokio executor with the specified number of worker threads. + * + * @param cores the number of worker threads; must be positive + * @return the executor + * @throws IllegalArgumentException if {@code cores} is zero or negative + */ public static AsyncExecutor createTokioExecutor(int cores) { return new AsyncExecutor(makeTokioExecutor(cores)); } diff --git a/bindings/java/src/test/java/org/apache/opendal/test/AsyncExecutorTest.java b/bindings/java/src/test/java/org/apache/opendal/test/AsyncExecutorTest.java index 23241c868588..abbb98ef9922 100644 --- a/bindings/java/src/test/java/org/apache/opendal/test/AsyncExecutorTest.java +++ b/bindings/java/src/test/java/org/apache/opendal/test/AsyncExecutorTest.java @@ -20,14 +20,71 @@ 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.concurrent.TimeUnit; import lombok.Cleanup; import org.apache.opendal.AsyncExecutor; import org.apache.opendal.AsyncOperator; import org.apache.opendal.ServiceConfig; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; public class AsyncExecutorTest { + @TempDir + Path tempDir; + + @Test + void testInvalidCoresDoNotTerminateJvm() 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("invalid-cores.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")), + InvalidCoresApplication.class.getName()) + .directory(tempDir.toFile()) + .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(exited).as("JVM must exit after main returns: %s", output).isTrue(); + assertThat(process.exitValue()).as(output).isZero(); + assertThat(output).contains("INVALID_CORES_REJECTED"); + } + + public static class InvalidCoresApplication { + public static void main(String[] args) { + for (int cores : new int[] {0, -1, Integer.MIN_VALUE}) { + assertThatThrownBy(() -> AsyncExecutor.createTokioExecutor(cores)) + .isInstanceOf(IllegalArgumentException.class); + } + try (final AsyncExecutor executor = AsyncExecutor.createTokioExecutor(1); + final AsyncOperator op = + AsyncOperator.of(ServiceConfig.Memory.builder().build(), executor)) { + final byte[] content = "Hello, OpenDAL!".getBytes(StandardCharsets.UTF_8); + op.write("example.txt", content).join(); + assertThat(op.read("example.txt").join()).isEqualTo(content); + } + System.out.println("INVALID_CORES_REJECTED"); + } + } + @Test void testDedicatedTokioExecutor() { final ServiceConfig.Memory memory =