Repository: searchbox-io/Jest
Pull request #317 (2016-03-22 09:21:09)

Backport - 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`.

This is a backport of #316 to the 1.0.x branch.


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);
+        }
+    }
 }