DATAREDIS-794 - Improve Jedis Cluster Topology caching.

We now made JedisClusterTopologyProvider public and configurable through overriding JedisConnectionFactory.createTopologyProvider(…).

The topology provider accepts a configurable timeout to configure the cache timeout. Alternatively, custom ClusterTopologyProvider implementations can be returned through JedisConnectionFactory.createTopologyProvider(…).

Original Pull Request: #383

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java

@@ -24,6 +24,7 @@ import redis.clients.jedis.JedisCluster;
 import redis.clients.jedis.JedisClusterConnectionHandler;
 import redis.clients.jedis.JedisPool;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -83,7 +84,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 
 	private boolean closed;
 
-	private final JedisClusterTopologyProvider topologyProvider;
+	private final ClusterTopologyProvider topologyProvider;
 	private ClusterCommandExecutor clusterCommandExecutor;
 	private final boolean disposeClusterCommandExecutorOnClose;
 
@@ -116,19 +117,34 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 
 	/**
 	 * Create new {@link JedisClusterConnection} utilizing native connections via {@link JedisCluster} running commands
-	 * across the cluster via given {@link ClusterCommandExecutor}.
+	 * across the cluster via given {@link ClusterCommandExecutor}. Uses {@link JedisClusterTopologyProvider} by default.
 	 *
 	 * @param cluster must not be {@literal null}.
 	 * @param executor must not be {@literal null}.
 	 */
 	public JedisClusterConnection(JedisCluster cluster, ClusterCommandExecutor executor) {
+		this(cluster, executor, new JedisClusterTopologyProvider(cluster));
+	}
+
+	/**
+	 * Create new {@link JedisClusterConnection} utilizing native connections via {@link JedisCluster} running commands
+	 * across the cluster via given {@link ClusterCommandExecutor} and using the given {@link ClusterTopologyProvider}.
+	 *
+	 * @param cluster must not be {@literal null}.
+	 * @param executor must not be {@literal null}.
+	 * @param topologyProvider must not be {@literal null}.
+	 * @since 2.2
+	 */
+	public JedisClusterConnection(JedisCluster cluster, ClusterCommandExecutor executor,
+			ClusterTopologyProvider topologyProvider) {
 
 		Assert.notNull(cluster, "JedisCluster must not be null.");
 		Assert.notNull(executor, "ClusterCommandExecutor must not be null.");
+		Assert.notNull(topologyProvider, "ClusterTopologyProvider must not be null.");
 
 		this.closed = false;
 		this.cluster = cluster;
-		this.topologyProvider = new JedisClusterTopologyProvider(cluster);
+		this.topologyProvider = topologyProvider;
 		this.clusterCommandExecutor = executor;
 		this.disposeClusterCommandExecutorOnClose = false;
 	}
@@ -955,22 +971,41 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 	 * Jedis specific implementation of {@link ClusterTopologyProvider}.
 	 *
 	 * @author Christoph Strobl
+	 * @author Mark Paluch
 	 * @since 1.7
 	 */
