Retry with backoff on cluster connection failures

pom.xml

@@ -91,6 +91,12 @@
 			<version>2.3.2</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-core</artifactId>
+			<version>3.7.7</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<distributionManagement>
@@ -129,8 +135,8 @@
 				<artifactId>maven-compiler-plugin</artifactId>
 				<version>3.8.1</version>
 				<configuration>
-					<source>1.7</source>
-					<target>1.7</target>
+					<source>1.8</source>
+					<target>1.8</target>
 				</configuration>
 			</plugin>
 			<plugin>

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

@@ -189,6 +189,11 @@ public BinaryJedis(final JedisSocketFactory jedisSocketFactory) {
     client = new Client(jedisSocketFactory);
   }
 
+  @Override
+  public String toString() {
+    return "BinaryJedis{" + client + '}';
+  }
+
   private void initializeClientFromURI(URI uri) {
     initializeClientFromURI(uri, null, null, null);
   }

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

@@ -57,6 +57,11 @@ public Connection(final JedisSocketFactory jedisSocketFactory) {
     this.jedisSocketFactory = jedisSocketFactory;
   }
 
+  @Override
+  public String toString() {
+    return "Connection{" + jedisSocketFactory + "}";
+  }
+
   public Socket getSocket() {
     return socket;
   }

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

@@ -121,4 +121,9 @@ public int getSoTimeout() {
   public void setSoTimeout(int soTimeout) {
     this.soTimeout = soTimeout;
   }
+
+  @Override
+  public String toString() {
+    return "DefaultJedisSocketFactory{" + host + ":" + +port + "}";
+  }
 }

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

@@ -1,5 +1,11 @@
 package redis.clients.jedis;
 
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import redis.clients.jedis.exceptions.JedisAskDataException;
 import redis.clients.jedis.exceptions.JedisClusterMaxAttemptsException;
 import redis.clients.jedis.exceptions.JedisClusterOperationException;
@@ -11,18 +17,33 @@
 
 public abstract class JedisClusterCommand<T> {
 
+  private static final Logger LOG = LoggerFactory.getLogger(JedisClusterCommand.class);
+
+  private static final Duration DEFAULT_MAX_TOTAL_RETRIES_DURATION = Duration.ofMillis(
+      BinaryJedisCluster.DEFAULT_TIMEOUT * BinaryJedisCluster.DEFAULT_MAX_ATTEMPTS);
+
   private final JedisClusterConnectionHandler connectionHandler;
   private final int maxAttempts;
+  private final Duration maxTotalRetriesDuration;
 
   public JedisClusterCommand(JedisClusterConnectionHandler connectionHandler, int maxAttempts) {
+    this(connectionHandler, maxAttempts, DEFAULT_MAX_TOTAL_RETRIES_DURATION);
+  }
+
+  /**
+   * @param maxTotalRetriesDuration No more attempts after we have been trying for this long.
+   */
+  public JedisClusterCommand(JedisClusterConnectionHandler connectionHandler, int maxAttempts,
+      Duration maxTotalRetriesDuration) {
     this.connectionHandler = connectionHandler;
     this.maxAttempts = maxAttempts;
+    this.maxTotalRetriesDuration = maxTotalRetriesDuration;
   }
 
   public abstract T execute(Jedis connection);
 
   public T run(String key) {
-    return runWithRetries(JedisClusterCRC16.getSlot(key), this.maxAttempts, false, null);
+    return runWithRetries(JedisClusterCRC16.getSlot(key));
   }
 
   public T run(int keyCount, String... keys) {
@@ -42,11 +63,11 @@ public T run(int keyCount, String... keys) {
       }
     }
 
-    return runWithRetries(slot, this.maxAttempts, false, null);
+    return runWithRetries(slot);
   }
 
   public T runBinary(byte[] key) {
-    return runWithRetries(JedisClusterCRC16.getSlot(key), this.maxAttempts, false, null);
+    return runWithRetries(JedisClusterCRC16.getSlot(key));
   }
 
   public T runBinary(int keyCount, byte[]... keys) {
@@ -66,7 +87,7 @@ public T runBinary(int keyCount, byte[]... keys) {
       }
     }
 
-    return runWithRetries(slot, this.maxAttempts, false, null);
+    return runWithRetries(slot);
   }
 
   public T runWithAnyNode() {
@@ -79,62 +100,131 @@ public T runWithAnyNode() {
     }
   }
 
