Fix race condition in RemoteClusterService.collectNodes()

It is possible for a linked remote to get unlinked in
between the containsKey() and get() calls in collectNodes().
This change adds a test that produces the NullPointerException
and adds a fix.

Author: JeremyDahlgren

server/src/main/java/org/elasticsearch/transport/RemoteClusterService.java

@@ -580,6 +580,13 @@ public void collectNodes(Set<String> clusters, ActionListener<BiFunction<String,
         Function<String, DiscoveryNode> nullFunction = s -> null;
         for (final String cluster : clusters) {
             RemoteClusterConnection connection = this.remoteClusters.get(cluster);
+            // Ensure the connection is not null, it could have been removed since the containsKey() call above.
+            if (connection == null) {
+                if (countDown.fastForward()) {
+                    listener.onFailure(new NoSuchRemoteClusterException(cluster));
+                }
+                break;
+            }
             connection.collectNodes(new ActionListener<Function<String, DiscoveryNode>>() {
                 @Override
                 public void onResponse(Function<String, DiscoveryNode> nodeLookup) {

server/src/test/java/org/elasticsearch/transport/RemoteClusterServiceTests.java

@@ -11,6 +11,7 @@
 import org.apache.logging.log4j.Level;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.LatchedActionListener;
 import org.elasticsearch.action.OriginalIndices;
 import org.elasticsearch.action.support.ActionTestUtils;
 import org.elasticsearch.action.support.IndicesOptions;
@@ -1060,6 +1061,79 @@ public void onFailure(Exception e) {
         }
     }
 
+    public void testCollectNodesConcurrentWithSettingsChanges() throws IOException {
+        final List<DiscoveryNode> knownNodes_c1 = new CopyOnWriteArrayList<>();
+
+        try (
+            var c1N1 = startTransport(
+                "cluster_1_node_1",
+                knownNodes_c1,
+                VersionInformation.CURRENT,
+                TransportVersion.current(),
+                Settings.EMPTY
+            );
+            var transportService = MockTransportService.createNewService(
+                Settings.EMPTY,
+                VersionInformation.CURRENT,
+                TransportVersion.current(),
+                threadPool,
+                null
+            )
+        ) {
+            final var c1N1Node = c1N1.getLocalNode();
+            knownNodes_c1.add(c1N1Node);
+            final var seedList = List.of(c1N1Node.getAddress().toString());
+            transportService.start();
+            transportService.acceptIncomingRequests();
+
+            try (RemoteClusterService service = new RemoteClusterService(createSettings("cluster_1", seedList), transportService)) {
+                service.initializeRemoteClusters();
+                assertTrue(service.isCrossClusterSearchEnabled());
+                final var numTasks = between(3, 5);
+                final var taskLatch = new CountDownLatch(numTasks);
+
+                ESTestCase.startInParallel(numTasks, threadNumber -> {
+                    if (threadNumber == 0) {
+                        taskLatch.countDown();
+                        boolean isLinked = true;
+                        while (taskLatch.getCount() != 0) {
+                            final var latch = new CountDownLatch(1);
+                            service.updateRemoteCluster(
+                                "cluster_1",
+                                createSettings("cluster_1", isLinked ? Collections.emptyList() : seedList),
+                                new LatchedActionListener<>(ActionListener.noop(), latch)
+                            );
+                            safeAwait(latch);
+                            isLinked = isLinked == false;
+                        }
+                        return;
+                    }
+
+                    // Verify collectNodes() always invokes the listener, even if the node is concurrently being unlinked.
+                    try {
+                        for (int i = 0; i < 1000; ++i) {
+                            final var latch = new CountDownLatch(1);
+                            service.collectNodes(Set.of("cluster_1"), new LatchedActionListener<>(new ActionListener<>() {
+                                @Override
+                                public void onResponse(BiFunction<String, String, DiscoveryNode> func) {
+                                    assertEquals(c1N1Node, func.apply("cluster_1", c1N1Node.getId()));
+                                }
+
+                                @Override
+                                public void onFailure(Exception e) {
+                                    assertNotNull(e);
+                                }
+                            }, latch));
+                            safeAwait(latch);
+                        }
+                    } finally {
+                        taskLatch.countDown();
+                    }
+                });
+            }
+        }
+    }
+
     public void testRemoteClusterSkipIfDisconnectedSetting() {
         {
             Settings settings = Settings.builder()