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,6 +1,13 @@
 package redis.clients.jedis;
 
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import redis.clients.jedis.exceptions.JedisAskDataException;
+import redis.clients.jedis.exceptions.JedisClusterException;
 import redis.clients.jedis.exceptions.JedisClusterMaxAttemptsException;
 import redis.clients.jedis.exceptions.JedisClusterOperationException;
 import redis.clients.jedis.exceptions.JedisConnectionException;
@@ -11,18 +18,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 +64,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 +88,7 @@ public T runBinary(int keyCount, byte[]... keys) {
       }
     }
 
-    return runWithRetries(slot, this.maxAttempts, false, null);
+    return runWithRetries(slot);
   }
 
   public T runWithAnyNode() {
@@ -79,61 +101,114 @@ 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.");
-    }
-
-    Jedis connection = null;
-    try {
-
-      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();
+  private T runWithRetries(final int slot) {
+    Instant deadline = Instant.now().plus(maxTotalRetriesDuration);
+
+    JedisRedirectionException redirect = null;
+    int consecutiveConnectionFailures = 0;
+    for (int attemptsLeft = this.maxAttempts; attemptsLeft > 0; attemptsLeft--) {
+      Jedis connection = null;
+      try {
+        if (redirect != null) {
+          connection = connectionHandler.getConnectionFromNode(redirect.getTargetNode());
+          if (redirect instanceof JedisAskDataException) {
+            // TODO: Pipeline asking with the original command to make it faster....
+            connection.asking();
+          }
         } else {
           connection = connectionHandler.getConnectionFromSlot(slot);
         }
+
+        return execute(connection);
+
+      } catch (JedisNoReachableClusterNodeException jnrcne) {
+        throw jnrcne;
+      } catch (JedisConnectionException jce) {
+        ++consecutiveConnectionFailures;
+        LOG.debug("Failed connecting to Redis: {}", connection, jce);
+        // "- 1" because we just did one, but the attemptsLeft counter hasn't been decremented yet
+        boolean reset = handleConnectionProblem(attemptsLeft - 1, consecutiveConnectionFailures, deadline);
+        if (reset) {
+          consecutiveConnectionFailures = 0;
+          redirect = null;
+        }
+      } catch (JedisRedirectionException jre) {
+        LOG.debug("Redirected by server to {}", jre.getTargetNode());
+        consecutiveConnectionFailures = 0;
+        redirect = jre;
+        // if MOVED redirection occurred,
+        if (jre instanceof JedisMovedDataException) {
+          // it rebuilds cluster's slot cache recommended by Redis cluster specification
+          this.connectionHandler.renewSlotCache(connection);
+        }
+      } finally {
+        releaseConnection(connection);
+      }
+      if (Instant.now().isAfter(deadline)) {
+        // TODO: change to JedisClusterOperationException or a new sub-class of it
+        throw new JedisClusterMaxAttemptsException("Deadline exceeded");
       }
+    }
 
-      return execute(connection);
+    throw new JedisClusterMaxAttemptsException("No more cluster attempts left.");
+  }
 
-    } 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
+  /**
+   * Related values should be reset if <code>TRUE</code> is returned.
+   *
+   * @param attemptsLeft
+   * @param consecutiveConnectionFailures
+   * @param doneDeadline
+   * @return true - if some actions are taken
+   * <br /> false - if no actions are taken
+   */
+  private boolean handleConnectionProblem(int attemptsLeft, int consecutiveConnectionFailures, Instant doneDeadline) {
+    if (this.maxAttempts < 3) {
+      // Since we only renew the slots cache after two consecutive connection
+      // failures (see consecutiveConnectionFailures above), we need to special
+      // case the situation where we max out after two or fewer attempts.
+      //
+      // Otherwise, on two or fewer max attempts, the slots cache would never be
+      // renewed.
+      if (attemptsLeft == 0) {
         this.connectionHandler.renewSlotCache();
+        return true;
       }
+      return false;
+    }
 
-      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);
-      }
+    if (consecutiveConnectionFailures < 2) {
+      return false;
+    }
 
-      // release current connection before recursion
-      releaseConnection(connection);
-      connection = null;
+    sleep(getBackoffSleepMillis(attemptsLeft, doneDeadline));
+    //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 true;
+  }
 
-      return runWithRetries(slot, attempts - 1, false, jre);
-    } finally {
-      releaseConnection(connection);
+  private static long getBackoffSleepMillis(int attemptsLeft, Instant deadline) {
+    if (attemptsLeft <= 0) {
+      return 0;
+    }
+
+    long millisLeft = Duration.between(Instant.now(), deadline).toMillis();
+    if (millisLeft < 0) {
+      // TODO: change to JedisClusterOperationException or a new sub-class of it
+      throw new JedisClusterMaxAttemptsException("Deadline exceeded");
+    }
+
+    return millisLeft / (attemptsLeft * attemptsLeft);
+  }
+
+  protected void sleep(long sleepMillis) {
+    try {
+      TimeUnit.MILLISECONDS.sleep(sleepMillis);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
     }
   }
 

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();