Repository: searchbox-io/Jest
Pull request #288 (2016-01-11 20:28:24)

Save the initial server list so that if NodeChecker fails it falls back
Fix for #287 .  Save the initial server list so that if NodeChecker fails it falls back to the initial server list.  This is useful if the entire cluster goes down and nodes come back up on different IP addresses.  The initial DNS entries can eventually be re-resolved.


Patch:
diff --git a/jest-common/src/main/java/io/searchbox/client/AbstractJestClient.java b/jest-common/src/main/java/io/searchbox/client/AbstractJestClient.java
index 319ee119..2e9d2c04 100644
--- a/jest-common/src/main/java/io/searchbox/client/AbstractJestClient.java
+++ b/jest-common/src/main/java/io/searchbox/client/AbstractJestClient.java
@@ -8,7 +8,7 @@ import com.google.gson.GsonBuilder;
 import io.searchbox.client.config.discovery.NodeChecker;
 import io.searchbox.client.config.exception.NoServerConfiguredException;
 import io.searchbox.client.config.idle.IdleConnectionReaper;
-import org.apache.commons.lang3.tuple.Pair;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,9 +29,8 @@ public abstract class AbstractJestClient implements JestClient {
 
     private final static Logger log = LoggerFactory.getLogger(AbstractJestClient.class);
 
-    // server pool = Pair of (pool size, pool iterator)
-    private final AtomicReference<Pair<Integer, Iterator<String>>> serverPoolReference =
-            new AtomicReference<Pair<Integer, Iterator<String>>>(Pair.<Integer, Iterator<String>>of(0, ImmutableSet.<String>of().iterator()));
+    private final AtomicReference<ServerPool> serverPoolReference =
+            new AtomicReference<ServerPool>(new ServerPool(ImmutableSet.<String>of()));
     private NodeChecker nodeChecker;
     private IdleConnectionReaper idleConnectionReaper;
     private boolean requestCompressionEnabled;
@@ -45,12 +44,21 @@ public abstract class AbstractJestClient implements JestClient {
     }
 
     public void setServers(Set<String> servers) {
-        serverPoolReference.set(Pair.of(servers.size(), Iterators.cycle(servers)));
+        if (servers.equals(serverPoolReference.get().getServers())) {
+            if (log.isDebugEnabled()) {
+                log.debug("Server pool already contains same list of servers: {}",
+                        StringUtils.join(servers, ","));
+            }
+            return;
+        }
+        if (log.isInfoEnabled()) {
+            log.info("Setting server pool to a list of {} servers: [{}]",
+                      servers.size(), StringUtils.join(servers, ","));
+        }
+        serverPoolReference.set(new ServerPool(servers));
 
         if (servers.isEmpty()) {
             log.warn("No servers are currently available to connect.");
-        } else if (log.isDebugEnabled()) {
-            log.debug("Server pool was updated to contain {} servers.", servers.size());
         }
     }
 
@@ -69,15 +77,11 @@ public abstract class AbstractJestClient implements JestClient {
      * @throws io.searchbox.client.config.exception.NoServerConfiguredException
      */
     protected String getNextServer() {
-        Iterator<String> iterator = serverPoolReference.get().getValue();
-        synchronized (iterator) {
-            if (iterator.hasNext()) return iterator.next();
-            else throw new NoServerConfiguredException("No Server is assigned to client to connect");
-        }
+        return serverPoolReference.get().getNextServer();
     }
 
     protected int getServerPoolSize() {
-        return serverPoolReference.get().getKey();
+        return serverPoolReference.get().getSize();
     }
 
     protected String getRequestURL(String elasticSearchServer, String uri) {
@@ -97,4 +101,27 @@ public abstract class AbstractJestClient implements JestClient {
         this.requestCompressionEnabled = requestCompressionEnabled;
     }
 
-}
\ No newline at end of file
+    private static final class ServerPool {
+        private final Set<String> servers;
+        private final Iterator<String> serverIterator;
+
+        public ServerPool(final Set<String> servers) {
+            this.servers = ImmutableSet.copyOf(servers);
+            this.serverIterator = Iterators.cycle(servers);
+        }
+
+        public Set<String> getServers() {
+            return servers;
+        }
+
+        public String getNextServer() {
+            if (serverIterator.hasNext()) return serverIterator.next();
+            else throw new NoServerConfiguredException("No Server is assigned to client to connect");
+        }
+
+        public int getSize() {
+            return servers.size();
+        }
+
+    }
+}
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 c357d5c9..1af8270a 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
@@ -1,5 +1,6 @@
 package io.searchbox.client.config.discovery;
 
+import com.google.common.collect.ImmutableSet;
 import com.google.common.util.concurrent.AbstractScheduledService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -18,6 +19,7 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.LinkedHashSet;
 import java.util.Map.Entry;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -38,12 +40,14 @@ public class NodeChecker extends AbstractScheduledService {
     protected JestClient client;
     protected Scheduler scheduler;
     protected String defaultScheme;
+    protected Set<String> bootstrapServerList;
 
     public NodeChecker(JestClient jestClient, ClientConfig clientConfig) {
-        this(jestClient, clientConfig.getDefaultSchemeForDiscoveredNodes(), clientConfig.getDiscoveryFrequency(), clientConfig.getDiscoveryFrequencyTimeUnit());
+        this(jestClient, clientConfig.getDefaultSchemeForDiscoveredNodes(), clientConfig.getDiscoveryFrequency(), clientConfig.getDiscoveryFrequencyTimeUnit(),
+				clientConfig.getServerList());
     }
 
-    public NodeChecker(JestClient jestClient, String defaultScheme, Long discoveryFrequency, TimeUnit discoveryFrequencyTimeUnit) {
+    public NodeChecker(JestClient jestClient, String defaultScheme, Long discoveryFrequency, TimeUnit discoveryFrequencyTimeUnit, Set<String> servers) {
         this.client = jestClient;
         this.defaultScheme = defaultScheme;
         this.scheduler = Scheduler.newFixedDelaySchedule(
@@ -51,6 +55,7 @@ public class NodeChecker extends AbstractScheduledService {
                 discoveryFrequency,
                 discoveryFrequencyTimeUnit
         );
+		this.bootstrapServerList = ImmutableSet.copyOf(servers);
     }
 
     @Override
@@ -60,6 +65,7 @@ public class NodeChecker extends AbstractScheduledService {
             result = client.execute(action);
         } 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);
@@ -84,10 +90,13 @@ public class NodeChecker extends AbstractScheduledService {
                     }
                 }
             }
-            log.info("Discovered {} HTTP hosts: {}", httpHosts.size(), StringUtils.join(httpHosts, ","));
+            if (log.isDebugEnabled()) {
+                log.debug("Discovered {} HTTP hosts: {}", httpHosts.size(), StringUtils.join(httpHosts, ","));
+            }
             client.setServers(httpHosts);
         } else {
             log.warn("NodesInfo request resulted in error: {}", result.getErrorMessage());
+            client.setServers(bootstrapServerList);
         }
     }
 


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 7d701740..4cb43b3a 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,6 +1,7 @@
 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;
