Refactor Sniffer and make it testable (#29638)

This commit reworks the Sniffer component to simplify it and make it possible to test it.

In particular, it no longer takes out the host that failed when sniffing on failure, but rather relies on whatever the cluster returns. This is the result of some valid comments from #27985. Taking out one single host is too naive, hard to test and debug.

A new Scheduler abstraction is introduced to abstract the tasks scheduling away and make it possible to plug in any test implementation and take out timing aspects when testing.

Concurrency aspects have also been improved, synchronized methods are no longer required. At the same time, we were able to take #27697 and #25701 into account and fix them, especially now that we can more easily add tests.

Last but not least, unit tests are added for the Sniffer component, long overdue.

Closes #27697
Closes #25701

client/rest/src/main/java/org/elasticsearch/client/RestClient.java

@@ -61,6 +61,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -132,7 +133,7 @@ public synchronized void setHosts(HttpHost... hosts) {
         if (hosts == null || hosts.length == 0) {
             throw new IllegalArgumentException("hosts must not be null nor empty");
         }
-        Set<HttpHost> httpHosts = new HashSet<>();
+        Set<HttpHost> httpHosts = new LinkedHashSet<>();
         AuthCache authCache = new BasicAuthCache();
         for (HttpHost host : hosts) {
             Objects.requireNonNull(host, "host cannot be null");
@@ -143,6 +144,13 @@ public synchronized void setHosts(HttpHost... hosts) {
         this.blacklist.clear();
     }
 
+    /**
+     * Returns the configured hosts
+     */
+    public List<HttpHost> getHosts() {
+        return new ArrayList<>(hostTuple.hosts);
+    }
+
     /**
      * Sends a request to the Elasticsearch cluster that the client points to.
      * Blocks until the request is completed and returns its response or fails

client/rest/src/test/java/org/elasticsearch/client/RestClientTests.java

@@ -25,6 +25,7 @@
 
 import java.io.IOException;
 import java.net.URI;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -251,6 +252,37 @@ public void testSetHostsWrongArguments() throws IOException {
         }
     }
 
+    public void testSetHostsPreservesOrdering() throws Exception {
+        try (RestClient restClient = createRestClient()) {
+            HttpHost[] hosts = randomHosts();
+            restClient.setHosts(hosts);
+            assertEquals(Arrays.asList(hosts), restClient.getHosts());
+        }
+    }
+
+    private static HttpHost[] randomHosts() {
+        int numHosts = randomIntBetween(1, 10);
+        HttpHost[] hosts = new HttpHost[numHosts];
+        for (int i = 0; i < hosts.length; i++) {
+            hosts[i] = new HttpHost("host-" + i, 9200);
+        }
+        return hosts;
+    }
+
+    public void testSetHostsDuplicatedHosts() throws Exception {
+        try (RestClient restClient = createRestClient()) {
+            int numHosts = randomIntBetween(1, 10);
+            HttpHost[] hosts = new HttpHost[numHosts];
+            HttpHost host = new HttpHost("host", 9200);
+            for (int i = 0; i < hosts.length; i++) {
+                hosts[i] = host;
+            }
+            restClient.setHosts(hosts);
+            assertEquals(1, restClient.getHosts().size());
+            assertEquals(host, restClient.getHosts().get(0));
+        }
+    }
+
     /**
      * @deprecated will remove method in 7.0 but needs tests until then. Replaced by {@link RequestTests#testConstructor()}.
      */

client/sniffer/src/main/java/org/elasticsearch/client/sniff/SniffOnFailureListener.java

@@ -58,7 +58,6 @@ public void onFailure(HttpHost host) {
         if (sniffer == null) {
             throw new IllegalStateException("sniffer was not set, unable to sniff on failure");
         }
-        //re-sniff immediately but take out the node that failed
-        sniffer.sniffOnFailure(host);
+        sniffer.sniffOnFailure();
     }
 }