Renew slot cache if ClusterPipeline.sync results in MOVED response

src/main/java/redis/clients/jedis/ClusterPipeline.java

@@ -3,6 +3,7 @@
 import java.time.Duration;
 import java.util.Set;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import redis.clients.jedis.exceptions.JedisMovedDataException;
 import redis.clients.jedis.providers.ClusterConnectionProvider;
 import redis.clients.jedis.util.IOUtils;
 
@@ -55,6 +56,16 @@ public void close() {
     }
   }
 
+  @Override
+  public void sync() {
+    try {
+      super.sync();
+    } catch (JedisMovedDataException e) {
+      provider.renewSlotCache();
+      throw e;
+    }
+  }
+
   @Override
   protected HostAndPort getNodeKey(CommandArguments args) {
     return provider.getNode(((ClusterCommandArguments) args).getCommandHashSlot());

src/main/java/redis/clients/jedis/MultiNodePipelineBase.java

@@ -1,19 +1,22 @@
 package redis.clients.jedis;
 
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import redis.clients.jedis.exceptions.JedisConnectionException;
+import redis.clients.jedis.exceptions.JedisException;
 import redis.clients.jedis.graph.GraphCommandObjects;
 import redis.clients.jedis.providers.ConnectionProvider;
 import redis.clients.jedis.util.IOUtils;
@@ -91,23 +94,24 @@ public void close() {
   }
 
   @Override
-  public final void sync() {
+  public void sync() {
     if (syncing) {
       return;
     }
     syncing = true;
 
     ExecutorService executorService = Executors.newFixedThreadPool(MULTI_NODE_PIPELINE_SYNC_WORKERS);
 
-    CountDownLatch countDownLatch = new CountDownLatch(pipelinedResponses.size());
     Iterator<Map.Entry<HostAndPort, Queue<Response<?>>>> pipelinedResponsesIterator
         = pipelinedResponses.entrySet().iterator();
+    List<Future<Void>> futures = new ArrayList<>(pipelinedResponses.size());
+
     while (pipelinedResponsesIterator.hasNext()) {
       Map.Entry<HostAndPort, Queue<Response<?>>> entry = pipelinedResponsesIterator.next();
       HostAndPort nodeKey = entry.getKey();
       Queue<Response<?>> queue = entry.getValue();
       Connection connection = connections.get(nodeKey);
-      executorService.submit(() -> {
+      Future<Void> future = executorService.submit(() -> {
         try {
           List<Object> unformatted = connection.getMany(queue.size());
           for (Object o : unformatted) {
@@ -119,19 +123,30 @@ public final void sync() {
           pipelinedResponsesIterator.remove();
           connections.remove(nodeKey);
           IOUtils.closeQuietly(connection);
-        } finally {
-          countDownLatch.countDown();
         }
+
+        return null;
       });
+
+      futures.add(future);
     }
 
     try {
-      countDownLatch.await();
+      for (Future<Void> future : futures) {
+        future.get();
+      }
     } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
       log.error("Thread is interrupted during sync.", e);
-    }
+    } catch (ExecutionException e) {
+      if (e.getCause() instanceof JedisException) {
+        throw (JedisException) e.getCause();
+      }
 
-    executorService.shutdownNow();
+      throw new JedisException(e.getCause());
+    } finally {
+      executorService.shutdownNow();
+    }
 
     syncing = false;
   }

src/test/java/redis/clients/jedis/ClusterPipeliningTest.java

@@ -1,6 +1,8 @@
 package redis.clients.jedis;
 
 import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
 import static redis.clients.jedis.Protocol.CLUSTER_HASHSLOTS;
 
 import java.util.*;
@@ -15,6 +17,7 @@
 
 import redis.clients.jedis.args.*;
 import redis.clients.jedis.exceptions.JedisDataException;
+import redis.clients.jedis.exceptions.JedisMovedDataException;
 import redis.clients.jedis.params.*;
 import redis.clients.jedis.providers.ClusterConnectionProvider;
 import redis.clients.jedis.resps.GeoRadiusResponse;
@@ -1082,6 +1085,20 @@ public void multiple() {
     }
   }
 
+  @Test
+  public void clusterPipelineHandlesMovedDataException() {
+    ClusterConnectionProvider provider = spy(new ClusterConnectionProvider(nodes, DEFAULT_CLIENT_CONFIG));
+    Connection connectionThatThrowsMovedData = mock(Connection.class);
+    doThrow(new JedisMovedDataException("test", new HostAndPort("", 0), 0))
+            .when(connectionThatThrowsMovedData)
+            .getMany(anyInt());
+    doReturn(connectionThatThrowsMovedData).when(provider).getConnection(any(HostAndPort.class));
+    ClusterPipeline p = new ClusterPipeline(provider);
+    p.set("foo", "bar");
+    assertThrows(JedisMovedDataException.class, p::sync);
+    verify(provider).renewSlotCache();
+  }
+
   private static void assertThreadsCount() {
     // Get the root thread group
     final ThreadGroup rootGroup = Thread.currentThread().getThreadGroup().getParent();