Repository: searchbox-io/Jest
Pull request #367 (2016-06-02 01:50:11)

Feature/node checker remove failed nodes
Hi @kramer I fixed up @bmlining's PR, hopefully this works for everyone


Patch:
diff --git a/jest-common/src/main/java/io/searchbox/client/config/discovery/NodeChecker.java b/jest-common/src/main/java/io/searchbox/client/config/discovery/NodeChecker.java
index e7fe7bc6..efead34a 100644
--- a/jest-common/src/main/java/io/searchbox/client/config/discovery/NodeChecker.java
+++ b/jest-common/src/main/java/io/searchbox/client/config/discovery/NodeChecker.java
@@ -6,19 +6,22 @@ import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
+
 import io.searchbox.client.JestClient;
 import io.searchbox.client.JestResult;
 import io.searchbox.client.config.ClientConfig;
+import io.searchbox.client.config.exception.CouldNotConnectException;
 import io.searchbox.cluster.NodesInfo;
+
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.LinkedHashSet;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -39,6 +42,7 @@ public class NodeChecker extends AbstractScheduledService {
     protected Scheduler scheduler;
     protected String defaultScheme;
     protected Set<String> bootstrapServerList;
+    protected Set<String> discoveredServerList;
 
     public NodeChecker(JestClient jestClient, ClientConfig clientConfig) {
         action = new NodesInfo.Builder()
@@ -60,13 +64,20 @@ public class NodeChecker extends AbstractScheduledService {
         JestResult result;
         try {
             result = client.execute(action);
+        } catch (CouldNotConnectException cnce) {
+            // Can't connect to this node, remove it from the list
+            log.error("Connect exception executing NodesInfo!", cnce);
+            removeNodeAndUpdateServers(cnce.getHost());
+            return;
+            // do not elevate the exception since that will stop the scheduled calls.
+            // throw new RuntimeException("Error executing NodesInfo!", e);
         } catch (Exception e) {
             log.error("Error executing NodesInfo!", e);
             client.setServers(bootstrapServerList);
             return;
             // do not elevate the exception since that will stop the scheduled calls.
             // throw new RuntimeException("Error executing NodesInfo!", e);
-        }
+        }  
 
         if (result.isSucceeded()) {
             LinkedHashSet<String> httpHosts = new LinkedHashSet<String>();
@@ -90,13 +101,27 @@ public class NodeChecker extends AbstractScheduledService {
             if (log.isDebugEnabled()) {
                 log.debug("Discovered {} HTTP hosts: {}", httpHosts.size(), StringUtils.join(httpHosts, ","));
             }
-            client.setServers(httpHosts);
+            discoveredServerList = httpHosts;
+            client.setServers(discoveredServerList);
         } else {
             log.warn("NodesInfo request resulted in error: {}", result.getErrorMessage());
             client.setServers(bootstrapServerList);
         }
     }
 
+    protected void removeNodeAndUpdateServers(final String hostToRemove) {
+        log.warn("Removing host {}", hostToRemove);
+        discoveredServerList.remove(hostToRemove);
+        if (log.isInfoEnabled()) {
+            log.info("Discovered server pool is now: {}", StringUtils.join(discoveredServerList, ","));
+        }
+        if (!discoveredServerList.isEmpty()) {
+          client.setServers(discoveredServerList);
+        } else {
+          client.setServers(bootstrapServerList);
+        }
+    }
+
     @Override
     protected Scheduler scheduler() {
         return scheduler;
diff --git a/jest-common/src/main/java/io/searchbox/client/config/exception/CouldNotConnectException.java b/jest-common/src/main/java/io/searchbox/client/config/exception/CouldNotConnectException.java
new file mode 100644
index 00000000..79db9e64
--- /dev/null
+++ b/jest-common/src/main/java/io/searchbox/client/config/exception/CouldNotConnectException.java
@@ -0,0 +1,21 @@
+package io.searchbox.client.config.exception;
+
+import java.io.IOException;
+
+/**
+ * Wrapper for a the host address of an HTTPConnectException
+ * @author Brian Harrington
+ */
+public class CouldNotConnectException extends IOException {
+
+    private final String host;
+
+    public CouldNotConnectException(String host, Throwable cause) {
+        super("Could not connect to " + host, cause);
+        this.host = host;
+    }
+
+    public String getHost() {
+        return host;
+    }
+}
diff --git a/jest-droid/src/main/java/com/searchly/jestdroid/JestDroidClient.java b/jest-droid/src/main/java/com/searchly/jestdroid/JestDroidClient.java
index 297a4b9a..b2e44fa0 100644
--- a/jest-droid/src/main/java/com/searchly/jestdroid/JestDroidClient.java
+++ b/jest-droid/src/main/java/com/searchly/jestdroid/JestDroidClient.java
@@ -8,6 +8,7 @@ import io.searchbox.client.AbstractJestClient;
 import io.searchbox.client.JestClient;
 import io.searchbox.client.JestResult;
 import io.searchbox.client.JestResultHandler;
+import io.searchbox.client.config.exception.CouldNotConnectException;
 import org.apache.http.HttpEntityEnclosingRequest;
 import org.apache.http.HttpResponse;
 import org.apache.http.StatusLine;
@@ -17,6 +18,7 @@ import org.apache.http.client.methods.HttpHeadHC4;
 import org.apache.http.client.methods.HttpPostHC4;
 import org.apache.http.client.methods.HttpPutHC4;
 import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.conn.HttpHostConnectException;
 import org.apache.http.entity.ContentType;
 import org.apache.http.util.EntityUtilsHC4;
 import org.slf4j.Logger;
@@ -48,9 +50,12 @@ public class JestDroidClient extends AbstractJestClient implements JestClient {
             }
         }
 
-        HttpResponse response = httpClient.execute(request);
-
-        return deserializeResponse(response, clientRequest);
+        try {
+            HttpResponse response = httpClient.execute(request);
+            return deserializeResponse(response, clientRequest);
+        } catch (HttpHostConnectException ex) {
+            throw new CouldNotConnectException(ex.getHost().toURI(), ex);
+        }
     }
 
     @Override
diff --git a/jest/src/main/java/io/searchbox/client/http/JestHttpClient.java b/jest/src/main/java/io/searchbox/client/http/JestHttpClient.java
index 665035fe..395d4888 100755
--- a/jest/src/main/java/io/searchbox/client/http/JestHttpClient.java
+++ b/jest/src/main/java/io/searchbox/client/http/JestHttpClient.java
@@ -4,6 +4,7 @@ import io.searchbox.action.Action;
 import io.searchbox.client.AbstractJestClient;
 import io.searchbox.client.JestResult;
 import io.searchbox.client.JestResultHandler;
+import io.searchbox.client.config.exception.CouldNotConnectException;
 import io.searchbox.client.http.apache.HttpDeleteWithEntity;
 import io.searchbox.client.http.apache.HttpGetWithEntity;
 import java.io.IOException;
@@ -19,6 +20,7 @@ import org.apache.http.client.methods.HttpPut;
 import org.apache.http.client.methods.HttpUriRequest;
 import org.apache.http.client.protocol.HttpClientContext;
 import org.apache.http.concurrent.FutureCallback;
+import org.apache.http.conn.HttpHostConnectException;
 import org.apache.http.entity.ContentType;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
@@ -45,13 +47,17 @@ public class JestHttpClient extends AbstractJestClient {
     /**
      * @throws IOException in case of a problem or the connection was aborted during request,
      *                     or in case of a problem while reading the response stream
+     * @throws CouldNotConnectException if an {@link HttpHostConnectException} is encountered
      */
     @Override
     public <T extends JestResult> T execute(Action<T> clientRequest) throws IOException {
         HttpUriRequest request = prepareRequest(clientRequest);
-        HttpResponse response = executeRequest(request);
-
-        return deserializeResponse(response, request, clientRequest);
+        try {
+            HttpResponse response = executeRequest(request);
+            return deserializeResponse(response, request, clientRequest);
+        } catch (HttpHostConnectException ex) {
+            throw new CouldNotConnectException(ex.getHost().toURI(), ex);
+        }
     }
 
     @Override
@@ -234,6 +240,11 @@ public class JestHttpClient extends AbstractJestClient {
         @Override
         public void failed(final Exception ex) {
             log.error("Exception occurred during async execution.", ex);
+            if (ex instanceof HttpHostConnectException) {
+                String host = ((HttpHostConnectException) ex).getHost().toURI();
+                resultHandler.failed(new CouldNotConnectException(host, ex));
+                return;
+            }
             resultHandler.failed(ex);
         }
 


Test patch:
diff --git a/jest-common/src/test/java/io/searchbox/client/config/discovery/NodeCheckerTest.java b/jest-common/src/test/java/io/searchbox/client/config/discovery/NodeCheckerTest.java
index 4cb43b3a..695b1846 100644
--- a/jest-common/src/test/java/io/searchbox/client/config/discovery/NodeCheckerTest.java
+++ b/jest-common/src/test/java/io/searchbox/client/config/discovery/NodeCheckerTest.java
@@ -1,23 +1,29 @@
 package io.searchbox.client.config.discovery;
 
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
 import com.google.gson.Gson;
 import io.searchbox.action.Action;
 import io.searchbox.client.JestClient;
 import io.searchbox.client.JestResult;
 import io.searchbox.client.config.ClientConfig;
+import io.searchbox.client.config.exception.CouldNotConnectException;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 
+import java.io.IOException;
+import java.util.Iterator;
 import java.util.LinkedHashSet;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertEquals;
 import static org.mockito.Matchers.isA;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
 
 public class NodeCheckerTest {
 
@@ -241,4 +247,77 @@ public class NodeCheckerTest {
         assertEquals("http://localhost:9200", servers.iterator().next());
     }
 
+
+    @Test
+    public void testNodesInfoExceptionRemovesServerFromList() throws Exception {
+        NodeChecker nodeChecker = new NodeChecker(jestClient, clientConfig);
+
+        JestResult result = new JestResult(new Gson());
+        result.setJsonMap(ImmutableMap.<String, Object>of(
+                "ok", "true",
+                "nodes", ImmutableMap.of(
+                        "node1", ImmutableMap.of(
+                                "http_address", "inet[/192.168.2.7:9200]"),
+                        "node2", ImmutableMap.of(
+                                "http_address", "inet[/192.168.2.8:9200]"),
+                        "node3", ImmutableMap.of(
+                                "http_address", "inet[/192.168.2.9:9200]"))));
+        result.setSucceeded(true);
+        when(jestClient.execute(isA(Action.class))).thenReturn(result);
+        nodeChecker.runOneIteration();
+
+        verify(jestClient).execute(isA(Action.class));
+        ArgumentCaptor<LinkedHashSet> argument = ArgumentCaptor.forClass(LinkedHashSet.class);
+        verify(jestClient).setServers(argument.capture());
+        verify(jestClient).execute(isA(Action.class));
+
+        Set servers = argument.getValue();
+        assertEquals(3, servers.size());
+
+//      	when(jestClient.execute(isA(Action.class))).thenThrow(new HttpHostConnectException(
+//            new ConnectException(), new HttpHost("192.168.2.7", 9200, "http"), null));
+        when(jestClient.execute(isA(Action.class))).thenThrow(new CouldNotConnectException("http://192.168.2.7:9200", new IOException("Test HttpHostException")));
+        nodeChecker.runOneIteration();
+
+        verify(jestClient, times(2)).execute(isA(Action.class));
+        verify(jestClient, times(2)).setServers(argument.capture());
+        verifyNoMoreInteractions(jestClient);
+
+        servers = argument.getValue();
+        assertEquals(2, servers.size());
+        Iterator serversItr = servers.iterator();
+        assertEquals("http://192.168.2.8:9200", serversItr.next());
+        assertEquals("http://192.168.2.9:9200", serversItr.next());
+
+        // fail at the 2nd node
+        //      when(jestClient.execute(isA(Action.class))).thenThrow(new HttpHostConnectException(
+        //            new ConnectException(), new HttpHost("192.168.2.8", 9200, "http"), null));
+        when(jestClient.execute(isA(Action.class))).thenThrow(new CouldNotConnectException("http://192.168.2.8:9200", new IOException("Test HttpHostException")));
+        nodeChecker.runOneIteration();
+
+        verify(jestClient, times(3)).execute(isA(Action.class));
+        verify(jestClient, times(3)).setServers(argument.capture());
+        verifyNoMoreInteractions(jestClient);
+
+        servers = argument.getValue();
+        assertEquals(1, servers.size());
+        serversItr = servers.iterator();
+        assertEquals("http://192.168.2.9:9200", serversItr.next());
+
+        // fail at the last node, fail back to bootstrap
+//      	when(jestClient.execute(isA(Action.class))).thenThrow(new HttpHostConnectException(
+//            new ConnectException(), new HttpHost("192.168.2.9", 9200, "http"), null));
+        when(jestClient.execute(isA(Action.class))).thenThrow(new CouldNotConnectException("http://192.168.2.9:9200", new IOException("Test HttpHostException")));
+        nodeChecker.runOneIteration();
+
+        verify(jestClient, times(4)).execute(isA(Action.class));
+        verify(jestClient, times(4)).setServers(argument.capture());
+        verifyNoMoreInteractions(jestClient);
+
+        servers = argument.getValue();
+        assertEquals(1, servers.size());
+        serversItr = servers.iterator();
+        assertEquals("http://localhost:9200", serversItr.next());
+    }
+
 }