ISPN-15239 HotRod client repeatedly retries iteration operation again…

…st failed server after removal from topology
infinispan · Oct 24, 2023 · 9199e2b · 9199e2b
1 parent 96de0b5
commit 9199e2b
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 2 deletions.
diff --git a/...rc/main/java/org/infinispan/client/hotrod/impl/iteration/RemoteInnerPublisherHandler.java b/...rc/main/java/org/infinispan/client/hotrod/impl/iteration/RemoteInnerPublisherHandler.java
@@ -1,6 +1,7 @@
 package org.infinispan.client.hotrod.impl.iteration;
 
 import java.lang.invoke.MethodHandles;
+import java.net.ConnectException;
 import java.net.SocketAddress;
 import java.util.List;
 import java.util.Map;
@@ -112,7 +113,7 @@ protected long handleNextResponse(IterationNextResponse<K, E> nextResponse, Map.
 
    @Override
    protected void handleThrowableInResponse(Throwable t, Map.Entry<SocketAddress, IntSet> target) {
-      if (t instanceof TransportException || t instanceof RemoteIllegalLifecycleStateException) {
+      if (t instanceof TransportException || t instanceof RemoteIllegalLifecycleStateException || t instanceof ConnectException) {
          log.throwableDuringPublisher(t);
          if (log.isTraceEnabled()) {
             IntSet targetSegments = target.getValue();

diff --git a/...rod-client/src/main/java/org/infinispan/client/hotrod/impl/iteration/RemotePublisher.java b/...rod-client/src/main/java/org/infinispan/client/hotrod/impl/iteration/RemotePublisher.java
@@ -130,6 +130,7 @@ protected void handleThrowableInResponse(Throwable t, Map.Entry<SocketAddress, I
                int batchSize = (this.batchSize / actualTargets.size()) + 1;
                return Flowable.fromIterable(actualTargets.entrySet())
                      .map(entry -> {
+                        log.tracef("Requesting next for: %s", entry);
                         RemoteInnerPublisherHandler<K, E> innerHandler = new RemoteInnerPublisherHandler<>(this,
                               batchSize, () -> null, entry);
                         return innerHandler.startPublisher();

diff --git a/.../src/test/java/org/infinispan/client/hotrod/impl/iteration/BaseIterationFailOverTest.java b/.../src/test/java/org/infinispan/client/hotrod/impl/iteration/BaseIterationFailOverTest.java
@@ -6,6 +6,7 @@
 import static org.testng.AssertJUnit.assertEquals;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -14,6 +15,7 @@
 import org.infinispan.client.hotrod.test.HotRodClientTestingUtil;
 import org.infinispan.client.hotrod.test.MultiHotRodServersTest;
 import org.infinispan.commons.util.CloseableIterator;
+import org.infinispan.commons.util.CloseableIteratorSet;
 import org.infinispan.configuration.cache.ConfigurationBuilder;
 import org.infinispan.protostream.SerializationContextInitializer;
 import org.infinispan.query.dsl.embedded.DslSCI;
@@ -74,6 +76,24 @@ public void testFailOver() throws InterruptedException {
       assertEquals(rangeAsSet(0, cacheSize), keys);
    }
 
+   @Test(groups = "functional")
+   public void testServerLeaveButIterateTwice() throws InterruptedException {
+      testFailOver();
+      int cacheSize = 1_000;
+
+      RemoteCache<Integer, AccountHS> cache = clients.get(0).getCache();
+      Collection<Map.Entry<Object, Object>> entries = new ArrayList<>();
+      CloseableIteratorSet<Map.Entry<Integer, AccountHS>> iterator = cache.entrySet();
+      for (Map.Entry<Integer, AccountHS> entry : iterator) {
+         entries.add((Map.Entry) entry);
+      }
+
+      assertEquals(cacheSize, entries.size());
+
+      Set<Integer> keys = extractKeys(entries);
+      assertEquals(rangeAsSet(0, cacheSize), keys);
+   }
+
    protected void killAnIterationServer() {
       servers.stream()
             .filter(s -> s.getIterationManager().activeIterations() > 0)

diff --git a/...otrod/src/main/java/org/infinispan/hotrod/impl/iteration/RemoteInnerPublisherHandler.java b/...otrod/src/main/java/org/infinispan/hotrod/impl/iteration/RemoteInnerPublisherHandler.java
@@ -1,6 +1,7 @@
 package org.infinispan.hotrod.impl.iteration;
 
 import java.lang.invoke.MethodHandles;
+import java.net.ConnectException;
 import java.net.SocketAddress;
 import java.util.List;
 import java.util.Map;
@@ -113,7 +114,7 @@ protected long handleNextResponse(IterationNextResponse<K, E> nextResponse, Map.
 
    @Override
    protected void handleThrowableInResponse(Throwable t, Map.Entry<SocketAddress, IntSet> target) {
-      if (t instanceof TransportException || t instanceof RemoteIllegalLifecycleStateException) {
+      if (t instanceof TransportException || t instanceof RemoteIllegalLifecycleStateException || t instanceof ConnectException) {
          log.throwableDuringPublisher(t);
          if (log.isTraceEnabled()) {
             IntSet targetSegments = target.getValue();