-	static class JedisClusterTopologyProvider implements ClusterTopologyProvider {
+	public static class JedisClusterTopologyProvider implements ClusterTopologyProvider {
 
 		private final Object lock = new Object();
 		private final JedisCluster cluster;
+		private final long cacheTimeMs;
 		private long time = 0;
 		private @Nullable ClusterTopology cached;
 
 		/**
-		 * Create new {@link JedisClusterTopologyProvider}.s
+		 * Create new {@link JedisClusterTopologyProvider}. Uses a default cache timeout of 100 milliseconds.
 		 *
-		 * @param cluster
+		 * @param cluster must not be {@literal null}.
 		 */
 		public JedisClusterTopologyProvider(JedisCluster cluster) {
+			this(cluster, Duration.ofMillis(100));
+		}
+
+		/**
+		 * Create new {@link JedisClusterTopologyProvider}.
+		 *
+		 * @param cluster must not be {@literal null}.
+		 * @param cacheTimeout must not be {@literal null}.
+		 * @since 2.2
+		 */
+		public JedisClusterTopologyProvider(JedisCluster cluster, Duration cacheTimeout) {
+
+			Assert.notNull(cluster, "JedisCluster must not be null!");
+			Assert.notNull(cacheTimeout, "Cache timeout must not be null!");
+			Assert.isTrue(!cacheTimeout.isNegative(), "Cache timeout must not be negative.");
+
 			this.cluster = cluster;
+			this.cacheTimeMs = cacheTimeout.toMillis();
 		}
 
 		/*
@@ -980,7 +1015,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 		@Override
 		public ClusterTopology getTopology() {
 
-			if (cached != null && time + 100 > System.currentTimeMillis()) {
+			if (cached != null && shouldUseCachedValue()) {
 				return cached;
 			}
 
@@ -1014,6 +1049,19 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 			throw new ClusterStateFailureException(
 					"Could not retrieve cluster information. CLUSTER NODES returned with error." + sb.toString());
 		}
+
+		/**
+		 * Returns whether {@link #getTopology()} should return the cached {@link ClusterTopology}. Uses a time-based
+		 * caching.
+		 *
+		 * @return {@literal true} to use the cached {@link ClusterTopology}; {@literal false} to fetch a new cluster
+		 *         topology.
+		 * @see #JedisClusterTopologyProvider(JedisCluster, Duration)
+		 * @since 2.2
+		 */
+		protected boolean shouldUseCachedValue() {
+			return time + cacheTimeMs > System.currentTimeMillis();
+		}
 	}
 
 	protected JedisCluster getCluster() {
@@ -1024,7 +1072,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 		return clusterCommandExecutor;
 	}
 
-	protected JedisClusterTopologyProvider getTopologyProvider() {
+	protected ClusterTopologyProvider getTopologyProvider() {
 		return topologyProvider;
 	}
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java

@@ -96,6 +96,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 	private @Nullable RedisConfiguration configuration;
 
 	private @Nullable JedisCluster cluster;
+	private @Nullable ClusterTopologyProvider topologyProvider;
 	private @Nullable ClusterCommandExecutor clusterCommandExecutor;
 
 	/**
@@ -342,7 +343,12 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 		}
 
 		if (isRedisClusterAware()) {
+
 			this.cluster = createCluster();
+			this.topologyProvider = createTopologyProvider(this.cluster);
+			this.clusterCommandExecutor = new ClusterCommandExecutor(this.topologyProvider,
+					new JedisClusterConnection.JedisClusterNodeResourceProvider(this.cluster, this.topologyProvider),
+					EXCEPTION_TRANSLATION);
 		}
 	}
 
@@ -384,12 +390,20 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 	}
 
 	private JedisCluster createCluster() {
+		return createCluster((RedisClusterConfiguration) this.configuration, getPoolConfig());
+	}
 
-		JedisCluster cluster = createCluster((RedisClusterConfiguration) this.configuration, getPoolConfig());
-		JedisClusterTopologyProvider topologyProvider = new JedisClusterTopologyProvider(cluster);
-		this.clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider,
-				new JedisClusterConnection.JedisClusterNodeResourceProvider(cluster, topologyProvider), EXCEPTION_TRANSLATION);
-		return cluster;
+	/**
+	 * Template method to create a {@link ClusterTopologyProvider} given {@link JedisCluster}. Creates
+	 * {@link JedisClusterTopologyProvider} by default.
+	 *
+	 * @param cluster the {@link JedisCluster}, must not be {@literal null}.
+	 * @return the {@link ClusterTopologyProvider}.
+	 * @see JedisClusterTopologyProvider
+	 * @see 2.2
+	 */
+	protected ClusterTopologyProvider createTopologyProvider(JedisCluster cluster) {
+		return new JedisClusterTopologyProvider(cluster);
 	}
 
 	/**
@@ -479,7 +493,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean,
 		if (!isRedisClusterAware()) {
 			throw new InvalidDataAccessApiUsageException("Cluster is not configured!");
 		}
-		return new JedisClusterConnection(cluster, clusterCommandExecutor);
+		return new JedisClusterConnection(this.cluster, this.clusterCommandExecutor, this.topologyProvider);
 	}
 
 	/*