Transport service

src/main/java/org/opensearch/sdk/ExtensionsRunner.java

@@ -91,7 +91,7 @@ public class ExtensionsRunner {
     private ExtensionsIndicesModuleNameRequestHandler extensionsIndicesModuleNameRequestHandler =
         new ExtensionsIndicesModuleNameRequestHandler();
     private ExtensionsRestRequestHandler extensionsRestRequestHandler = new ExtensionsRestRequestHandler(extensionRestPathRegistry);
-    private NettyTransport nettyTransport = new NettyTransport();
+
     private SDKClient client = new SDKClient();
 
     /*
@@ -131,9 +131,9 @@ protected ExtensionsRunner(Extension extension) throws IOException {
         this.customSettings = extension.getSettings();
         // save custom transport actions
         this.transportActions = new TransportActions(extension.getActions());
-        // initialize the transport service
+
         ThreadPool threadPool = new ThreadPool(this.getSettings());
-        nettyTransport.initializeExtensionTransportService(this.getSettings(), threadPool, this);
+
         // create components
         extension.createComponents(client, null, threadPool);
     }
@@ -461,6 +461,8 @@ public static void run(Extension extension) throws IOException {
         logger.info("Starting extension " + extension.getExtensionSettings().getExtensionName());
         @SuppressWarnings("unused")
         ExtensionsRunner runner = new ExtensionsRunner(extension);
+        // initialize the transport service
+        new NettyTransport(runner).initializeExtensionTransportService(runner.getSettings(), new ThreadPool(runner.getSettings()), runner);
         runner.startActionListener(0);
     }
 }

src/main/java/org/opensearch/sdk/NettyTransport.java

@@ -40,9 +40,17 @@
 
 public class NettyTransport {
     private static final String NODE_NAME_SETTING = "node.name";
+    private final ExtensionsRunner extensionsRunner;
     private final TransportInterceptor NOOP_TRANSPORT_INTERCEPTOR = new TransportInterceptor() {
     };
 
+    /**
+     * @param extensionsRunner Instantiate this object with a reference to the ExtensionsRunner.
+     */
+    public NettyTransport(ExtensionsRunner extensionsRunner) {
+        this.extensionsRunner = extensionsRunner;
+    }
+
     /**
      * Initializes a Netty4Transport object. This object will be wrapped in a {@link TransportService} object.
      *
@@ -103,7 +111,7 @@ public TransportService initializeExtensionTransportService(
         }
 
         // create transport service
-        extensionsRunner.extensionTransportService = new TransportService(
+        TransportService transportService = new TransportService(
             settings,
             transport,
             threadPool,
@@ -116,8 +124,8 @@ public TransportService initializeExtensionTransportService(
             null,
             emptySet()
         );
-        extensionsRunner.startTransportService(extensionsRunner.extensionTransportService);
-        return extensionsRunner.extensionTransportService;
+        extensionsRunner.startTransportService(transportService);
+        return transportService;
     }
 
 }

src/test/java/org/opensearch/sdk/TestExtensionsRunner.java

@@ -29,7 +29,6 @@
 import java.util.Collections;
 import java.util.HashMap;
 
-import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.opensearch.Version;
@@ -76,14 +75,12 @@ public class TestExtensionsRunner extends OpenSearchTestCase {
     private ExtensionsRunner extensionsRunner;
     private TransportService transportService;
 
-    private TransportService initialTransportService;
-
     @Override
     @BeforeEach
     public void setUp() throws Exception {
         super.setUp();
         this.extensionsRunner = new ExtensionsRunnerForTest();
-        this.initialTransportService = extensionsRunner.extensionTransportService;
+
         this.transportService = spy(
             new TransportService(
                 Settings.EMPTY,
@@ -97,17 +94,6 @@ public void setUp() throws Exception {
         );
     }
 
-    @Override
-    @AfterEach
-    public void tearDown() throws Exception {
-        super.tearDown();
-        if (initialTransportService != null) {
-            this.initialTransportService.stop();
-            this.initialTransportService.close();
-            Thread.sleep(1000);
-        }
-    }
-
     // test manager method invokes start on transport service
     @Test
     public void testTransportServiceStarted() {

src/test/java/org/opensearch/sdk/TestNetty4Transport.java

@@ -12,6 +12,7 @@
 package org.opensearch.sdk;
 
 import java.io.IOException;
+
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.opensearch.common.component.Lifecycle;
@@ -25,11 +26,14 @@
 public class TestNetty4Transport extends OpenSearchTestCase {
 
     private ThreadPool threadPool;
-    private NettyTransport nettyTransport = new NettyTransport();
+    private ExtensionsRunner extensionsRunner;
+    private NettyTransport nettyTransport;
 
     @BeforeEach
     public void setUp() throws IOException {
         this.threadPool = new TestThreadPool("test");
+        this.extensionsRunner = new ExtensionsRunnerForTest();
+        this.nettyTransport = new NettyTransport(extensionsRunner);
     }
 
     // test Netty can bind to multiple ports, default and additional client

src/test/java/org/opensearch/sdk/TransportCommunicationIT.java

@@ -35,18 +35,21 @@ public class TransportCommunicationIT extends OpenSearchIntegTestCase {
     private final int port = 7777;
     private final String host = "127.0.0.1";
     private volatile String clientResult;
-    private NettyTransport nettyTransport = new NettyTransport();
+    private ExtensionsRunner extensionsRunner;
+    private NettyTransport nettyTransport;
 
     @Override
     @BeforeEach
-    public void setUp() {
+    public void setUp() throws IOException {
 
         // Configure settings for transport serivce using the same port number used to bind the client
         settings = Settings.builder()
             .put("node.name", "node_extension_test")
             .put(TransportSettings.BIND_HOST.getKey(), host)
             .put(TransportSettings.PORT.getKey(), port)
             .build();
+        this.extensionsRunner = new ExtensionsRunnerForTest();
+        this.nettyTransport = new NettyTransport(extensionsRunner);
     }
 
     @Test