Remove I/O pool blocking sniffing call from onFailure callback, add a…

… some logic around host exclusion

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

@@ -29,8 +29,11 @@
 import java.io.IOException;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadFactory;
@@ -53,15 +56,16 @@ public class Sniffer implements Closeable {
 
  private final Task task;
 
- Sniffer(RestClient restClient, HostsSniffer hostsSniffer, long sniffInterval, long sniffAfterFailureDelay) {
- this.task = new Task(hostsSniffer, restClient, sniffInterval, sniffAfterFailureDelay);
+ Sniffer(RestClient restClient, HostsSniffer hostsSniffer, long sniffInterval, long sniffAfterFailureDelay, int maxExcludedRounds) {
+ this.task = new Task(hostsSniffer, restClient, sniffInterval, sniffAfterFailureDelay, maxExcludedRounds);
  }
 
  /**
  * Triggers a new sniffing round and explicitly takes out the failed host provided as argument
  */
  public void sniffOnFailure(HttpHost failedHost) {
- this.task.sniffOnFailure(failedHost);
+ this.task.failedHosts.putIfAbsent(failedHost, 0L);
+ this.task.scheduleNextRun(0);
  }
 
  @Override
@@ -75,15 +79,24 @@ private static class Task implements Runnable {
 
  private final long sniffIntervalMillis;
  private final long sniffAfterFailureDelayMillis;
+ private final int maxExcludedRounds;
  private final ScheduledExecutorService scheduledExecutorService;
  private final AtomicBoolean running = new AtomicBoolean(false);
  private ScheduledFuture<?> scheduledFuture;
+ private ConcurrentHashMap<HttpHost, Long> failedHosts = new ConcurrentHashMap<>();
+
+ private Task(
+ HostsSniffer hostsSniffer,
+ RestClient restClient,
+ long sniffIntervalMillis,
+ long sniffAfterFailureDelayMillis,
+ int maxExcludedRounds) {
 
- private Task(HostsSniffer hostsSniffer, RestClient restClient, long sniffIntervalMillis, long sniffAfterFailureDelayMillis) {
  this.hostsSniffer = hostsSniffer;
  this.restClient = restClient;
  this.sniffIntervalMillis = sniffIntervalMillis;
  this.sniffAfterFailureDelayMillis = sniffAfterFailureDelayMillis;
+ this.maxExcludedRounds = maxExcludedRounds;
  SnifferThreadFactory threadFactory = new SnifferThreadFactory(SNIFFER_THREAD_NAME);
  this.scheduledExecutorService = Executors.newScheduledThreadPool(1, threadFactory);
  scheduleNextRun(0);
@@ -106,35 +119,63 @@ synchronized void scheduleNextRun(long delayMillis) {
 
  @Override
  public void run() {
- sniff(null, sniffIntervalMillis);
- }
-
- void sniffOnFailure(HttpHost failedHost) {
- sniff(failedHost, sniffAfterFailureDelayMillis);
+ sniff(sniffIntervalMillis);
  }
 
- void sniff(HttpHost excludeHost, long nextSniffDelayMillis) {
+ void sniff(long nextSniffDelayMillis) {
  if (running.compareAndSet(false, true)) {
+ long nextSniffDelay = nextSniffDelayMillis;
  try {
  List<HttpHost> sniffedHosts = hostsSniffer.sniffHosts();
  logger.debug("sniffed hosts: " + sniffedHosts);
- if (excludeHost != null) {
- sniffedHosts.remove(excludeHost);
- }
- if (sniffedHosts.isEmpty()) {
+
+ List<HttpHost> hostsFiltered = removeExcludedAndCycle(sniffedHosts);
+ logger.debug("sniffed hosts after filtering: " + sniffedHosts);
+
+ if (hostsFiltered.isEmpty()) {
  logger.warn("no hosts to set, hosts will be updated at the next sniffing round");
  } else {
- this.restClient.setHosts(sniffedHosts.toArray(new HttpHost[sniffedHosts.size()]));
+ this.restClient.setHosts(hostsFiltered.toArray(new HttpHost[hostsFiltered.size()]));
  }
  } catch (Exception e) {
  logger.error("error while sniffing nodes", e);
+ nextSniffDelay = sniffAfterFailureDelayMillis;
  } finally {
- scheduleNextRun(nextSniffDelayMillis);
+ scheduleNextRun(nextSniffDelay);
  running.set(false);
  }
  }
  }
 
+ /**
+ * Remove excluded hosts from the list of all sniffed hosts, and cycle through the map. Hosts in the map remain
+ * there for {@link org.elasticsearch.client.sniff.Sniffer.Task#maxExcludedRounds} cycles
+ * @param allHosts the list of all sniffed hosts
+ * @return a new list containing the remaining hosts
+ */
+ private List<HttpHost> removeExcludedAndCycle(List<HttpHost> allHosts) {
+ final List<HttpHost> excluded = Collections.list(failedHosts.keys());
+
+ if (excluded.isEmpty()) {
+ return allHosts;
+ }
+
+ try {
+ List<HttpHost> copy = new ArrayList<>(allHosts);
+ copy.removeAll(excluded);
+ return copy;
+ } finally {
+ for (HttpHost host : excluded) {
+ long excludedCycles = failedHosts.get(host) + 1;
+ if (excludedCycles >= maxExcludedRounds) {
+ failedHosts.remove(host);
+ } else {
+ failedHosts.put(host, excludedCycles);
+ }
+ }
+ }
+ }
+
  synchronized void shutdown() {
  scheduledExecutorService.shutdown();
  try {

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

@@ -30,10 +30,12 @@
 public final class SnifferBuilder {
  public static final long DEFAULT_SNIFF_INTERVAL = TimeUnit.MINUTES.toMillis(5);
  public static final long DEFAULT_SNIFF_AFTER_FAILURE_DELAY = TimeUnit.MINUTES.toMillis(1);
+ public static final int DEFAULT_HOST_EXCLUDED_SNIFF_ROUNDS = 1;
 
  private final RestClient restClient;
  private long sniffIntervalMillis = DEFAULT_SNIFF_INTERVAL;
  private long sniffAfterFailureDelayMillis = DEFAULT_SNIFF_AFTER_FAILURE_DELAY;
+ private int maxExcludedRounds = DEFAULT_HOST_EXCLUDED_SNIFF_ROUNDS;
  private HostsSniffer hostsSniffer;
 
  /**
@@ -79,13 +81,24 @@ public SnifferBuilder setHostsSniffer(HostsSniffer hostsSniffer) {
  return this;
  }
 
+ /**
+ * Sets the amount of future sniffing calls from which a host that had failed a request would be excluded. Will not
+ * be used if client wasn't built utilizing
+ * {@link org.elasticsearch.client.RestClientBuilder#setFailureListener(RestClient.FailureListener)}
+ * @param maxExcludedRounds the number of sniffing calls to exclude a host from
+ */
+ public SnifferBuilder setMaxExcludedRounds(int maxExcludedRounds) {
+ this.maxExcludedRounds = maxExcludedRounds;
+ return this;
+ }
+
  /**
  * Creates the {@link Sniffer} based on the provided configuration.
  */
  public Sniffer build() {
  if (hostsSniffer == null) {
  this.hostsSniffer = new ElasticsearchHostsSniffer(restClient);
  }
- return new Sniffer(restClient, hostsSniffer, sniffIntervalMillis, sniffAfterFailureDelayMillis);
+ return new Sniffer(restClient, hostsSniffer, sniffIntervalMillis, sniffAfterFailureDelayMillis, maxExcludedRounds);
  }
 }

client/sniffer/src/test/java/org/elasticsearch/client/sniff/documentation/SnifferDocumentation.java

@@ -82,8 +82,9 @@ public void testUsage() throws IOException {
  .build();
  Sniffer sniffer = Sniffer.builder(restClient)
  .setSniffAfterFailureDelayMillis(30000) // <2>
+ .setMaxExcludedRounds(3) // <3>
  .build();
- sniffOnFailureListener.setSniffer(sniffer); // <3>
+ sniffOnFailureListener.setSniffer(sniffer); // <4>
  //end::sniff-on-failure
  }
  {

docs/java-rest/low-level/sniffer.asciidoc

@@ -98,7 +98,8 @@ normal and we want to detect that as soon as possible. Said interval can be
 customized at `Sniffer` creation time through the `setSniffAfterFailureDelayMillis`
 method. Note that this last configuration parameter has no effect in case sniffing
 on failure is not enabled like explained above.
-<3> Set the `Sniffer` instance to the failure listener
+<3> Set the amount of sniffing rounds to exclude a failed host for
+<4> Set the `Sniffer` instance to the failure listener
 
 The Elasticsearch Nodes Info api doesn't return the protocol to use when
 connecting to the nodes but only their `host:port` key-pair, hence `http`