From c8557505353bc82b892db4bf486f71e515408b41 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 31 Aug 2017 12:10:54 +0200 Subject: [PATCH] DATAREDIS-679 - Reconfigure Jedis cluster connection pools for discovered cluster nodes. We now request a cluster node connection directly from the driver's connection handler and initiate reconfiguration if a cluster node is known through the topology but has no connection pool yet. This can happen if a cluster node is added to the cluster after the connection was initialized. The connection handler cannot be obtained directly although the field where it's held is protected which increases the amount of necessary test code and reflection access to the connection handler. Jedis discovers nodes during startup, on demand (full scan) or when requested (e.g. by a cluster redirection). Original Pull Request: #270 --- .../jedis/JedisClusterConnection.java | 101 ++++++++------- .../jedis/JedisConnectionFactory.java | 66 +++++----- .../JedisClusterConnectionUnitTests.java | 116 ++++++++++++++---- 3 files changed, 179 insertions(+), 104 deletions(-) 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; + } + } }