Only connect to new nodes on new cluster state

server/src/main/java/org/elasticsearch/cluster/NodeConnectionsService.java

@@ -78,17 +78,18 @@ public NodeConnectionsService(Settings settings, ThreadPool threadPool, Transpor
         this.reconnectInterval = NodeConnectionsService.CLUSTER_NODE_RECONNECT_INTERVAL_SETTING.get(settings);
     }
 
-    public void connectToNodes(DiscoveryNodes discoveryNodes) {
+    public void connectToNodes(DiscoveryNodes discoveryNodes, boolean reconnectToKnownNodes) {
         CountDownLatch latch = new CountDownLatch(discoveryNodes.getSize());
         for (final DiscoveryNode node : discoveryNodes) {
-            final boolean connected;
+            final boolean shouldConnect;
             try (Releasable ignored = nodeLocks.acquire(node)) {
-                nodes.putIfAbsent(node, 0);
-                connected = transportService.nodeConnected(node);
+                // We try and connect to any new nodes before returning. However, on the elected master the connections are established
+                // during joining, so we also check that we're not already connected to avoid the need to execute any background tasks in
+                // that case.
+                shouldConnect = (nodes.putIfAbsent(node, 0) == null || reconnectToKnownNodes)
+                    && transportService.nodeConnected(node) == false;
             }
-            if (connected) {
-                latch.countDown();
-            } else {
+            if (shouldConnect) {
                 // spawn to another thread to do in parallel
                 threadPool.executor(ThreadPool.Names.MANAGEMENT).execute(new AbstractRunnable() {
                     @Override
@@ -112,6 +113,8 @@ public void onAfter() {
                         latch.countDown();
                     }
                 });
+            } else {
+                latch.countDown();
             }
         }
         try {

server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java

@@ -462,7 +462,7 @@ private void applyChanges(UpdateTask task, ClusterState previousClusterState, Cl
             }
         }
 
-        nodeConnectionsService.connectToNodes(newClusterState.nodes());
+        nodeConnectionsService.connectToNodes(newClusterState.nodes(), false);
 
         logger.debug("applying cluster state version {}", newClusterState.version());
         try {

server/src/test/java/org/elasticsearch/cluster/NodeConnectionsServiceTests.java

@@ -45,7 +45,6 @@
 import org.junit.Before;
 
 import java.io.IOException;
-import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -89,7 +88,7 @@ public void testConnectAndDisconnect() {
         ClusterState current = clusterStateFromNodes(Collections.emptyList());
         ClusterChangedEvent event = new ClusterChangedEvent("test", clusterStateFromNodes(randomSubsetOf(nodes)), current);
 
-        service.connectToNodes(event.state().nodes());
+        service.connectToNodes(event.state().nodes(), false);
         assertConnected(event.state().nodes());
 
         service.disconnectFromNodesExcept(event.state().nodes());
@@ -98,7 +97,7 @@ public void testConnectAndDisconnect() {
         current = event.state();
         event = new ClusterChangedEvent("test", clusterStateFromNodes(randomSubsetOf(nodes)), current);
 
-        service.connectToNodes(event.state().nodes());
+        service.connectToNodes(event.state().nodes(), false);
         assertConnected(event.state().nodes());
 
         service.disconnectFromNodesExcept(event.state().nodes());
@@ -115,7 +114,7 @@ public void testReconnect() {
 
         transport.randomConnectionExceptions = true;
 
-        service.connectToNodes(event.state().nodes());
+        service.connectToNodes(event.state().nodes(), false);
 
         for (int i = 0; i < 3; i++) {
             // simulate disconnects
@@ -131,6 +130,45 @@ public void testReconnect() {
         assertConnectedExactlyToNodes(event.state());
     }
 
+    public void testDoesNotReconnectToKnownNodesOnNewClusterState() {
+        List<DiscoveryNode> nodes = generateNodes();
+        NodeConnectionsService service = new NodeConnectionsService(Settings.EMPTY, threadPool, transportService);
+
+        final ClusterState state1 = clusterStateFromNodes(randomSubsetOf(nodes));
+        final ClusterState state2 = clusterStateFromNodes(randomSubsetOf(nodes));
+
+        service.connectToNodes(state1.nodes(), false);
+
+        final Set<DiscoveryNode> disconnectedNodes = new HashSet<>(randomSubsetOf(nodes));
+        for (final DiscoveryNode node : disconnectedNodes) {
+            transport.disconnectFromNode(node);
+        }
+
+        boolean shouldReconnect = randomBoolean();
+
+        service.connectToNodes(state2.nodes(), shouldReconnect);
+
+        final Set<DiscoveryNode> expectedNodes = new HashSet<>(nodes.size());
+        state1.nodes().forEach(expectedNodes::add);
+        disconnectedNodes.forEach(expectedNodes::remove);
+        for (final DiscoveryNode discoveryNode : state2.nodes()) {
+            if (shouldReconnect || state1.nodes().get(discoveryNode.getId()) == null) {
+                // Only expect to be connected to _new_ nodes in state2 unless shouldReconnect is set
+                expectedNodes.add(discoveryNode);
+            }
+        }
+
+        assertConnected(expectedNodes);
+        assertThat(transport.connectedNodes.size(), equalTo(expectedNodes.size()));
+
+        service.new ConnectionChecker().run();
+
+        state1.nodes().forEach(expectedNodes::add);
+
+        assertConnected(expectedNodes);
+        assertThat(transport.connectedNodes.size(), equalTo(expectedNodes.size()));
+    }
+
     private void assertConnectedExactlyToNodes(ClusterState state) {
         assertConnected(state.nodes());
         assertThat(transport.connectedNodes.size(), equalTo(state.nodes().getSize()));
@@ -142,12 +180,6 @@ private void assertConnected(Iterable<DiscoveryNode> nodes) {
         }
     }
 
-    private void assertNotConnected(Iterable<DiscoveryNode> nodes) {
-        for (DiscoveryNode node : nodes) {
-            assertFalse("still connected to " + node, transport.connectedNodes.contains(node));
-        }
-    }
-
     @Override
     @Before
     public void setUp() throws Exception {
@@ -190,7 +222,7 @@ public Map<String, BoundTransportAddress> profileBoundAddresses() {
         }
 
         @Override
-        public TransportAddress[] addressesFromString(String address, int perAddressLimit) throws UnknownHostException {
+        public TransportAddress[] addressesFromString(String address, int perAddressLimit) {
             return new TransportAddress[0];
         }
 
@@ -226,19 +258,19 @@ public DiscoveryNode getNode() {
 
                 @Override
                 public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
-                    throws IOException, TransportException {
+                    throws TransportException {
 
                 }
 
                 @Override
-                public void close() throws IOException {
+                public void close() {
 
                 }
             };
         }
 
         @Override
-        public Connection openConnection(DiscoveryNode node, ConnectionProfile profile) throws IOException {
+        public Connection openConnection(DiscoveryNode node, ConnectionProfile profile) {
             return getConnection(node);
         }
 

server/src/test/java/org/elasticsearch/cluster/service/ClusterApplierServiceTests.java

@@ -91,8 +91,9 @@ TimedClusterApplierService createTimedClusterService(boolean makeMaster) throws
             "ClusterApplierServiceTests").build(), new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
             threadPool);
         timedClusterApplierService.setNodeConnectionsService(new NodeConnectionsService(Settings.EMPTY, null, null) {
+
             @Override
-            public void connectToNodes(DiscoveryNodes discoveryNodes) {
+            public void connectToNodes(DiscoveryNodes discoveryNodes, boolean reconnectToKnownNodes) {
                 // skip
             }
 

server/src/test/java/org/elasticsearch/discovery/ClusterDisruptionIT.java

@@ -47,6 +47,7 @@
 import org.elasticsearch.test.disruption.NetworkDisruption.TwoPartitions;
 import org.elasticsearch.test.disruption.ServiceDisruptionScheme;
 import org.elasticsearch.test.junit.annotations.TestLogging;
+import org.elasticsearch.transport.NodeNotConnectedException;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -207,7 +208,7 @@ public void testAckedIndexing() throws Exception {
                                 assertTrue("doc [" + id + "] indexed via node [" + ackedDocs.get(id) + "] not found",
                                     client(node).prepareGet("test", "type", id).setPreference("_local").get().isExists());
                             }
-                        } catch (AssertionError | NoShardAvailableActionException e) {
+                        } catch (AssertionError | NoShardAvailableActionException | NodeNotConnectedException e) {
                             throw new AssertionError(e.getMessage() + " (checked via node [" + node + "]", e);
                         }
                     }

test/framework/src/main/java/org/elasticsearch/test/ClusterServiceUtils.java

@@ -135,7 +135,7 @@ public static ClusterService createClusterService(ThreadPool threadPool, Discove
             clusterSettings, threadPool, Collections.emptyMap());
         clusterService.setNodeConnectionsService(new NodeConnectionsService(Settings.EMPTY, null, null) {
             @Override
-            public void connectToNodes(DiscoveryNodes discoveryNodes) {
+            public void connectToNodes(DiscoveryNodes discoveryNodes, boolean reconnectToKnownNodes) {
                 // skip
             }
 

test/framework/src/main/java/org/elasticsearch/test/disruption/NetworkDisruption.java

@@ -104,7 +104,7 @@ public void ensureHealthy(InternalTestCluster cluster) {
     public static void ensureFullyConnectedCluster(InternalTestCluster cluster) {
         for (String node: cluster.getNodeNames()) {
             ClusterState stateOnNode = cluster.getInstance(ClusterService.class, node).state();
-            cluster.getInstance(NodeConnectionsService.class, node).connectToNodes(stateOnNode.nodes());
+            cluster.getInstance(NodeConnectionsService.class, node).connectToNodes(stateOnNode.nodes(), true);
         }
     }