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 7c3bbe78e..2d2c88308 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 @@ -15,25 +15,26 @@ */ package org.springframework.data.redis.connection.jedis; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.LinkedHashMap; -import java.util.LinkedHashSet; -import java.util.List; -import java.util.Map; +import redis.clients.jedis.BinaryJedisPubSub; +import redis.clients.jedis.GeoCoordinate; +import redis.clients.jedis.GeoUnit; +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.ScanParams; +import redis.clients.jedis.ZParams; +import redis.clients.jedis.params.geo.GeoRadiusParam; + +import java.util.*; import java.util.Map.Entry; -import java.util.Properties; -import java.util.Random; -import java.util.Set; import java.util.concurrent.TimeUnit; 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.geo.Circle; @@ -44,27 +45,11 @@ import org.springframework.data.geo.Point; import org.springframework.data.redis.ClusterStateFailureException; import org.springframework.data.redis.ExceptionTranslationStrategy; import org.springframework.data.redis.PassThroughExceptionTranslationStrategy; -import org.springframework.data.redis.connection.ClusterCommandExecutor; +import org.springframework.data.redis.connection.*; import org.springframework.data.redis.connection.ClusterCommandExecutor.ClusterCommandCallback; import org.springframework.data.redis.connection.ClusterCommandExecutor.MultiKeyClusterCommandCallback; import org.springframework.data.redis.connection.ClusterCommandExecutor.NodeResult; -import org.springframework.data.redis.connection.ClusterInfo; -import org.springframework.data.redis.connection.ClusterNodeResourceProvider; -import org.springframework.data.redis.connection.ClusterSlotHashUtil; -import org.springframework.data.redis.connection.ClusterTopology; -import org.springframework.data.redis.connection.ClusterTopologyProvider; -import org.springframework.data.redis.connection.DataType; -import org.springframework.data.redis.connection.MessageListener; -import org.springframework.data.redis.connection.RedisClusterConnection; -import org.springframework.data.redis.connection.RedisClusterNode; import org.springframework.data.redis.connection.RedisClusterNode.SlotRange; -import org.springframework.data.redis.connection.RedisNode; -import org.springframework.data.redis.connection.RedisPipelineException; -import org.springframework.data.redis.connection.RedisSentinelConnection; -import org.springframework.data.redis.connection.RedisSubscribedConnectionException; -import org.springframework.data.redis.connection.ReturnType; -import org.springframework.data.redis.connection.SortParameters; -import org.springframework.data.redis.connection.Subscription; import org.springframework.data.redis.connection.convert.Converters; import org.springframework.data.redis.connection.util.ByteArraySet; import org.springframework.data.redis.core.Cursor; @@ -74,20 +59,11 @@ import org.springframework.data.redis.core.ScanOptions; import org.springframework.data.redis.core.types.Expiration; import org.springframework.data.redis.core.types.RedisClientInfo; import org.springframework.data.redis.util.ByteUtils; +import org.springframework.data.util.DirectFieldAccessFallbackBeanWrapper; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; -import redis.clients.jedis.BinaryJedisPubSub; -import redis.clients.jedis.GeoCoordinate; -import redis.clients.jedis.GeoUnit; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisCluster; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.ScanParams; -import redis.clients.jedis.ZParams; -import redis.clients.jedis.params.geo.GeoRadiusParam; - /** * {@link RedisClusterConnection} implementation on top of {@link JedisCluster}.
* Uses the native {@link JedisCluster} api where possible and falls back to direct node communication using @@ -128,8 +104,8 @@ public class JedisClusterConnection implements RedisClusterConnection { 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 { @@ -457,6 +433,7 @@ public class JedisClusterConnection implements RedisClusterConnection { throw convertJedisAccessException(ex); } } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisKeyCommands#ttl(byte[], java.util.concurrent.TimeUnit) @@ -4109,19 +4086,34 @@ public class JedisClusterConnection implements RedisClusterConnection { * 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; + } } /* @@ -4132,17 +4124,23 @@ public class JedisClusterConnection implements RedisClusterConnection { @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())) { @@ -4152,6 +4150,16 @@ public class JedisClusterConnection implements RedisClusterConnection { 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) @@ -4160,7 +4168,6 @@ public class JedisClusterConnection implements RedisClusterConnection { 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 5f46a60e3..11522e2ff 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 @@ -58,7 +58,7 @@ import redis.clients.util.Pool; /** * Connection factory creating Jedis based connections. - * + * * @author Costin Leau * @author Thomas Darimont * @author Christoph Strobl @@ -118,7 +118,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Constructs a new JedisConnectionFactory instance. Will override the other connection parameters passed * to the factory. - * + * * @param shardInfo shard information */ public JedisConnectionFactory(JedisShardInfo shardInfo) { @@ -127,7 +127,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Constructs a new JedisConnectionFactory instance using the given pool configuration. - * + * * @param poolConfig pool configuration */ public JedisConnectionFactory(JedisPoolConfig poolConfig) { @@ -137,7 +137,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Constructs a new {@link JedisConnectionFactory} instance using the given {@link JedisPoolConfig} applied to * {@link JedisSentinelPool}. - * + * * @param sentinelConfig * @since 1.4 */ @@ -148,7 +148,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Constructs a new {@link JedisConnectionFactory} instance using the given {@link JedisPoolConfig} applied to * {@link JedisSentinelPool}. - * + * * @param sentinelConfig * @param poolConfig pool configuration. Defaulted to new instance if {@literal null}. * @since 1.4 @@ -161,7 +161,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisClusterConfiguration} applied * to create a {@link JedisCluster}. - * + * * @param clusterConfig * @since 1.7 */ @@ -172,7 +172,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisClusterConfiguration} applied * to create a {@link JedisCluster}. - * + * * @param clusterConfig * @since 1.7 */ @@ -184,7 +184,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Returns a Jedis instance to be used as a Redis connection. The instance can be newly created or retrieved from a * pool. - * + * * @return Jedis instance ready for wrapping into a {@link RedisConnection}. */ protected Jedis fetchJedisConnector() { @@ -208,7 +208,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Post process a newly retrieved connection. Useful for decorating or executing initialization commands on a new * connection. This implementation simply returns the connection. - * + * * @param connection * @return processed connection */ @@ -252,7 +252,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Creates {@link JedisSentinelPool}. - * + * * @param config * @return * @since 1.4 @@ -265,7 +265,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Creates {@link JedisPool}. - * + * * @return * @since 1.4 */ @@ -277,16 +277,16 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, private JedisCluster createCluster() { - JedisCluster cluster = createCluster(this.clusterConfig, this.poolConfig); - this.clusterCommandExecutor = new ClusterCommandExecutor( - new JedisClusterConnection.JedisClusterTopologyProvider(cluster), - new JedisClusterConnection.JedisClusterNodeResourceProvider(cluster), EXCEPTION_TRANSLATION); + JedisCluster cluster = createCluster(this.clusterConfig, getPoolConfig()); + JedisClusterConnection.JedisClusterTopologyProvider topologyProvider = new JedisClusterConnection.JedisClusterTopologyProvider(cluster); + this.clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, + new JedisClusterConnection.JedisClusterNodeResourceProvider(cluster, topologyProvider), EXCEPTION_TRANSLATION); return cluster; } /** * Creates {@link JedisCluster} for given {@link RedisClusterConfiguration} and {@link GenericObjectPoolConfig}. - * + * * @param clusterConfig must not be {@literal null}. * @param poolConfig can be {@literal null}. * @return @@ -375,7 +375,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Returns the Redis hostName. - * + * * @return the hostName. */ public String getHostName() { @@ -384,7 +384,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Sets the Redis hostName. - * + * * @param hostName the hostName to set. */ public void setHostName(String hostName) { @@ -413,7 +413,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Returns the password used for authenticating with the Redis server. - * + * * @return password for authentication. */ public String getPassword() { @@ -422,7 +422,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Sets the password used for authenticating with the Redis server. - * + * * @param password the password to set. */ public void setPassword(String password) { @@ -431,7 +431,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Returns the port used to connect to the Redis instance. - * + * * @return the Redis port. */ public int getPort() { @@ -440,7 +440,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Sets the port used to connect to the Redis instance. - * + * * @param port the Redis port. */ public void setPort(int port) { @@ -449,7 +449,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Returns the shardInfo. - * + * * @return the shardInfo. */ public JedisShardInfo getShardInfo() { @@ -458,7 +458,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Sets the shard info for this factory. - * + * * @param shardInfo the shardInfo to set. */ public void setShardInfo(JedisShardInfo shardInfo) { @@ -467,7 +467,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Returns the timeout. - * + * * @return the timeout. */ public int getTimeout() { @@ -485,7 +485,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Indicates the use of a connection pool. - * + * * @return the use of connection pooling. */ public boolean getUsePool() { @@ -494,7 +494,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Turns on or off the use of connection pooling. - * + * * @param usePool the usePool to set. */ public void setUsePool(boolean usePool) { @@ -503,7 +503,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Returns the poolConfig. - * + * * @return the poolConfig */ public JedisPoolConfig getPoolConfig() { @@ -512,7 +512,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Sets the pool configuration for this factory. - * + * * @param poolConfig the poolConfig to set. */ public void setPoolConfig(JedisPoolConfig poolConfig) { @@ -521,7 +521,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Returns the index of the database. - * + * * @return the database index. */ public int getDatabase() { @@ -530,7 +530,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, /** * Sets the index of the database used by this connection factory. Default is 0. - * + * * @param index database index. */ public void setDatabase(int index) { @@ -562,7 +562,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, * Specifies if pipelined results should be converted to the expected data type. If false, results of * {@link JedisConnection#closePipeline()} and {@link JedisConnection#exec()} will be of the type returned by the * Jedis driver. - * + * * @return Whether or not to convert pipeline and tx results. */ public boolean getConvertPipelineAndTxResults() { @@ -573,7 +573,7 @@ public class JedisConnectionFactory implements InitializingBean, DisposableBean, * Specifies if pipelined results should be converted to the expected data type. If false, results of * {@link JedisConnection#closePipeline()} and {@link JedisConnection#exec()} will be of the type returned by the * Jedis driver. - * + * * @param convertPipelineAndTxResults Whether or not to convert pipeline and tx results. */ public void setConvertPipelineAndTxResults(boolean convertPipelineAndTxResults) { 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 9a0e8f8bf..0686e93cd 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,15 +15,24 @@ */ 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.Matchers.*; +import static org.mockito.Matchers.anyVararg; +import static org.mockito.Matchers.eq; 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; @@ -34,18 +43,16 @@ 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.runners.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; import org.springframework.data.redis.connection.RedisClusterNode; import org.springframework.data.redis.connection.jedis.JedisClusterConnection.JedisClusterTopologyProvider; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisCluster; -import redis.clients.jedis.JedisPool; -import redis.clients.jedis.exceptions.JedisConnectionException; - /** * @author Christoph Strobl * @author Mark Paluch @@ -54,26 +61,21 @@ import redis.clients.jedis.exceptions.JedisConnectionException; public class JedisClusterConnectionUnitTests { private static final String CLUSTER_NODES_RESPONSE = "" // - + MASTER_NODE_1_ID + " " + CLUSTER_HOST + ":" + MASTER_NODE_1_PORT - + " myself,master - 0 0 1 connected 0-5460" - + "\n" + MASTER_NODE_2_ID + " " + CLUSTER_HOST + ":" - + MASTER_NODE_2_PORT - + " master - 0 1427718161587 2 connected 5461-10922" + "\n" - + MASTER_NODE_2_ID - + " " + CLUSTER_HOST + ":" + MASTER_NODE_3_PORT + " master - 0 1427718161587 3 connected 10923-16383"; + + MASTER_NODE_1_ID + " " + CLUSTER_HOST + ":" + MASTER_NODE_1_PORT + " myself,master - 0 0 1 connected 0-5460" + + "\n" + MASTER_NODE_2_ID + " " + CLUSTER_HOST + ":" + MASTER_NODE_2_PORT + + " master - 0 1427718161587 2 connected 5461-10922" + "\n" + MASTER_NODE_2_ID + " " + CLUSTER_HOST + ":" + + MASTER_NODE_3_PORT + " master - 0 1427718161587 3 connected 10923-16383"; - static final String CLUSTER_INFO_RESPONSE = "cluster_state:ok" + "\n" - + "cluster_slots_assigned:16384" + "\n" + "cluster_slots_ok:16384" - + "\n" + "cluster_slots_pfail:0" + "\n" - + "cluster_slots_fail:0" + "\n" + "cluster_known_nodes:4" - + "\n" + "cluster_size:3" + "\n" - + "cluster_current_epoch:30" + "\n" + "cluster_my_epoch:2" - + "\n" + "cluster_stats_messages_sent:2560260" - + "\n" + "cluster_stats_messages_received:2560086"; + static final String CLUSTER_INFO_RESPONSE = "cluster_state:ok" + "\n" + "cluster_slots_assigned:16384" + "\n" + + "cluster_slots_ok:16384" + "\n" + "cluster_slots_pfail:0" + "\n" + "cluster_slots_fail:0" + "\n" + + "cluster_known_nodes:4" + "\n" + "cluster_size:3" + "\n" + "cluster_current_epoch:30" + "\n" + + "cluster_my_epoch:2" + "\n" + "cluster_stats_messages_sent:2560260" + "\n" + + "cluster_stats_messages_received:2560086"; JedisClusterConnection connection; - @Mock JedisCluster clusterMock; + @Spy StubJedisCluster clusterMock; + @Mock JedisClusterConnectionHandler connectionHandlerMock; @Mock JedisPool node1PoolMock; @Mock JedisPool node2PoolMock; @@ -83,12 +85,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); @@ -99,6 +102,7 @@ public class JedisClusterConnectionUnitTests { when(node3PoolMock.getResource()).thenReturn(con3Mock); when(con1Mock.clusterNodes()).thenReturn(CLUSTER_NODES_RESPONSE); + clusterMock.setConnectionHandler(connectionHandlerMock); connection = new JedisClusterConnection(clusterMock); } @@ -252,6 +256,53 @@ public class JedisClusterConnectionUnitTests { verifyInvocationsAcross("time", times(1), con1Mock, con2Mock, con3Mock); } + @Test // DATAREDIS-679 + public void shouldFailWithUnknownNode() { + + try { + connection.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.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.dbSize(new RedisClusterNode(CLUSTER_HOST, MASTER_NODE_3_PORT)); + + assertThat(result, is(42L)); + } + @Test // DATAREDIS-315 public void timeShouldBeExecutedOnSingleNode() { @@ -298,4 +349,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; + } + } }