Refactor SdkClient to a concrete class with delegate implementation

common/src/main/java/org/opensearch/sdk/SdkClient.java

@@ -17,22 +17,30 @@
 import org.opensearch.OpenSearchException;
 import static org.opensearch.sdk.SdkClientUtils.unwrapAndConvertToException;
 
-public interface SdkClient {
+public class SdkClient {
+
+    private final SdkClientDelegate delegate;
+
+    public SdkClient(SdkClientDelegate delegate) {
+        this.delegate = delegate;
+    }
 
     /**
      * Create/Put/Index a data object/document into a table/index.
      * @param request A request encapsulating the data object to store
      * @param executor the executor to use for asynchronous execution
      * @return A completion stage encapsulating the response or exception
      */
-    public CompletionStage<PutDataObjectResponse> putDataObjectAsync(PutDataObjectRequest request, Executor executor);
+    public CompletionStage<PutDataObjectResponse> putDataObjectAsync(PutDataObjectRequest request, Executor executor) {
+        return delegate.putDataObjectAsync(request, executor);
+    }
 
     /**
      * Create/Put/Index a data object/document into a table/index.
      * @param request A request encapsulating the data object to store
      * @return A completion stage encapsulating the response or exception
      */
-    default CompletionStage<PutDataObjectResponse> putDataObjectAsync(PutDataObjectRequest request) {
+    public CompletionStage<PutDataObjectResponse> putDataObjectAsync(PutDataObjectRequest request) {
         return putDataObjectAsync(request, ForkJoinPool.commonPool());
     }
 
@@ -41,7 +49,7 @@ default CompletionStage<PutDataObjectResponse> putDataObjectAsync(PutDataObjectR
      * @param request A request encapsulating the data object to store
      * @return A response on success. Throws unchecked exceptions or {@link OpenSearchException} wrapping the cause on checked exception.
      */
-    default PutDataObjectResponse putDataObject(PutDataObjectRequest request) {
+    public PutDataObjectResponse putDataObject(PutDataObjectRequest request) {
         try {
             return putDataObjectAsync(request).toCompletableFuture().join();
         } catch (CompletionException e) {
@@ -56,15 +64,17 @@ default PutDataObjectResponse putDataObject(PutDataObjectRequest request) {
      * @param executor the executor to use for asynchronous execution
      * @return A completion stage encapsulating the response or exception
      */
-    public CompletionStage<GetDataObjectResponse> getDataObjectAsync(GetDataObjectRequest request, Executor executor);
+    public CompletionStage<GetDataObjectResponse> getDataObjectAsync(GetDataObjectRequest request, Executor executor) {
+        return delegate.getDataObjectAsync(request, executor);
+    }
 
     /**
      * Read/Get a data object/document from a table/index.
      *
      * @param request A request identifying the data object to retrieve
      * @return A completion stage encapsulating the response or exception
      */
-    default CompletionStage<GetDataObjectResponse> getDataObjectAsync(GetDataObjectRequest request) {
+    public CompletionStage<GetDataObjectResponse> getDataObjectAsync(GetDataObjectRequest request) {
         return getDataObjectAsync(request, ForkJoinPool.commonPool());
     }
 
@@ -73,7 +83,7 @@ default CompletionStage<GetDataObjectResponse> getDataObjectAsync(GetDataObjectR
      * @param request A request identifying the data object to retrieve
      * @return A response on success. Throws unchecked exceptions or {@link OpenSearchException} wrapping the cause on checked exception.
      */
-    default GetDataObjectResponse getDataObject(GetDataObjectRequest request) {
+    public GetDataObjectResponse getDataObject(GetDataObjectRequest request) {
         try {
             return getDataObjectAsync(request).toCompletableFuture().join();
         } catch (CompletionException e) {
@@ -88,15 +98,17 @@ default GetDataObjectResponse getDataObject(GetDataObjectRequest request) {
      * @param executor the executor to use for asynchronous execution
      * @return A completion stage encapsulating the response or exception
      */
-    public CompletionStage<UpdateDataObjectResponse> updateDataObjectAsync(UpdateDataObjectRequest request, Executor executor);
+    public CompletionStage<UpdateDataObjectResponse> updateDataObjectAsync(UpdateDataObjectRequest request, Executor executor) {
+        return delegate.updateDataObjectAsync(request, executor);
+    }
 
     /**
      * Update a data object/document in a table/index.
      *
      * @param request A request identifying the data object to update
      * @return A completion stage encapsulating the response or exception
      */
-    default CompletionStage<UpdateDataObjectResponse> updateDataObjectAsync(UpdateDataObjectRequest request) {
+    public CompletionStage<UpdateDataObjectResponse> updateDataObjectAsync(UpdateDataObjectRequest request) {
         return updateDataObjectAsync(request, ForkJoinPool.commonPool());
     }
 
@@ -105,7 +117,7 @@ default CompletionStage<UpdateDataObjectResponse> updateDataObjectAsync(UpdateDa
      * @param request A request identifying the data object to update
      * @return A response on success. Throws unchecked exceptions or {@link OpenSearchException} wrapping the cause on checked exception.
      */
-    default UpdateDataObjectResponse updateDataObject(UpdateDataObjectRequest request) {
+    public UpdateDataObjectResponse updateDataObject(UpdateDataObjectRequest request) {
         try {
             return updateDataObjectAsync(request).toCompletableFuture().join();
         } catch (CompletionException e) {
@@ -120,15 +132,17 @@ default UpdateDataObjectResponse updateDataObject(UpdateDataObjectRequest reques
      * @param executor the executor to use for asynchronous execution
      * @return A completion stage encapsulating the response or exception
      */
-    public CompletionStage<DeleteDataObjectResponse> deleteDataObjectAsync(DeleteDataObjectRequest request, Executor executor);
+    public CompletionStage<DeleteDataObjectResponse> deleteDataObjectAsync(DeleteDataObjectRequest request, Executor executor) {
+        return delegate.deleteDataObjectAsync(request, executor);
+    }
 
     /**
      * Delete a data object/document from a table/index.
      *
      * @param request A request identifying the data object to delete
      * @return A completion stage encapsulating the response or exception
      */
-    default CompletionStage<DeleteDataObjectResponse> deleteDataObjectAsync(DeleteDataObjectRequest request) {
+    public CompletionStage<DeleteDataObjectResponse> deleteDataObjectAsync(DeleteDataObjectRequest request) {
         return deleteDataObjectAsync(request, ForkJoinPool.commonPool());
     }
 
@@ -137,7 +151,7 @@ default CompletionStage<DeleteDataObjectResponse> deleteDataObjectAsync(DeleteDa
      * @param request A request identifying the data object to delete
      * @return A response on success. Throws unchecked exceptions or {@link OpenSearchException} wrapping the cause on checked exception.
      */
-    default DeleteDataObjectResponse deleteDataObject(DeleteDataObjectRequest request) {
+    public DeleteDataObjectResponse deleteDataObject(DeleteDataObjectRequest request) {
         try {
             return deleteDataObjectAsync(request).toCompletableFuture().join();
         } catch (CompletionException e) {
@@ -152,15 +166,17 @@ default DeleteDataObjectResponse deleteDataObject(DeleteDataObjectRequest reques
      * @param executor the executor to use for asynchronous execution
      * @return A completion stage encapsulating the response or exception
      */
-    public CompletionStage<SearchDataObjectResponse> searchDataObjectAsync(SearchDataObjectRequest request, Executor executor);
+    public CompletionStage<SearchDataObjectResponse> searchDataObjectAsync(SearchDataObjectRequest request, Executor executor) {
+        return delegate.searchDataObjectAsync(request, executor);
+    }
 
     /**
      * Search for data objects/documents in a table/index.
      *
      * @param request A request identifying the data objects to search for
      * @return A completion stage encapsulating the response or exception
      */
-    default CompletionStage<SearchDataObjectResponse> searchDataObjectAsync(SearchDataObjectRequest request) {
+    public CompletionStage<SearchDataObjectResponse> searchDataObjectAsync(SearchDataObjectRequest request) {
         return searchDataObjectAsync(request, ForkJoinPool.commonPool());
     }
 
@@ -169,11 +185,19 @@ default CompletionStage<SearchDataObjectResponse> searchDataObjectAsync(SearchDa
      * @param request A request identifying the data objects to search for
      * @return A response on success. Throws unchecked exceptions or {@link OpenSearchException} wrapping the cause on checked exception.
      */
-    default SearchDataObjectResponse searchDataObject(SearchDataObjectRequest request) {
+    public SearchDataObjectResponse searchDataObject(SearchDataObjectRequest request) {
         try {
             return searchDataObjectAsync(request).toCompletableFuture().join();
         } catch (CompletionException e) {
             throw ExceptionsHelper.convertToRuntime(unwrapAndConvertToException(e));
         }
     }
+
+    /**
+     * Get the delegate client implementation. 
+     * @return the delegate implementation
+     */
+    public SdkClientDelegate getDelegate() {
+        return delegate;
+    }
 }

common/src/main/java/org/opensearch/sdk/SdkClientDelegate.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.sdk;
+
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
+
+public interface SdkClientDelegate {
+
+    /**
+     * Create/Put/Index a data object/document into a table/index.
+     * @param request A request encapsulating the data object to store
+     * @param executor the executor to use for asynchronous execution
+     * @return A completion stage encapsulating the response or exception
+     */
+    CompletionStage<PutDataObjectResponse> putDataObjectAsync(PutDataObjectRequest request, Executor executor);
+
+    /**
+     * Read/Get a data object/document from a table/index.
+     *
+     * @param request  A request identifying the data object to retrieve
+     * @param executor the executor to use for asynchronous execution
+     * @return A completion stage encapsulating the response or exception
+     */
+    CompletionStage<GetDataObjectResponse> getDataObjectAsync(GetDataObjectRequest request, Executor executor);
+
+    /**
+     * Update a data object/document in a table/index.
+     *
+     * @param request  A request identifying the data object to update
+     * @param executor the executor to use for asynchronous execution
+     * @return A completion stage encapsulating the response or exception
+     */
+    CompletionStage<UpdateDataObjectResponse> updateDataObjectAsync(UpdateDataObjectRequest request, Executor executor);
+
+    /**
+     * Delete a data object/document from a table/index.
+     *
+     * @param request  A request identifying the data object to delete
+     * @param executor the executor to use for asynchronous execution
+     * @return A completion stage encapsulating the response or exception
+     */
+    CompletionStage<DeleteDataObjectResponse> deleteDataObjectAsync(DeleteDataObjectRequest request, Executor executor);
+
+    /**
+     * Search for data objects/documents in a table/index.
+     *
+     * @param request  A request identifying the data objects to search for
+     * @param executor the executor to use for asynchronous execution
+     * @return A completion stage encapsulating the response or exception
+     */
+    CompletionStage<SearchDataObjectResponse> searchDataObjectAsync(SearchDataObjectRequest request, Executor executor);
+}

common/src/main/java/org/opensearch/sdk/SdkClientSettings.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.sdk;
+
+import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Setting.Property;
+
+/** Settings applicable to the SdkClient */
+public class SdkClientSettings {
+
+    /** The key for remote metadata type. */
+    public static final String REMOTE_METADATA_TYPE_KEY = "plugins.ml_commons.remote_metadata_type";
+
+    /** The value for remote metadata type for a remote cluster. */
+    public static final String REMOTE_OPENSEARCH = "RemoteOpenSearch";
+    /** The value for remote metadata type for a remote cluster on AWS OpenSearch Service. */
+    public static final String AWS_OPENSEARCH_SERVICE = "AWSOpenSearchService";
+    /** The value for remote metadata type for a remote cluster on AWS Dynamo DB. */
+    public static final String AWS_DYNAMO_DB = "AWSDynamoDB";
+
+    /** The key for remote metadata endpoint, applicable to remote clusters or DynamoDB. */
+    public static final String REMOTE_METADATA_ENDPOINT_KEY = "plugins.ml_commons.remote_metadata_endpoint";
+    /** The key for remote metadata region, applicable to AWS remote clusters or DynamoDB. */
+    public static final String REMOTE_METADATA_REGION_KEY = "plugins.ml_commons.remote_metadata_region";
+
+    public static final Setting<String> REMOTE_METADATA_TYPE = Setting.simpleString(REMOTE_METADATA_TYPE_KEY, Property.NodeScope, Property.Final);
+    public static final Setting<String> REMOTE_METADATA_ENDPOINT = Setting.simpleString(REMOTE_METADATA_ENDPOINT_KEY, Property.NodeScope, Property.Final);
+    public static final Setting<String> REMOTE_METADATA_REGION = Setting.simpleString(REMOTE_METADATA_REGION_KEY, Property.NodeScope, Property.Final);
+
+}

common/src/test/java/org/opensearch/sdk/SdkClientSettingsTests.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.sdk;
+
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Settings;
+
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.assertTrue;
+import static org.opensearch.sdk.SdkClientSettings.REMOTE_METADATA_ENDPOINT;
+import static org.opensearch.sdk.SdkClientSettings.REMOTE_METADATA_ENDPOINT_KEY;
+import static org.opensearch.sdk.SdkClientSettings.REMOTE_METADATA_REGION;
+import static org.opensearch.sdk.SdkClientSettings.REMOTE_METADATA_REGION_KEY;
+import static org.opensearch.sdk.SdkClientSettings.REMOTE_METADATA_TYPE;
+import static org.opensearch.sdk.SdkClientSettings.REMOTE_METADATA_TYPE_KEY;
+
+public class SdkClientSettingsTests {
+
+    private Settings settings;
+    private ClusterSettings clusterSettings;
+
+    public void setUp() throws Exception {
+        settings = Settings.builder().build();
+        final Set<Setting<?>> settingsSet = Stream
+            .concat(
+                ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.stream(),
+                Stream
+                    .of(
+                        REMOTE_METADATA_TYPE,
+                        REMOTE_METADATA_ENDPOINT,
+                        REMOTE_METADATA_REGION
+                    )
+            )
+            .collect(Collectors.toSet());
+        clusterSettings = new ClusterSettings(settings, settingsSet);
+    }
+
+    public void testSettings() {
+        assertTrue(clusterSettings.isFinalSetting(REMOTE_METADATA_TYPE_KEY));
+        assertTrue(clusterSettings.isFinalSetting(REMOTE_METADATA_ENDPOINT_KEY));
+        assertTrue(clusterSettings.isFinalSetting(REMOTE_METADATA_REGION_KEY));
+    }
+}