Ordering

Author: jasontedor

core/src/main/java/org/elasticsearch/transport/TcpTransport.java

@@ -589,9 +589,6 @@ public final NodeChannels openConnection(DiscoveryNode node, ConnectionProfile c
                     }
                 };
                 nodeChannels = connectToChannels(node, connectionProfile, this::onChannelOpen, onClose);
-                if (!Arrays.stream(nodeChannels.channels).allMatch(this::isOpen)) {
-                    throw new ConnectTransportException(node, "a channel closed while connecting");
-                }
                 final Channel channel = nodeChannels.getChannels().get(0); // one channel is guaranteed by the connection profile
                 final TimeValue connectTimeout = connectionProfile.getConnectTimeout() == null ?
                     defaultConnectionProfile.getConnectTimeout() :
@@ -600,8 +597,11 @@ public final NodeChannels openConnection(DiscoveryNode node, ConnectionProfile c
                     connectTimeout : connectionProfile.getHandshakeTimeout();
                 final Version version = executeHandshake(node, channel, handshakeTimeout);
                 nodeChannels = new NodeChannels(nodeChannels, version); // clone the channels - we now have the correct version
-                transportService.onConnectionOpened(nodeChannels);
                 connectionRef.set(nodeChannels);
+                transportService.onConnectionOpened(nodeChannels);
+                if (!Arrays.stream(nodeChannels.channels).allMatch(this::isOpen)) {
+                    throw new ConnectTransportException(node, "a channel closed while connecting");
+                }
                 success = true;
                 return nodeChannels;
             } catch (ConnectTransportException e) {

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

@@ -2624,7 +2624,17 @@ public void testProfilesIncludesDefault() {
     }
 
     public void testChannelCloseWhileConnecting() throws IOException {
-        final MockTransportService service = buildService("service", version0, clusterSettings, Settings.EMPTY, true, true, this::close);
+        /*
+         * We have to keep the first connection open so the handshake succeeds and then we later fail when checking if all the connections
+         * are open.
+         */
+        final AtomicBoolean first = new AtomicBoolean();
+        final MockTransportService service =
+                buildService("service", version0, clusterSettings, Settings.EMPTY, true, true, channel -> {
+                    if (!first.compareAndSet(false, true)) {
+                        close(channel);
+                    }
+                });
         final TcpTransport underlyingTransport = (TcpTransport) service.getOriginalTransport();
 
         final String otherName = "other_service";