@@ -203,4 +204,41 @@ public class NodeCheckerTest {
         assertEquals(0, servers.size());
     }
 
+    @Test
+    public void testNodesInfoExceptionUsesBootstrapServerList() throws Exception {
+        NodeChecker nodeChecker = new NodeChecker(jestClient, clientConfig);
+
+        when(jestClient.execute(isA(Action.class))).thenThrow(Exception.class);
+
+        nodeChecker.runOneIteration();
+
+        verify(jestClient).execute(isA(Action.class));
+        ArgumentCaptor<LinkedHashSet> argument = ArgumentCaptor.forClass(LinkedHashSet.class);
+        verify(jestClient).setServers(argument.capture());
+        verifyNoMoreInteractions(jestClient);
+
+        Set servers = argument.getValue();
+        assertEquals(1, servers.size());
+        assertEquals("http://localhost:9200", servers.iterator().next());
+    }
+
+    @Test
+    public void testNodesInfoFailureUsesBootstrapServerList() throws Exception {
+        NodeChecker nodeChecker = new NodeChecker(jestClient, clientConfig);
+        JestResult result = new JestResult(new Gson());
+        result.setSucceeded(false);
+        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());
+        verifyNoMoreInteractions(jestClient);
+
+        Set servers = argument.getValue();
+        assertEquals(1, servers.size());
+        assertEquals("http://localhost:9200", servers.iterator().next());
+    }
+
 }