Fix the failing tests due to connection profile missing STREAM type

Signed-off-by: Rishabh Maurya <rishabhmaurya05@gmail.com>

Author: rishabhmaurya

server/src/main/java/org/opensearch/transport/ConnectionProfile.java

@@ -112,8 +112,8 @@ public static ConnectionProfile buildDefaultConnectionProfile(Settings settings)
         // if we are not a data-node we don't need any dedicated channels for recovery
         builder.addConnections(DiscoveryNode.isDataNode(settings) ? connectionsPerNodeRecovery : 0, TransportRequestOptions.Type.RECOVERY);
         builder.addConnections(connectionsPerNodeReg, TransportRequestOptions.Type.REG);
-        // TODO use different setting for connectionsPerNodeReg for stream request
-        builder.addConnections(connectionsPerNodeReg, TransportRequestOptions.Type.STREAM);
+        // we build a single channel profile with only supported type as STREAM for stream transport defined in StreamTransportService
+        builder.addConnections(0, TransportRequestOptions.Type.STREAM);
         return builder.build();
     }
 

server/src/main/java/org/opensearch/transport/StreamTransportService.java

@@ -113,7 +113,7 @@ public Transport.Connection getConnection(DiscoveryNode node) {
         try {
             return connectionManager.getConnection(node);
         } catch (Exception e) {
-            logger.error("Failed to get streaming connection to node [{}]", node, e);
+            logger.error("Failed to get streaming connection to node [{}]: {}", node, e.getMessage());
             throw new ConnectTransportException(node, "Failed to get streaming connection", e);
         }
     }

server/src/test/java/org/opensearch/transport/ConnectionProfileTests.java

