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 d1414020e..6458b9590 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 @@ -17,8 +17,10 @@ package org.springframework.data.redis.connection.jedis; import redis.clients.jedis.BinaryJedis; import redis.clients.jedis.BinaryJedisPubSub; +import redis.clients.jedis.HostAndPort; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.JedisClusterConnectionHandler; import redis.clients.jedis.JedisPool; import java.util.Collection; @@ -32,6 +34,7 @@ import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.PropertyAccessor; import org.springframework.dao.DataAccessException; import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.data.redis.ClusterStateFailureException; @@ -43,6 +46,7 @@ import org.springframework.data.redis.connection.ClusterCommandExecutor.MultiKey import org.springframework.data.redis.connection.ClusterCommandExecutor.NodeResult; import org.springframework.data.redis.connection.RedisClusterNode.SlotRange; import org.springframework.data.redis.connection.convert.Converters; +import org.springframework.data.util.DirectFieldAccessFallbackBeanWrapper; import org.springframework.util.Assert; /** @@ -85,8 +89,8 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection { closed = false; topologyProvider = new JedisClusterTopologyProvider(cluster); - clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, new JedisClusterNodeResourceProvider(cluster), - EXCEPTION_TRANSLATION); + clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, + new JedisClusterNodeResourceProvider(cluster, topologyProvider), EXCEPTION_TRANSLATION); disposeClusterCommandExecutorOnClose = true; try { @@ -755,19 +759,35 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection { * Jedis specific implementation of {@link ClusterNodeResourceProvider}. * * @author Christoph Strobl + * @author Mark Paluch * @since 1.7 */ static class JedisClusterNodeResourceProvider implements ClusterNodeResourceProvider { private final JedisCluster cluster; + private final ClusterTopologyProvider topologyProvider; + private final JedisClusterConnectionHandler connectionHandler; /** * Creates new {@link JedisClusterNodeResourceProvider}. * * @param cluster must not be {@literal null}. + * @param topologyProvider must not be {@literal null}. */ - public JedisClusterNodeResourceProvider(JedisCluster cluster) { + JedisClusterNodeResourceProvider(JedisCluster cluster, ClusterTopologyProvider topologyProvider) { + this.cluster = cluster; + this.topologyProvider = topologyProvider; + + if (cluster != null) { + PropertyAccessor accessor = new DirectFieldAccessFallbackBeanWrapper(cluster); + + this.connectionHandler = accessor.isReadableProperty("connectionHandler") + ? (JedisClusterConnectionHandler) accessor.getPropertyValue("connectionHandler") + : null; + } else { + this.connectionHandler = null; + } } /* @@ -778,17 +798,23 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection { @SuppressWarnings("unchecked") public Jedis getResourceForSpecificNode(RedisClusterNode node) { + Assert.notNull(node, "Cannot get Pool for 'null' node!"); + JedisPool pool = getResourcePoolForSpecificNode(node); if (pool != null) { return pool.getResource(); } + Jedis connection = getConnectionForSpecificNode(node); + + if (connection != null) { + return connection; + } + throw new IllegalArgumentException(String.format("Node %s is unknown to cluster", node)); } - protected JedisPool getResourcePoolForSpecificNode(RedisNode node) { - - Assert.notNull(node, "Cannot get Pool for 'null' node!"); + private JedisPool getResourcePoolForSpecificNode(RedisClusterNode node) { Map clusterNodes = cluster.getClusterNodes(); if (clusterNodes.containsKey(node.asString())) { @@ -798,6 +824,16 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection { return null; } + private Jedis getConnectionForSpecificNode(RedisClusterNode node) { + + RedisClusterNode member = topologyProvider.getTopology().lookup(node); + if (connectionHandler != null && member != null) { + return connectionHandler.getConnectionFromNode(new HostAndPort(member.getHost(), member.getPort())); + } + + return null; + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ClusterNodeResourceProvider#returnResourceForSpecificNode(org.springframework.data.redis.connection.RedisClusterNode, java.lang.Object) @@ -806,7 +842,6 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection { public void returnResourceForSpecificNode(RedisClusterNode node, Object client) { getResourcePoolForSpecificNode(node).returnResource((Jedis) client); } - } /** 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 1a947eb64..e4c801ccf 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 @@ -59,6 +59,7 @@ import org.springframework.data.redis.connection.RedisPassword; import org.springframework.data.redis.connection.RedisSentinelConfiguration; import org.springframework.data.redis.connection.RedisSentinelConnection; import org.springframework.data.redis.connection.RedisStandaloneConfiguration; +import org.springframework.data.redis.connection.jedis.JedisClusterConnection.JedisClusterTopologyProvider; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; @@ -389,9 +390,9 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, private JedisCluster createCluster() { JedisCluster cluster = createCluster(this.clusterConfig, getPoolConfig()); - this.clusterCommandExecutor = new ClusterCommandExecutor( - new JedisClusterConnection.JedisClusterTopologyProvider(cluster), - new JedisClusterConnection.JedisClusterNodeResourceProvider(cluster), EXCEPTION_TRANSLATION); + JedisClusterTopologyProvider topologyProvider = new JedisClusterTopologyProvider(cluster); + this.clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, + new JedisClusterConnection.JedisClusterNodeResourceProvider(cluster, topologyProvider), EXCEPTION_TRANSLATION); return cluster; } 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 8183e40e0..62071c36c 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 @@ -15,19 +15,22 @@ */ package org.springframework.data.redis.connection.jedis; -import static org.hamcrest.core.Is.*; +import static org.hamcrest.Matchers.*; import static org.junit.Assert.*; import static org.mockito.Mockito.*; import static org.springframework.data.redis.connection.ClusterTestVariables.*; import static org.springframework.data.redis.test.util.MockitoUtils.*; +import redis.clients.jedis.HostAndPort; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.JedisClusterConnectionHandler; import redis.clients.jedis.JedisPool; import redis.clients.jedis.exceptions.JedisConnectionException; import java.io.IOException; import java.util.Arrays; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -38,7 +41,10 @@ import org.junit.Test; import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.invocation.InvocationOnMock; import org.mockito.junit.MockitoJUnitRunner; +import org.mockito.stubbing.Answer; import org.springframework.data.redis.ClusterStateFailureException; import org.springframework.data.redis.connection.ClusterInfo; import org.springframework.data.redis.connection.RedisClusterCommands.AddSlots; @@ -72,7 +78,8 @@ public class JedisClusterConnectionUnitTests { JedisClusterConnection connection; - @Mock JedisCluster clusterMock; + @Spy StubJedisCluster clusterMock; + @Mock JedisClusterConnectionHandler connectionHandlerMock; @Mock JedisPool node1PoolMock; @Mock JedisPool node2PoolMock; @@ -82,12 +89,13 @@ public class JedisClusterConnectionUnitTests { @Mock Jedis con2Mock; @Mock Jedis con3Mock; + Map nodes = new LinkedHashMap<>(); + public @Rule ExpectedException expectedException = ExpectedException.none(); @Before public void setUp() { - Map nodes = new LinkedHashMap<>(3); nodes.put(CLUSTER_HOST + ":" + MASTER_NODE_1_PORT, node1PoolMock); nodes.put(CLUSTER_HOST + ":" + MASTER_NODE_2_PORT, node2PoolMock); nodes.put(CLUSTER_HOST + ":" + MASTER_NODE_3_PORT, node3PoolMock); @@ -98,6 +106,7 @@ public class JedisClusterConnectionUnitTests { when(node3PoolMock.getResource()).thenReturn(con3Mock); when(con1Mock.clusterNodes()).thenReturn(CLUSTER_NODES_RESPONSE); + clusterMock.setConnectionHandler(connectionHandlerMock); connection = new JedisClusterConnection(clusterMock); } @@ -251,6 +260,53 @@ public class JedisClusterConnectionUnitTests { verifyInvocationsAcross("time", times(1), con1Mock, con2Mock, con3Mock); } + @Test // DATAREDIS-679 + public void shouldFailWithUnknownNode() { + + try { + connection.serverCommands().dbSize(new RedisClusterNode(CLUSTER_HOST, SLAVEOF_NODE_1_PORT)); + } catch (IllegalArgumentException e) { + assertThat(e.getMessage(), + containsString("Node " + CLUSTER_HOST + ":" + SLAVEOF_NODE_1_PORT + " is unknown to cluster")); + } + } + + @Test // DATAREDIS-679 + public void shouldFailWithAbsentConnection() { + + nodes.remove(CLUSTER_HOST + ":" + MASTER_NODE_3_PORT); + + try { + connection.serverCommands().dbSize(new RedisClusterNode(CLUSTER_HOST, MASTER_NODE_3_PORT)); + } catch (IllegalArgumentException e) { + assertThat(e.getMessage(), + containsString("Node " + CLUSTER_HOST + ":" + MASTER_NODE_3_PORT + " is unknown to cluster")); + } + } + + @Test // DATAREDIS-679 + public void shouldReconfigureJedisWithDiscoveredNode() { + + nodes.remove(CLUSTER_HOST + ":" + MASTER_NODE_3_PORT); + + when(connectionHandlerMock.getConnectionFromNode(new HostAndPort(CLUSTER_HOST, MASTER_NODE_3_PORT))) + .thenReturn(con3Mock); + when(con3Mock.dbSize()).thenAnswer(new Answer() { + @Override + public Long answer(InvocationOnMock invocation) throws Throwable { + + // Required to return the resource properly after invocation. + nodes.put(CLUSTER_HOST + ":" + MASTER_NODE_3_PORT, node3PoolMock); + + return 42L; + } + }); + + Long result = connection.serverCommands().dbSize(new RedisClusterNode(CLUSTER_HOST, MASTER_NODE_3_PORT)); + + assertThat(result, is(42L)); + } + @Test // DATAREDIS-315 public void timeShouldBeExecutedOnSingleNode() { @@ -297,4 +353,21 @@ public class JedisClusterConnectionUnitTests { new JedisClusterTopologyProvider(clusterMock).getTopology(); } + + static class StubJedisCluster extends JedisCluster { + + JedisClusterConnectionHandler connectionHandler; + + public StubJedisCluster() { + super(Collections.emptySet()); + } + + JedisClusterConnectionHandler getConnectionHandler() { + return connectionHandler; + } + + void setConnectionHandler(JedisClusterConnectionHandler connectionHandler) { + this.connectionHandler = connectionHandler; + } + } }