fix rebase conflict

Signed-off-by: Jiaping Zeng <[email protected]>

Author: jiapingzeng

common/src/test/java/org/opensearch/ml/common/utils/StringUtilsTest.java

@@ -17,7 +17,6 @@
 
 import java.io.IOException;
 import java.lang.reflect.Constructor;
-import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -1202,9 +1201,9 @@ public void testSerializeFloatNaNAndInfinity_BecomesNull_InPojo() {
     @Test
     public void testDeserializeScientificNotation_ToFloatAndPrimitive() {
         String jsonObj = "{\"fObj\":1.23e-5}";
-        Type mapType = new TypeToken<Map<String, Float>>() {
+        java.lang.reflect.Type mapType = new com.google.gson.reflect.TypeToken<java.util.Map<String, Float>>() {
         }.getType();
-        Map<String, Float> m = StringUtils.PLAIN_NUMBER_GSON.fromJson(jsonObj, mapType);
+        java.util.Map<String, Float> m = StringUtils.PLAIN_NUMBER_GSON.fromJson(jsonObj, mapType);
         assertEquals(1.23e-5f, m.get("fObj"), 1e-9f);
 
         String jsonArr = "[4.56e1]";
@@ -1216,9 +1215,9 @@ public void testDeserializeScientificNotation_ToFloatAndPrimitive() {
     public void testDeserializeNullFloat_ToNull() {
         String json = "{\"fObj\":null,\"fPrim\":1.0}";
 
-        Type mapType = new TypeToken<Map<String, JsonElement>>() {
+        java.lang.reflect.Type mapType = new TypeToken<java.util.Map<String, JsonElement>>() {
         }.getType();
-        Map<String, JsonElement> m = StringUtils.PLAIN_NUMBER_GSON.fromJson(json, mapType);
+        java.util.Map<String, JsonElement> m = StringUtils.PLAIN_NUMBER_GSON.fromJson(json, mapType);
 
         assertTrue(m.containsKey("fObj"));
         assertTrue(m.get("fObj").isJsonNull());

ml-algorithms/src/main/java/org/opensearch/ml/engine/algorithms/remote/AwsConnectorExecutor.java

@@ -18,6 +18,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.text.StringEscapeUtils;
 import org.apache.logging.log4j.Logger;
@@ -40,6 +41,8 @@
 import org.opensearch.transport.StreamTransportService;
 import org.opensearch.transport.client.Client;
 
+import com.google.common.annotations.VisibleForTesting;
+
 import lombok.Getter;
 import lombok.Setter;
 import lombok.extern.log4j.Log4j2;
@@ -70,19 +73,18 @@ public class AwsConnectorExecutor extends AbstractConnectorExecutor {
     @Getter
     private MLGuard mlGuard;
 
-    private SdkAsyncHttpClient httpClient;
+    private final AtomicReference<SdkAsyncHttpClient> httpClientRef = new AtomicReference<>();
 
     @Setter
     @Getter
     private StreamTransportService streamTransportService;
 
+    @Setter
+    private boolean connectorPrivateIpEnabled;
+
     public AwsConnectorExecutor(Connector connector) {
         super.initialize(connector);
         this.connector = (AwsConnector) connector;
-        Duration connectionTimeout = Duration.ofSeconds(super.getConnectorClientConfig().getConnectionTimeout());
-        Duration readTimeout = Duration.ofSeconds(super.getConnectorClientConfig().getReadTimeout());
-        Integer maxConnection = super.getConnectorClientConfig().getMaxConnections();
-        this.httpClient = MLHttpClientFactory.getAsyncHttpClient(connectionTimeout, readTimeout, maxConnection);
     }
 
     @Override
@@ -129,7 +131,8 @@ public void invokeRemoteService(
                     )
                 )
                 .build();
-            AccessController.doPrivileged((PrivilegedExceptionAction<CompletableFuture<Void>>) () -> httpClient.execute(executeRequest));
+            AccessController
+                .doPrivileged((PrivilegedExceptionAction<CompletableFuture<Void>>) () -> getHttpClient().execute(executeRequest));
         } catch (RuntimeException exception) {
             log.error("Failed to execute {} in aws connector: {}", action, exception.getMessage(), exception);
             actionListener.onFailure(exception);
@@ -180,4 +183,19 @@ private void validateLLMInterface(String llmInterface) {
                 throw new IllegalArgumentException(String.format("Unsupported llm interface: %s", llmInterface));
         }
     }
+
+    @VisibleForTesting
+    protected SdkAsyncHttpClient getHttpClient() {
+        if (httpClientRef.get() == null) {
+            Duration connectionTimeout = Duration.ofSeconds(super.getConnectorClientConfig().getConnectionTimeout());
+            Duration readTimeout = Duration.ofSeconds(super.getConnectorClientConfig().getReadTimeout());
+            Integer maxConnection = super.getConnectorClientConfig().getMaxConnections();
+            this.httpClientRef
+                .compareAndSet(
+                    null,
+                    MLHttpClientFactory.getAsyncHttpClient(connectionTimeout, readTimeout, maxConnection, connectorPrivateIpEnabled)
+                );
+        }
+        return httpClientRef.get();
+    }
 }