@@ -77,19 +77,23 @@ public void testBuildConnectionProfile() {
         builder.addConnections(2, TransportRequestOptions.Type.STATE, TransportRequestOptions.Type.RECOVERY);
         builder.addConnections(3, TransportRequestOptions.Type.PING);
         IllegalStateException illegalStateException = expectThrows(IllegalStateException.class, builder::build);
-        assertEquals("not all types are added for this connection profile - missing types: [REG]", illegalStateException.getMessage());
+        assertEquals(
+            "not all types are added for this connection profile - missing types: [REG, STREAM]",
+            illegalStateException.getMessage()
+        );
 
         IllegalArgumentException illegalArgumentException = expectThrows(
             IllegalArgumentException.class,
             () -> builder.addConnections(4, TransportRequestOptions.Type.REG, TransportRequestOptions.Type.PING)
         );
         assertEquals("type [PING] is already registered", illegalArgumentException.getMessage());
         builder.addConnections(4, TransportRequestOptions.Type.REG);
+        builder.addConnections(1, TransportRequestOptions.Type.STREAM);
         ConnectionProfile build = builder.build();
         if (randomBoolean()) {
             build = new ConnectionProfile.Builder(build).build();
         }
-        assertEquals(10, build.getNumConnections());
+        assertEquals(11, build.getNumConnections());
         if (setConnectTimeout) {
             assertEquals(connectTimeout, build.getConnectTimeout());
         } else {
@@ -114,12 +118,12 @@ public void testBuildConnectionProfile() {
             assertNull(build.getPingInterval());
         }
 
-        List<Integer> list = new ArrayList<>(10);
-        for (int i = 0; i < 10; i++) {
+        List<Integer> list = new ArrayList<>(11);
+        for (int i = 0; i < 11; i++) {
             list.add(i);
         }
         final int numIters = randomIntBetween(5, 10);
-        assertEquals(4, build.getHandles().size());
+        assertEquals(5, build.getHandles().size());
         assertEquals(0, build.getHandles().get(0).offset);
         assertEquals(1, build.getHandles().get(0).length);
         assertEquals(EnumSet.of(TransportRequestOptions.Type.BULK), build.getHandles().get(0).getTypes());
@@ -155,11 +159,20 @@ public void testBuildConnectionProfile() {
             assertThat(channel, Matchers.anyOf(Matchers.is(6), Matchers.is(7), Matchers.is(8), Matchers.is(9)));
         }
 
+        assertEquals(10, build.getHandles().get(4).offset);
+        assertEquals(1, build.getHandles().get(4).length);
+        assertEquals(EnumSet.of(TransportRequestOptions.Type.STREAM), build.getHandles().get(4).getTypes());
+        channel = build.getHandles().get(4).getChannel(list);
+        for (int i = 0; i < numIters; i++) {
+            assertEquals(10, channel.intValue());
+        }
+
         assertEquals(3, build.getNumConnectionsPerType(TransportRequestOptions.Type.PING));
         assertEquals(4, build.getNumConnectionsPerType(TransportRequestOptions.Type.REG));
         assertEquals(2, build.getNumConnectionsPerType(TransportRequestOptions.Type.STATE));
         assertEquals(2, build.getNumConnectionsPerType(TransportRequestOptions.Type.RECOVERY));
         assertEquals(1, build.getNumConnectionsPerType(TransportRequestOptions.Type.BULK));
+        assertEquals(1, build.getNumConnectionsPerType(TransportRequestOptions.Type.STREAM));
     }
 
     public void testNoChannels() {
@@ -169,7 +182,8 @@ public void testNoChannels() {
             TransportRequestOptions.Type.BULK,
             TransportRequestOptions.Type.STATE,
             TransportRequestOptions.Type.RECOVERY,
-            TransportRequestOptions.Type.REG
+            TransportRequestOptions.Type.REG,
+            TransportRequestOptions.Type.STREAM
         );
         builder.addConnections(0, TransportRequestOptions.Type.PING);
         ConnectionProfile build = builder.build();
@@ -188,6 +202,7 @@ public void testConnectionProfileResolve() {
         builder.addConnections(randomIntBetween(0, 5), TransportRequestOptions.Type.REG);
         builder.addConnections(randomIntBetween(0, 5), TransportRequestOptions.Type.STATE);
         builder.addConnections(randomIntBetween(0, 5), TransportRequestOptions.Type.PING);
+        builder.addConnections(randomIntBetween(0, 5), TransportRequestOptions.Type.STREAM);
 
         final boolean connectionTimeoutSet = randomBoolean();
         if (connectionTimeoutSet) {
@@ -235,6 +250,7 @@ public void testDefaultConnectionProfile() {
         assertEquals(1, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STATE));
         assertEquals(2, profile.getNumConnectionsPerType(TransportRequestOptions.Type.RECOVERY));
         assertEquals(3, profile.getNumConnectionsPerType(TransportRequestOptions.Type.BULK));
+        assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STREAM));
         assertEquals(TransportSettings.CONNECT_TIMEOUT.get(Settings.EMPTY), profile.getConnectTimeout());
         assertEquals(TransportSettings.CONNECT_TIMEOUT.get(Settings.EMPTY), profile.getHandshakeTimeout());
         assertEquals(TransportSettings.TRANSPORT_COMPRESS.get(Settings.EMPTY), profile.getCompressionEnabled());
@@ -247,6 +263,7 @@ public void testDefaultConnectionProfile() {
         assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STATE));
         assertEquals(2, profile.getNumConnectionsPerType(TransportRequestOptions.Type.RECOVERY));
         assertEquals(3, profile.getNumConnectionsPerType(TransportRequestOptions.Type.BULK));
+        assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STREAM));
 
         profile = ConnectionProfile.buildDefaultConnectionProfile(nonDataNode());
         assertEquals(11, profile.getNumConnections());
@@ -255,6 +272,7 @@ public void testDefaultConnectionProfile() {
         assertEquals(1, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STATE));
         assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.RECOVERY));
         assertEquals(3, profile.getNumConnectionsPerType(TransportRequestOptions.Type.BULK));
+        assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STREAM));
 
         profile = ConnectionProfile.buildDefaultConnectionProfile(
             removeRoles(
@@ -267,5 +285,6 @@ public void testDefaultConnectionProfile() {
         assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STATE));
         assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.RECOVERY));
         assertEquals(3, profile.getNumConnectionsPerType(TransportRequestOptions.Type.BULK));
+        assertEquals(0, profile.getNumConnectionsPerType(TransportRequestOptions.Type.STREAM));
     }
 }

test/framework/src/main/java/org/opensearch/transport/AbstractSimpleTransportTestCase.java

@@ -2242,6 +2242,7 @@ public void testHandshakeUpdatesVersion() throws IOException {
                 TransportRequestOptions.Type.REG,
                 TransportRequestOptions.Type.STATE
             );
+            builder.addConnections(0, TransportRequestOptions.Type.STREAM);
             try (Transport.Connection connection = serviceA.openConnection(node, builder.build())) {
                 assertEquals(version, connection.getVersion());
             }