-  private T runWithRetries(final int slot, int attempts, boolean tryRandomNode, JedisRedirectionException redirect) {
-    if (attempts <= 0) {
-      throw new JedisClusterMaxAttemptsException("No more cluster attempts left.");
+  private boolean shouldBackOff(int attemptsLeft) {
+    int attemptsDone = maxAttempts - attemptsLeft;
+    return attemptsDone >= maxAttempts / 3;
+  }
+
+  private static long getBackoffSleepMillis(int attemptsLeft, Instant deadline) {
+    if (attemptsLeft <= 0) {
+      return 0;
     }
 
-    Jedis connection = null;
-    try {
+    long millisLeft = Duration.between(Instant.now(), deadline).toMillis();
+    if (millisLeft < 0) {
+      throw new JedisClusterMaxAttemptsException("Deadline exceeded");
+    }
 
-      if (redirect != null) {
-        connection = this.connectionHandler.getConnectionFromNode(redirect.getTargetNode());
-        if (redirect instanceof JedisAskDataException) {
-          // TODO: Pipeline asking with the original command to make it faster....
-          connection.asking();
-        }
-      } else {
-        if (tryRandomNode) {
-          connection = connectionHandler.getConnection();
+    return millisLeft / (attemptsLeft * attemptsLeft);
+  }
+
+  private T runWithRetries(final int slot) {
+    Instant deadline = Instant.now().plus(maxTotalRetriesDuration);
+    Supplier<Jedis> connectionSupplier = () -> connectionHandler.getConnectionFromSlot(slot);
+
+    // If we got one redirection, stick with that and don't try anything else
+    Supplier<Jedis> redirectionSupplier = null;
+
+    for (int currentAttempt = 0; currentAttempt < this.maxAttempts; currentAttempt++) {
+      Jedis connection = null;
+      try {
+        if (redirectionSupplier != null) {
+          connection = redirectionSupplier.get();
         } else {
-          connection = connectionHandler.getConnectionFromSlot(slot);
+          connection = connectionSupplier.get();
         }
-      }
 
-      return execute(connection);
+        if (shouldBackOff(maxAttempts - currentAttempt)) {
+          // Don't just stick to this any more, start asking around
+          redirectionSupplier = null;
+        }
 
-    } catch (JedisNoReachableClusterNodeException jnrcne) {
-      throw jnrcne;
-    } catch (JedisConnectionException jce) {
-      // release current connection before recursion
-      releaseConnection(connection);
-      connection = null;
-
-      if (attempts <= 1) {
-        //We need this because if node is not reachable anymore - we need to finally initiate slots
-        //renewing, or we can stuck with cluster state without one node in opposite case.
-        //But now if maxAttempts = [1 or 2] we will do it too often.
-        //TODO make tracking of successful/unsuccessful operations for node - do renewing only
-        //if there were no successful responses from this node last few seconds
-        this.connectionHandler.renewSlotCache();
+        final T result = execute(connection);
+        if (currentAttempt > 0) {
+          LOG.info("Success after {} attempts", currentAttempt + 1);
+        }
+        return result;
+      } catch (JedisNoReachableClusterNodeException e) {
+        throw e;
+      } catch (JedisConnectionException e) {
+        LOG.warn("Failed connecting to Redis: {}", connection, e);
+        // "- 1" because we just did one, but the currentAttempt counter hasn't increased yet
+        int attemptsLeft = maxAttempts - currentAttempt - 1;
+        connectionSupplier = handleConnectionProblem(connection, slot, attemptsLeft, deadline);
+      } catch (JedisRedirectionException e) {
+        redirectionSupplier = handleRedirection(connection, e);
+      } finally {
+        releaseConnection(connection);
       }
 
-      return runWithRetries(slot, attempts - 1, tryRandomNode, redirect);
-    } catch (JedisRedirectionException jre) {
-      // if MOVED redirection occurred,
-      if (jre instanceof JedisMovedDataException) {
-        // it rebuilds cluster's slot cache recommended by Redis cluster specification
-        this.connectionHandler.renewSlotCache(connection);
-      }
+      LOG.info("{} retries left...", maxAttempts - currentAttempt - 1);
+    }
 
-      // release current connection before recursion
-      releaseConnection(connection);
-      connection = null;
+    throw new JedisClusterMaxAttemptsException("No more cluster attempts left.");
+  }
 
-      return runWithRetries(slot, attempts - 1, false, jre);
-    } finally {
-      releaseConnection(connection);
+  protected void sleep(long sleepMillis) {
+    try {
+      LOG.info("Backing off, sleeping {}ms before trying again...", sleepMillis);
+      TimeUnit.MILLISECONDS.sleep(sleepMillis);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private Supplier<Jedis> handleConnectionProblem(Jedis failedConnection, final int slot, int attemptsLeft,
+      Instant doneDeadline) {
+    if (!shouldBackOff(attemptsLeft)) {
+      return () -> {
+        Jedis connection = connectionHandler.getConnectionFromSlot(slot);
+        LOG.info("Retrying with {}", connection);
+        return connection;
+      };
     }
+
+    // Must release current connection before renewing the slot cache below. If we fail to do this,
+    // then JedisClusterTest.testReturnConnectionOnJedisClusterConnection will start failing
+    // intermittently.
+    releaseConnection(failedConnection);
+
+    //We need this because if node is not reachable anymore - we need to finally initiate slots
+    //renewing, or we can stuck with cluster state without one node in opposite case.
+    //TODO make tracking of successful/unsuccessful operations for node - do renewing only
+    //if there were no successful responses from this node last few seconds
+    this.connectionHandler.renewSlotCache();
+
+    return () -> {
+      sleep(getBackoffSleepMillis(attemptsLeft, doneDeadline));
+      // Get a random connection, it will redirect us if it's not the right one
+      LOG.info("Retrying with a random node...");
+      Jedis connection = connectionHandler.getConnection();
+      LOG.info("Retrying with random pick: {}", connection);
+      return connection;
+    };
+  }
+
+  private Supplier<Jedis> handleRedirection(Jedis connection, final JedisRedirectionException jre) {
+    LOG.debug("Redirected by server to {}", jre.getTargetNode());
+
+    // if MOVED redirection occurred,
+    if (jre instanceof JedisMovedDataException) {
+      // it rebuilds cluster's slot cache recommended by Redis cluster specification
+      this.connectionHandler.renewSlotCache(connection);
+    }
+
+    // release current connection before iteration
+    releaseConnection(connection);
+
+    return () -> {
+      Jedis redirectedConnection = connectionHandler.getConnectionFromNode(jre.getTargetNode());
+      LOG.info("Retrying with redirection target {}", connection);
+      if (jre instanceof JedisAskDataException) {
+        // TODO: Pipeline asking with the original command to make it faster....
+        redirectedConnection.asking();
+      }
+
+      return redirectedConnection;
+    };
   }
 
   private void releaseConnection(Jedis connection) {

src/test/java/redis/clients/jedis/tests/JedisClusterTest.java

@@ -583,7 +583,11 @@ public void testReturnConnectionOnJedisConnectionException() throws InterruptedE
     jedisClusterNode.add(new HostAndPort("127.0.0.1", 7379));
     JedisPoolConfig config = new JedisPoolConfig();
     config.setMaxTotal(1);
-    JedisCluster jc = new JedisCluster(jedisClusterNode, DEFAULT_TIMEOUT, DEFAULT_TIMEOUT,
+
+    // Otherwise the test can time out before we are done
+    int shorterThanTheTestTimeoutMs = DEFAULT_TIMEOUT / 2;
+
+    JedisCluster jc = new JedisCluster(jedisClusterNode, shorterThanTheTestTimeoutMs, shorterThanTheTestTimeoutMs,
         DEFAULT_REDIRECTIONS, "cluster", config);
 
     Jedis j = jc.getClusterNodes().get("127.0.0.1:7380").getResource();

src/test/java/redis/clients/jedis/tests/SSLJedisClusterTest.java

@@ -114,7 +114,7 @@ public void connectToNodesFailsWithSSLParametersAndNoHostMapping() {
         null, sslParameters, null, portMap)){
       jc.get("foo");
       Assert.fail("The code did not throw the expected JedisClusterMaxAttemptsException.");
-    } catch (JedisClusterMaxAttemptsException e) {
+    } catch (JedisNoReachableClusterNodeException e) {
       // initial connection to localhost works, but subsequent connections to nodes use 127.0.0.1
       // and fail hostname verification
     }
@@ -159,7 +159,7 @@ public void connectWithCustomHostNameVerifier() {
         null, null, hostnameVerifier, portMap)){
       jc.get("foo");
       Assert.fail("The code did not throw the expected JedisClusterMaxAttemptsException.");
-    } catch (JedisClusterMaxAttemptsException e) {
+    } catch (JedisNoReachableClusterNodeException e) {
       // initial connection made with 'localhost' but subsequent connections to nodes use 127.0.0.1
       // which causes custom hostname verification to fail
     }

src/test/java/redis/clients/jedis/tests/SSLJedisClusterWithCompleteCredentialsTest.java

@@ -4,6 +4,7 @@
 import redis.clients.jedis.*;
 import redis.clients.jedis.exceptions.JedisClusterMaxAttemptsException;
 import redis.clients.jedis.exceptions.JedisConnectionException;
+import redis.clients.jedis.exceptions.JedisNoReachableClusterNodeException;
 import redis.clients.jedis.tests.SSLJedisTest.BasicHostnameVerifier;
 import redis.clients.jedis.tests.utils.RedisVersionUtil;
 
@@ -103,7 +104,7 @@ public void connectToNodesFailsWithSSLParametersAndNoHostMapping() {
         null, sslParameters, null, portMap)){
       jc.get("foo");
       Assert.fail("The code did not throw the expected JedisClusterMaxAttemptsException.");
-    } catch (JedisClusterMaxAttemptsException e) {
+    } catch (JedisNoReachableClusterNodeException e) {
       // initial connection to localhost works, but subsequent connections to nodes use 127.0.0.1
       // and fail hostname verification
     }
@@ -147,7 +148,7 @@ public void connectWithCustomHostNameVerifier() {
         null, null, hostnameVerifier, portMap)){
       jc.get("foo");
       Assert.fail("The code did not throw the expected JedisClusterMaxAttemptsException.");
-    } catch (JedisClusterMaxAttemptsException e) {
+    } catch (JedisNoReachableClusterNodeException e) {
       // initial connection made with 'localhost' but subsequent connections to nodes use 127.0.0.1
       // which causes custom hostname verification to fail
     }