Update MockTransportService to the age of Transport.Connection (#25320)

MockTransportServices allows us to simulate network disruptions in our testing infra. Sadly it wasn't updated to the state of the art in Transport land. This PR brings it up to speed. Specifically:

1) Opening a connection is now also blocked (before only node connections were blocked)
2) Simplifies things using the latest connection based notification between TcpTransport and TransportService for when a disconnect happens.
3) By 2, it fixes a race condition where we may fail to respond to a sent request when it is sent concurrently with the closing of a connection. The old code relied on a node based bridge between tcp transport and transport service. Sadly, the following doesn't work any more:

```
   if (transport.nodeConnected(node)) {
            // this a connected node, disconnecting from it will be up the exception
            transport.disconnectFromNode(node); <-- this may now be a noop and it doesn't mean that the transport service was notified of the disconnect between the nodeConnected check and here.
   } else {
            throw new ConnectTransportException(node, reason, e);
   }
```

Author: bleskes

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

@@ -47,7 +47,6 @@
 import org.elasticsearch.transport.ActionNotFoundTransportException;
 import org.elasticsearch.transport.ConnectTransportException;
 import org.elasticsearch.transport.ConnectionProfile;
-import org.elasticsearch.transport.NodeDisconnectedException;
 import org.elasticsearch.transport.NodeNotConnectedException;
 import org.elasticsearch.transport.RemoteTransportException;
 import org.elasticsearch.transport.RequestHandlerRegistry;
@@ -253,6 +252,7 @@ public Object getCacheKey() {
             @Override
             public void close() throws IOException {
                 transportServiceAdapter.onConnectionClosed(this);
+                LocalTransport.this.disconnectFromNode(node);
             }
         };
     }

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

@@ -39,7 +39,6 @@
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
-import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.plugins.Plugin;
@@ -263,10 +262,17 @@ public void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfil
                 }
             }
 
+            @Override
+            public Connection openConnection(DiscoveryNode node, ConnectionProfile profile) throws IOException {
+                throw new ConnectTransportException(node, "DISCONNECT: simulated");
+            }
+
             @Override
             protected void sendRequest(Connection connection, long requestId, String action, TransportRequest request,
                                        TransportRequestOptions options) throws IOException {
-                simulateDisconnect(connection, original, "DISCONNECT: simulated");
+                connection.close();
+                // send the request, which will blow up
+                connection.sendRequest(requestId, action, request, options);
             }
         });
     }
@@ -301,19 +307,12 @@ public void addFailToSendNoConnectRule(TransportAddress transportAddress, final
 
         addDelegate(transportAddress, new DelegateTransport(original) {
 
-            @Override
-            public void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfile,
-                                      CheckedBiConsumer<Connection, ConnectionProfile, IOException> connectionValidator)
-                throws ConnectTransportException {
-                original.connectToNode(node, connectionProfile, connectionValidator);
-            }
-
             @Override
             protected void sendRequest(Connection connection, long requestId, String action, TransportRequest request,
                                        TransportRequestOptions options) throws IOException {
                 if (blockedActions.contains(action)) {
                     logger.info("--> preventing {} request", action);
-                    simulateDisconnect(connection, original, "DISCONNECT: prevented " + action + " request");
+                    connection.close();
                 }
                 connection.sendRequest(requestId, action, request, options);
             }
@@ -347,6 +346,11 @@ public void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfil
                 }
             }
 
+            @Override
+            public Connection openConnection(DiscoveryNode node, ConnectionProfile profile) throws IOException {
+                throw new ConnectTransportException(node, "UNRESPONSIVE: simulated");
+            }
+
             @Override
             protected void sendRequest(Connection connection, long requestId, String action, TransportRequest request,
                                        TransportRequestOptions options) throws IOException {
@@ -413,6 +417,28 @@ public void connectToNode(DiscoveryNode node, ConnectionProfile connectionProfil
                 }
             }
 
+            @Override
+            public Connection openConnection(DiscoveryNode node, ConnectionProfile profile) throws IOException {
+                TimeValue delay = getDelay();
+                if (delay.millis() <= 0) {
+                    return original.openConnection(node, profile);
+                }
+
+                // TODO: Replace with proper setting
+                TimeValue connectingTimeout = NetworkService.TcpSettings.TCP_CONNECT_TIMEOUT.getDefault(Settings.EMPTY);
+                try {
+                    if (delay.millis() < connectingTimeout.millis()) {
+                        Thread.sleep(delay.millis());
+                        return original.openConnection(node, profile);
+                    } else {
+                        Thread.sleep(connectingTimeout.millis());
+                        throw new ConnectTransportException(node, "UNRESPONSIVE: simulated");
+                    }
+                } catch (InterruptedException e) {
+                    throw new ConnectTransportException(node, "UNRESPONSIVE: simulated");
+                }
+            }
+
             @Override
             protected void sendRequest(Connection connection, long requestId, String action, TransportRequest request,
                                        TransportRequestOptions options) throws IOException {
@@ -494,37 +520,6 @@ private LookupTestTransport transport() {
         return (LookupTestTransport) transport;
     }
 
-    /**
-     * simulates a disconnect by disconnecting from the underlying transport and throwing a
-     * {@link ConnectTransportException}
-     */
-    private void simulateDisconnect(DiscoveryNode node, Transport transport, String reason) {
-        simulateDisconnect(node, transport, reason, null);
-    }
-
-    /**
-     * simulates a disconnect by disconnecting from the underlying transport and throwing a
-     * {@link ConnectTransportException}, due to a specific cause exception
-     */
-    private void simulateDisconnect(DiscoveryNode node, Transport transport, String reason, @Nullable Throwable e) {
-        if (transport.nodeConnected(node)) {
-            // this a connected node, disconnecting from it will be up the exception
-            transport.disconnectFromNode(node);
-        } else {
-            throw new ConnectTransportException(node, reason, e);
-        }
-    }
-
-    /**
-     * simulates a disconnect by closing the connection and throwing a
-     * {@link ConnectTransportException}
-     */
-    private void simulateDisconnect(Transport.Connection connection, Transport transport, String reason) throws IOException {
-        connection.close();
-        simulateDisconnect(connection.getNode(), transport, reason);
-    }
-
-
     /**
      * A lookup transport that has a list of potential Transport implementations to delegate to for node operations,
      * if none is registered, then the default one is used.

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

@@ -1472,12 +1472,7 @@ public void handleException(TransportException exp) {
             // all is well
         }
 
-        try (Transport.Connection connection = serviceB.openConnection(nodeA, MockTcpTransport.LIGHT_PROFILE)){
-            serviceB.handshake(connection, 100);
-            fail("exception should be thrown");
-        } catch (IllegalStateException e) {
-            // all is well
-        }
+        expectThrows(ConnectTransportException.class, () -> serviceB.openConnection(nodeA, MockTcpTransport.LIGHT_PROFILE));
     }
 
     public void testMockUnresponsiveRule() throws IOException {
@@ -1528,12 +1523,7 @@ public void handleException(TransportException exp) {
             // all is well
         }
 
-        try (Transport.Connection connection = serviceB.openConnection(nodeA, MockTcpTransport.LIGHT_PROFILE)){
-            serviceB.handshake(connection, 100);
-            fail("exception should be thrown");
-        } catch (IllegalStateException e) {
-            // all is well
-        }
+        expectThrows(ConnectTransportException.class, () -> serviceB.openConnection(nodeA, MockTcpTransport.LIGHT_PROFILE));
     }