[DE-131] Bugfix activefailover concurrency

docker/start_db.sh

@@ -96,3 +96,9 @@ for a in ${COORDINATORS[*]} ; do
     echo "$SCHEME://$a"
     echo ""
 done
+
+if [ "$STARTER_MODE" == "activefailover" ]; then
+  LEADER=$("$LOCATION"/find_active_endpoint.sh)
+  echo "Leader: $SCHEME://$LEADER"
+  echo ""
+fi

src/main/java/com/arangodb/async/internal/velocystream/VstCommunicationAsync.java

@@ -85,8 +85,7 @@ protected CompletableFuture<Response> execute(final Request request, final VstCo
                         }
                         final String location = e.getLocation();
                         final HostDescription redirectHost = HostUtils.createFromLocation(location);
-                        hostHandler.closeCurrentOnError();
-                        hostHandler.fail(e);
+                        hostHandler.failIfNotMatch(redirectHost, e);
                         execute(request, new HostHandle().setHost(redirectHost), attemptCount + 1)
                                 .whenComplete((v, err) -> {
                                     if (v != null) {

src/main/java/com/arangodb/internal/http/HttpCommunication.java

@@ -106,8 +106,7 @@ private Response execute(final Request request, final HostHandle hostHandle, fin
             if (e instanceof ArangoDBRedirectException && attemptCount < 3) {
                 final String location = ((ArangoDBRedirectException) e).getLocation();
                 final HostDescription redirectHost = HostUtils.createFromLocation(location);
-                hostHandler.closeCurrentOnError();
-                hostHandler.fail(e);
+                hostHandler.failIfNotMatch(redirectHost, e);
                 return execute(request, new HostHandle().setHost(redirectHost), attemptCount + 1);
             } else {
                 throw e;

src/main/java/com/arangodb/internal/net/DirtyReadHostHandler.java

@@ -60,6 +60,11 @@ public void fail(Exception exception) {
         determineHostHandler().fail(exception);
     }
 
+    @Override
+    public void failIfNotMatch(HostDescription host, Exception exception) {
+        determineHostHandler().failIfNotMatch(host, exception);
+    }
+
     @Override
     public void reset() {
         determineHostHandler().reset();
@@ -81,6 +86,11 @@ public void closeCurrentOnError() {
         determineHostHandler().closeCurrentOnError();
     }
 
+    @Override
+    public void closeCurrentOnErrorIfNotMatch(HostDescription host) {
+        determineHostHandler().closeCurrentOnErrorIfNotMatch(host);
+    }
+
     @Override
     public void setJwt(String jwt) {
         master.setJwt(jwt);

src/main/java/com/arangodb/internal/net/FallbackHostHandler.java

@@ -79,6 +79,13 @@ public void fail(Exception exception) {
         lastFailExceptions.add(exception);
     }
 
+    @Override
+    public synchronized void failIfNotMatch(HostDescription host, Exception exception) {
+        if (!host.equals(current.getDescription())) {
+            fail(exception);
+        }
+    }
+
     @Override
     public void reset() {
         iterations = 0;
@@ -104,6 +111,13 @@ public void closeCurrentOnError() {
         current.closeOnError();
     }
 
+    @Override
+    public synchronized void closeCurrentOnErrorIfNotMatch(HostDescription host) {
+        if (!host.equals(current.getDescription())) {
+            closeCurrentOnError();
+        }
+    }
+
     @Override
     public void setJwt(String jwt) {
         hosts.setJwt(jwt);

src/main/java/com/arangodb/internal/net/HostHandler.java

@@ -33,6 +33,8 @@ public interface HostHandler {
 
     void fail(Exception exception);
 
+    void failIfNotMatch(HostDescription host, Exception exception);
+
     void reset();
 
     void confirm();
@@ -41,6 +43,8 @@ public interface HostHandler {
 
     void closeCurrentOnError();
 
+    void closeCurrentOnErrorIfNotMatch(HostDescription host);
+
     void setJwt(String jwt);
 
 }

src/main/java/com/arangodb/internal/net/RandomHostHandler.java

@@ -59,6 +59,13 @@ public void fail(Exception exception) {
         current = fallback.get(null, null);
     }
 
+    @Override
+    public synchronized void failIfNotMatch(HostDescription host, Exception exception) {
+        if (!host.equals(current.getDescription())) {
+            fail(exception);
+        }
+    }
+
     private Host getRandomHost(final boolean initial, final boolean closeConnections) {
         hosts = resolver.resolve(initial, closeConnections);
         final ArrayList<Host> hostList = new ArrayList<>(hosts.getHostsList());
@@ -85,6 +92,13 @@ public void closeCurrentOnError() {
         current.closeOnError();
     }
 
+    @Override
+    public synchronized void closeCurrentOnErrorIfNotMatch(HostDescription host) {
+        if (!host.equals(current.getDescription())) {
+            closeCurrentOnError();
+        }
+    }
+
     @Override
     public void setJwt(String jwt) {
         fallback.setJwt(jwt);

src/main/java/com/arangodb/internal/net/RoundRobinHostHandler.java

@@ -89,6 +89,11 @@ public void fail(Exception exception) {
         lastFailExceptions.add(exception);
     }
 
+    @Override
+    public void failIfNotMatch(HostDescription host, Exception exception) {
+        fail(exception);
+    }
+
     @Override
     public void reset() {
         fails = 0;
@@ -109,6 +114,11 @@ public void closeCurrentOnError() {
         currentHost.closeOnError();
     }
 
+    @Override
+    public void closeCurrentOnErrorIfNotMatch(HostDescription host) {
+        closeCurrentOnError();
+    }
+
     @Override
     public void setJwt(String jwt) {
         hosts.setJwt(jwt);

src/main/java/com/arangodb/internal/velocystream/VstCommunicationSync.java

@@ -151,8 +151,7 @@ protected Response execute(final Request request, final VstConnectionSync connec
             }
             final String location = e.getLocation();
             final HostDescription redirectHost = HostUtils.createFromLocation(location);
-            hostHandler.closeCurrentOnError();
-            hostHandler.fail(e);
+            hostHandler.failIfNotMatch(redirectHost, e);
             return execute(request, new HostHandle().setHost(redirectHost), attemptCount + 1);
         }
     }

src/test/java/com/arangodb/ArangoDBTest.java

@@ -44,6 +44,9 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -725,13 +728,15 @@ public void accessMultipleDatabases() {
     }
 
     @Test
-    public void queueTime() throws InterruptedException {
-        List<Thread> threads = IntStream.range(0, 80)
-                .mapToObj(__ -> new Thread(() -> arangoDB.db().query("RETURN SLEEP(1)", Void.class)))
+    public void queueTime() throws InterruptedException, ExecutionException {
+        List<CompletableFuture<Void>> futures = IntStream.range(0, 80)
+                .mapToObj(i -> CompletableFuture.runAsync(
+                        () -> arangoDB.db().query("RETURN SLEEP(1)", Void.class),
+                        Executors.newFixedThreadPool(80))
+                )
                 .collect(Collectors.toList());
-        threads.forEach(Thread::start);
-        for (Thread it : threads) {
-            it.join();
+        for (CompletableFuture<Void> f : futures) {
+            f.get();
         }
 
         QueueTimeMetrics qt = arangoDB.metrics().getQueueTime();

src/test/java/com/arangodb/ConcurrencyTests.java

@@ -0,0 +1,43 @@
+package com.arangodb;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+@RunWith(Parameterized.class)
+public class ConcurrencyTests {
+
+    final Protocol protocol;
+
+    public ConcurrencyTests(Protocol protocol) {
+        this.protocol = protocol;
+    }
+
+    @Parameterized.Parameters
+    public static Protocol[] protocols() {
+        return Protocol.values();
+    }
+
+    @Test
+    public void concurrentPendingRequests() throws ExecutionException, InterruptedException {
+        ArangoDB adb = new ArangoDB.Builder().useProtocol(protocol).build();
+        List<CompletableFuture<Void>> futures = IntStream.range(0, 10)
+                .mapToObj(i -> CompletableFuture.runAsync(
+                        () -> adb.db().query("RETURN SLEEP(1)", Void.class),
+                        Executors.newFixedThreadPool(10))
+                )
+                .collect(Collectors.toList());
+        for (CompletableFuture<Void> f : futures) {
+            f.get();
+        }
+        adb.shutdown();
+    }
+
+}

src/test/java/com/arangodb/async/ConcurrencyTests.java

@@ -0,0 +1,25 @@
+package com.arangodb.async;
+
+import org.junit.Test;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+public class ConcurrencyTests {
+
+    @Test
+    public void concurrentPendingRequests() throws ExecutionException, InterruptedException {
+        ArangoDBAsync adb = new ArangoDBAsync.Builder().build();
+        List<CompletableFuture<ArangoCursorAsync<Void>>> reqs = IntStream.range(0, 10)
+                .mapToObj(__ -> adb.db().query("RETURN SLEEP(1)", Void.class))
+                .collect(Collectors.toList());
+        for (CompletableFuture<ArangoCursorAsync<Void>> req : reqs) {
+            req.get();
+        }
+        adb.shutdown();
+    }
+
+}