From bbd4482f267b4132f1bafb7d746f0a5bac7607ba 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 | 49 ++++++++++-- .../jedis/JedisConnectionFactory.java | 7 +- .../JedisClusterConnectionUnitTests.java | 79 ++++++++++++++++++- 3 files changed, 122 insertions(+), 13 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 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; + } + } }