feat(bindings/java): add layers onto ops (#3392)

* feat(bindings/java): support consturct ops with native layers

Signed-off-by: tison <[email protected]>

* impl the native part

Signed-off-by: tison <[email protected]>

* license header

Signed-off-by: tison <[email protected]>

* add a simple consturct test

Signed-off-by: tison <[email protected]>

* spotless

Signed-off-by: tison <[email protected]>

* change API

Signed-off-by: tison <[email protected]>

* change API

Signed-off-by: tison <[email protected]>

* save

Signed-off-by: tison <[email protected]>

* Revert "save"

This reverts commit 62f5e27.

* rename

Signed-off-by: tison <[email protected]>

* fixup

Signed-off-by: tison <[email protected]>

---------

Signed-off-by: tison <[email protected]>

bindings/java/src/layer.rs

@@ -0,0 +1,50 @@
+// 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.
+
+use std::time::Duration;
+
+use jni::objects::JClass;
+use jni::sys::{jboolean, jfloat, jlong};
+use jni::JNIEnv;
+
+use opendal::layers::RetryLayer;
+use opendal::Operator;
+
+#[no_mangle]
+pub extern "system" fn Java_org_apache_opendal_layer_RetryLayer_doLayer(
+    _: JNIEnv,
+    _: JClass,
+    op: *mut Operator,
+    jitter: jboolean,
+    factor: jfloat,
+    min_delay: jlong,
+    max_delay: jlong,
+    max_times: jlong,
+) -> jlong {
+    let op = unsafe { &*op };
+    let mut retry = RetryLayer::new();
+    retry = retry.with_factor(factor);
+    retry = retry.with_min_delay(Duration::from_nanos(min_delay as u64));
+    retry = retry.with_max_delay(Duration::from_nanos(max_delay as u64));
+    if jitter != 0 {
+        retry = retry.with_jitter()
+    }
+    if max_times >= 0 {
+        retry = retry.with_max_times(max_times as usize);
+    }
+    Box::into_raw(Box::new(op.clone().layer(retry))) as jlong
+}

bindings/java/src/lib.rs

@@ -41,6 +41,7 @@ use tokio::runtime::Runtime;
 mod blocking_operator;
 mod convert;
 mod error;
+mod layer;
 mod operator;
 mod utility;
 

bindings/java/src/main/java/org/apache/opendal/BlockingOperator.java

@@ -39,7 +39,7 @@ public class BlockingOperator extends NativeObject {
      * and see what config options each service supports.
      *
      * @param schema the name of the underneath service to access data from.
-     * @param map a map of properties to construct the underneath operator.
+     * @param map    a map of properties to construct the underneath operator.
      */
     public static BlockingOperator of(String schema, Map<String, String> map) {
         try (final Operator operator = Operator.of(schema, map)) {
@@ -54,7 +54,6 @@ public static BlockingOperator of(String schema, Map<String, String> map) {
 
     /**
      * @return the cloned blocking operator.
-     *
      * @see Operator#duplicate()
      */
     public BlockingOperator duplicate() {

bindings/java/src/main/java/org/apache/opendal/Layer.java

@@ -0,0 +1,24 @@
+/*
+ * 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;
+
+public abstract class Layer {
+    protected abstract long layer(long nativeOp);
+}

bindings/java/src/main/java/org/apache/opendal/Operator.java

@@ -120,7 +120,7 @@ public static Operator of(String schema, Map<String, String> map) {
         return new Operator(nativeHandle, info);
     }
 
-    Operator(long nativeHandle, OperatorInfo info) {
+    private Operator(long nativeHandle, OperatorInfo info) {
         super(nativeHandle);
         this.info = info;
     }
@@ -139,6 +139,11 @@ public Operator duplicate() {
         return new Operator(nativeHandle, this.info);
     }
 
+    public Operator layer(Layer layer) {
+        final long nativeHandle = layer.layer(this.nativeHandle);
+        return new Operator(nativeHandle, makeOperatorInfo(nativeHandle));
+    }
+
     public BlockingOperator blocking() {
         final long nativeHandle = makeBlockingOp(this.nativeHandle);
         final OperatorInfo info = this.info;

bindings/java/src/main/java/org/apache/opendal/layer/RetryLayer.java

@@ -0,0 +1,50 @@
+/*
+ * 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.layer;
+
+import java.time.Duration;
+import lombok.Builder;
+import org.apache.opendal.Layer;
+
+@Builder
+public class RetryLayer extends Layer {
+
+    private final boolean jitter;
+
+    @Builder.Default
+    private final float factor = 2;
+
+    @Builder.Default
+    private final Duration minDelay = Duration.ofSeconds(1);
+
+    @Builder.Default
+    private final Duration maxDelay = Duration.ofSeconds(60);
+
+    @Builder.Default
+    private final long maxTimes = 3;
+
+    @Override
+    protected long layer(long nativeOp) {
+        return doLayer(nativeOp, jitter, factor, minDelay.toNanos(), maxDelay.toNanos(), maxTimes);
+    }
+
+    private static native long doLayer(
+            long nativeHandle, boolean jitter, float factor, long minDelay, long maxDelay, long maxTimes);
+}

bindings/java/src/operator.rs

@@ -27,6 +27,7 @@ use jni::objects::JValueOwned;
 use jni::sys::jsize;
 use jni::sys::{jlong, jobject};
 use jni::JNIEnv;
+
 use opendal::layers::BlockingLayer;
 use opendal::raw::PresignedRequest;
 use opendal::Operator;

bindings/java/src/test/java/org/apache/opendal/test/LayerTest.java

@@ -0,0 +1,41 @@
+/*
+ * 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.util.HashMap;
+import java.util.Map;
+import lombok.Cleanup;
+import org.apache.opendal.Layer;
+import org.apache.opendal.Operator;
+import org.apache.opendal.layer.RetryLayer;
+import org.junit.jupiter.api.Test;
+
+public class LayerTest {
+    @Test
+    void testOperatorWithRetryLayer() {
+        final Map<String, String> conf = new HashMap<>();
+        conf.put("root", "/opendal/");
+        final Layer retryLayer = RetryLayer.builder().build();
+        @Cleanup final Operator op = Operator.of("memory", conf);
+        @Cleanup final Operator layeredOp = op.layer(retryLayer);
+        assertThat(layeredOp.info).isNotNull();
+    }
+}

bindings/java/src/test/java/org/apache/opendal/test/behavior/BehaviorExtension.java

@@ -26,9 +26,11 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
+import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.opendal.BlockingOperator;
 import org.apache.opendal.Operator;
+import org.apache.opendal.layer.RetryLayer;
 import org.junit.jupiter.api.extension.AfterAllCallback;
 import org.junit.jupiter.api.extension.BeforeAllCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
@@ -61,8 +63,9 @@ public void beforeAll(ExtensionContext context) {
                 config.put("root", root);
             }
 
-            this.operator = Operator.of(scheme, config);
-            this.blockingOperator = BlockingOperator.of(scheme, config);
+            @Cleanup final Operator op = Operator.of(scheme, config);
+            this.operator = op.layer(RetryLayer.builder().build());
+            this.blockingOperator = this.operator.blocking();
 
             this.testName = String.format("%s(%s)", context.getDisplayName(), scheme);
             log.info(