refresh connection for pipeline when getting JedisMovedDataException

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

@@ -72,6 +72,11 @@ protected Connection getConnection(HostAndPort nodeKey) {
     return provider.getConnection(nodeKey);
   }
 
+  @Override
+  protected void refreshConnection() {
+    provider.renewSlotCache();
+  }
+
   public Response<Long> spublish(String channel, String message) {
     return appendCommand(commandObjects.spublish(channel, message));
   }

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

@@ -14,6 +14,7 @@
 import org.slf4j.LoggerFactory;
 
 import redis.clients.jedis.exceptions.JedisConnectionException;
+import redis.clients.jedis.exceptions.JedisMovedDataException;
 import redis.clients.jedis.graph.GraphCommandObjects;
 import redis.clients.jedis.providers.ConnectionProvider;
 import redis.clients.jedis.util.IOUtils;
@@ -53,6 +54,8 @@ protected final void prepareGraphCommands(ConnectionProvider connectionProvider)
 
   protected abstract Connection getConnection(HostAndPort nodeKey);
 
+  protected abstract void refreshConnection();
+
   @Override
   protected final <T> Response<T> appendCommand(CommandObject<T> commandObject) {
     HostAndPort nodeKey = getNodeKey(commandObject.getArguments());
@@ -111,6 +114,9 @@ public final void sync() {
         try {
           List<Object> unformatted = connection.getMany(queue.size());
           for (Object o : unformatted) {
+            if (o instanceof JedisMovedDataException) {
+              refreshConnection();
+            }
             queue.poll().set(o);
           }
         } catch (JedisConnectionException jce) {

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

@@ -58,6 +58,11 @@ protected Connection getConnection(HostAndPort nodeKey) {
     return provider.getConnection(nodeKey);
   }
 
+  @Override
+  protected void refreshConnection() {
+    // do nothing
+  }
+
   /**
    * This method must be called after constructor, if graph commands are going to be used.
    */