Adapt LocalTransport & fix TCPTransportTests, AbstractSimpleTransportTestCase & Netty3ScheduledPingTests broken by backport of #22828

Author: bleskes

core/src/main/java/org/elasticsearch/transport/local/LocalTransport.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.CheckedBiConsumer;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -169,7 +170,9 @@ public boolean nodeConnected(DiscoveryNode node) {
     }
 
     @Override
-    public void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfile) throws ConnectTransportException {
+    public void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfile,
+                              CheckedBiConsumer<Connection, ConnectionProfile, IOException> connectionValidator)
+        throws ConnectTransportException {
         synchronized (this) {
             if (connectedNodes.containsKey(node)) {
                 return;

core/src/test/java/org/elasticsearch/transport/TCPTransportTests.java

@@ -256,11 +256,11 @@ public void testConnectionProfileResolve() {
 
         final boolean connectionTimeoutSet = randomBoolean();
         if (connectionTimeoutSet) {
-            builder.setConnectTimeout(TimeValue.timeValueMillis(randomNonNegativeLong()));
+            builder.setConnectTimeout(TimeValue.timeValueMillis(randomPositiveLong()));
         }
         final boolean connectionHandshakeSet = randomBoolean();
         if (connectionHandshakeSet) {
-            builder.setHandshakeTimeout(TimeValue.timeValueMillis(randomNonNegativeLong()));
+            builder.setHandshakeTimeout(TimeValue.timeValueMillis(randomPositiveLong()));
         }
 
         final ConnectionProfile profile = builder.build();

modules/transport-netty3/src/test/java/org/elasticsearch/transport/netty3/Netty3ScheduledPingTests.java

@@ -81,10 +81,8 @@ public void testScheduledPing() throws Exception {
         serviceB.start();
         serviceB.acceptIncomingRequests();
 
-        DiscoveryNode nodeA =
-            new DiscoveryNode("TS_A", "TS_A", serviceA.boundAddress().publishAddress(), emptyMap(), emptySet(), Version.CURRENT);
-        DiscoveryNode nodeB =
-            new DiscoveryNode("TS_B", "TS_B", serviceB.boundAddress().publishAddress(), emptyMap(), emptySet(), Version.CURRENT);
+        DiscoveryNode nodeA = serviceA.getLocalDiscoNode();
+        DiscoveryNode nodeB = serviceB.getLocalDiscoNode();
 
         if (randomBoolean()) {
             // use connection profile with different connect timeout

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

@@ -1962,12 +1962,11 @@ protected String handleRequest(MockChannel mockChannel, String profileName, Stre
             // this acts like a node that doesn't have support for handshakes
             DiscoveryNode node =
                 new DiscoveryNode("TS_TPC", "TS_TPC", transport.boundAddress().publishAddress(), emptyMap(), emptySet(), version0);
-            serviceA.connectToNode(node);
-            TcpTransport.NodeChannels connection = originalTransport.getConnection(node);
-            Version version = originalTransport.executeHandshake(node, connection.channel(TransportRequestOptions.Type.PING),
-                TimeValue.timeValueSeconds(10));
-            assertNull(version);
-            serviceA.disconnectFromNode(node);
+            try(TcpTransport.NodeChannels connection = originalTransport.openConnection(node, null)) {
+                Version version = originalTransport.executeHandshake(node, connection.channel(TransportRequestOptions.Type.PING),
+                    TimeValue.timeValueSeconds(10));
+                assertNull(version);
+            }
         }
 
         try (TransportService service = buildService("TS_TPC", Version.CURRENT, null);