Repository: searchbox-io/Jest (java)

Issue:
NoSuchElementException - Concurrency issue on accessing the servers
FIrst of all thank you for this amazing library. 
Now to the point, I'm doing a stress test where there is only one ES node. Everything works great, but randomly I'm receiving the error mentioned in the title and below. I'm using one ES node. I'm using Jest 1.0.1 with ES 1.7.2

```
JestClientFactory factory = new JestClientFactory();
factory.setHttpClientConfig(new HttpClientConfig
                .Builder(restEndpoint)
                .multiThreaded(true)
                .defaultCredentials(user, password)
                .requestCompressionEnabled(false)
                .build());

this.client = factory.getObject();
```

> java.util.NoSuchElementException: null
> 	at java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:713)
> 	at java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.jaThva:734)
> 	at com.google.common.collect.Iterators$4.next(Iterators.java:411)
> 	at io.searchbox.client.AbstractJestClient$ServerPool.getNextServer(AbstractJestClient.java:118)
> 	at io.searchbox.client.AbstractJestClient.getNextServer(AbstractJestClient.java:80)
> 	at io.searchbox.client.http.JestHttpClient.prepareRequest(JestHttpClient.java:80)
> 	at io.searchbox.client.http.JestHttpClient.execute(JestHttpClient.java:46)

The query is valid, as I've tested it via the marvel plugin. 

Pull request:
Use a server ring instead of a cycling iterator
Guava's cycling iterator does not seem to be threadsafe, as it replaces the underlying iterator between
calling `hasNext()` and `next()`, which can cause a `ConcurrentModificationException`.

Taking into account the research and design suggestions from @alkiskal and @jasonculverhouse, here
is a reworked version of #313 for #311 which simply keeps a next index pointer, and a list of servers
that acts as a ring.

Closes #313
Closes #311

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 2e9d2c04..4afab801 100644
--- a/jest-common/src/main/java/io/searchbox/client/AbstractJestClient.java
+++ b/jest-common/src/main/java/io/searchbox/client/AbstractJestClient.java
@@ -1,8 +1,8 @@
 package io.searchbox.client;
 
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Iterators;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import io.searchbox.client.config.discovery.NodeChecker;
@@ -12,8 +12,9 @@ import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Iterator;
+import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -102,26 +103,38 @@ public abstract class AbstractJestClient implements JestClient {
     }
 
     private static final class ServerPool {
-        private final Set<String> servers;
-        private final Iterator<String> serverIterator;
+        private final List<String> serversRing;
+        private final AtomicInteger nextServerIndex = new AtomicInteger(0);
 
         public ServerPool(final Set<String> servers) {
-            this.servers = ImmutableSet.copyOf(servers);
-            this.serverIterator = Iterators.cycle(servers);
+            this.serversRing = ImmutableList.copyOf(servers);
         }
 
         public Set<String> getServers() {
-            return servers;
+            return ImmutableSet.copyOf(serversRing);
         }
 
         public String getNextServer() {
-            if (serverIterator.hasNext()) return serverIterator.next();
-            else throw new NoServerConfiguredException("No Server is assigned to client to connect");
+            if (serversRing.size() > 0) {
+                try {
+                    return serversRing.get(nextServerIndex.getAndIncrement() % serversRing.size());
+                } catch (IndexOutOfBoundsException outOfBoundsException) {
+                    // In the very rare case where nextServerIndex overflowed, this will end up with a negative number,
+                    // resulting in an IndexOutOfBoundsException.
+                    // We should then start back at the beginning of the server list.
+                    // Note that this might happen on several threads at once, in which the reset might happen a few times
+                    log.info("Resetting next server index");
+                    nextServerIndex.set(0);
+                    return serversRing.get(nextServerIndex.getAndIncrement() % serversRing.size());
+                }
+            }
+            else {
+                throw new NoServerConfiguredException("No Server is assigned to client to connect");
+            }
         }
 
         public int getSize() {
-            return servers.size();
+            return serversRing.size();
         }
-
     }
 }


Test patch:
diff --git a/jest-common/src/test/java/io/searchbox/client/AbstractJestClientTest.java b/jest-common/src/test/java/io/searchbox/client/AbstractJestClientTest.java
index ad7695f0..bbbaddce 100644
--- a/jest-common/src/test/java/io/searchbox/client/AbstractJestClientTest.java
+++ b/jest-common/src/test/java/io/searchbox/client/AbstractJestClientTest.java
@@ -1,12 +1,18 @@
 package io.searchbox.client;
 
+import com.google.common.collect.Maps;
 import io.searchbox.action.Action;
 import org.junit.Test;
 
 import java.io.IOException;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
+import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static junit.framework.Assert.assertEquals;
 import static junit.framework.Assert.assertTrue;
@@ -56,4 +62,68 @@ public class AbstractJestClientTest {
         assertTrue(set.contains("http://localhost:9300"));
         assertTrue(set.contains("http://localhost:9400"));
     }
+
+    @Test
+    public void testGetElasticSearchServerIsThreadsafe() throws Exception {
+        final int NUM_THREADS = 12;
+        final int NUM_ITERATIONS = 12000;
+        final int MIN_ACCEPTABLE_PER_SERVER = 3900;
+        final int MAX_ACCEPTABLE_PER_SERVER = 4100;
+
+        // do NUM_ITERATIONS of getNextServer, across NUM_THREADS
+        // we should ensure that no exceptions are thrown,
+        // and that we get a rather even share of results for each possible server
+
+        final Set<String> servers = new LinkedHashSet<String>();
+        servers.add("http://localhost:9200");
+        servers.add("http://localhost:9300");
+        servers.add("http://localhost:9400");
+        client.setServers(servers);
+
+        final Map<String, AtomicInteger> hits = Maps.newConcurrentMap();
+        for (String server : servers) {
+            hits.put(server, new AtomicInteger());
+        }
+
+        final AtomicInteger numExceptions = new AtomicInteger();
+        final ExecutorService pool = Executors.newFixedThreadPool(NUM_THREADS);
+
+        for (int i = 0; i < NUM_ITERATIONS; i++) {
+            if (numExceptions.get() == 0) { // don't bother submitting more if there are exceptions already
+                pool.submit(new Runnable() {
+                    @Override
+                    public void run() {
+                        try {
+                            final String nextServer = client.getNextServer();
+                            if (nextServer == null) {
+                                throw new IllegalStateException("acquired null server!");
+                            } else if (!hits.containsKey(nextServer)) {
+                                throw new IllegalStateException("acquired server " + nextServer + ", but this is unknown!");
+                            } else {
+                                hits.get(nextServer).incrementAndGet();
+                            }
+                        } catch (Throwable t) {
+                            System.err.println("Error occurred: " + t.getMessage());
+                            t.printStackTrace();
+                            numExceptions.incrementAndGet();
+                        }
+                    }
+                });
+            } else {
+                System.err.println("Exception detected, not submitting more.");
+                break;
+            }
+        }
+
+        pool.shutdown();
+        pool.awaitTermination(5, TimeUnit.SECONDS);
+
+        assertEquals("should see 0 exceptions", 0, numExceptions.get());
+
+        System.out.println(hits);
+        for (Map.Entry<String, AtomicInteger> entry : hits.entrySet()) {
+            assertTrue("should have roughly same entries in each of the hits buckets but " + entry.getKey() + " has " + entry.getValue(),
+                    entry.getValue().get() >= MIN_ACCEPTABLE_PER_SERVER && entry.getValue().get() <= MAX_ACCEPTABLE_PER_SERVER);
+        }
+    }
 }