diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java index 569496b0a..8b8497b3a 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java +++ b/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; } } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java index db7a0dbd5..ea83951d0 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java +++ b/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); } /* diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionUnitTests.java index efcd676f5..0c192a1df 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionUnitTests.java @@ -30,6 +30,7 @@ import redis.clients.jedis.JedisPool; import redis.clients.jedis.exceptions.JedisConnectionException; import java.io.IOException; +import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.LinkedHashMap; @@ -370,6 +371,32 @@ public class JedisClusterConnectionUnitTests { connection.set("foo".getBytes(), "bar".getBytes()); } + @Test // DATAREDIS-794 + public void clusterTopologyProviderShouldUseCachedTopology() { + + when(clusterMock.getClusterNodes()).thenReturn(Collections.singletonMap("mock", node1PoolMock)); + when(con1Mock.clusterNodes()).thenReturn(CLUSTER_NODES_RESPONSE); + + JedisClusterTopologyProvider provider = new JedisClusterTopologyProvider(clusterMock, Duration.ofSeconds(5)); + provider.getTopology(); + provider.getTopology(); + + verify(con1Mock).clusterNodes(); + } + + @Test // DATAREDIS-794 + public void clusterTopologyProviderShouldRequestTopology() { + + when(clusterMock.getClusterNodes()).thenReturn(Collections.singletonMap("mock", node1PoolMock)); + when(con1Mock.clusterNodes()).thenReturn(CLUSTER_NODES_RESPONSE); + + JedisClusterTopologyProvider provider = new JedisClusterTopologyProvider(clusterMock, Duration.ZERO); + provider.getTopology(); + provider.getTopology(); + + verify(con1Mock, times(2)).clusterNodes(); + } + static class StubJedisCluster extends JedisCluster { JedisClusterConnectionHandler connectionHandler; diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryUnitTests.java index 048dbb7d8..8252ffc28 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryUnitTests.java @@ -309,8 +309,9 @@ public class JedisConnectionFactoryUnitTests { private JedisConnectionFactory initSpyedConnectionFactory(RedisClusterConfiguration clusterConfig, JedisPoolConfig poolConfig) { + JedisCluster clusterMock = mock(JedisCluster.class); JedisConnectionFactory factorySpy = spy(new JedisConnectionFactory(clusterConfig)); - doReturn(null).when(factorySpy).createCluster(any(RedisClusterConfiguration.class), + doReturn(clusterMock).when(factorySpy).createCluster(any(RedisClusterConfiguration.class), any(GenericObjectPoolConfig.class)); doReturn(null).when(factorySpy).createRedisPool(); return factorySpy;