Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-637 - Shut down connection-exclusive ClusterCommandExecutor.

Browse Source 
We now shut down ClusterCommandExecutor via LettuceClusterConnection and JedisClusterConnection if the ClusterCommandExecutor was created through the connection instance. In such case ClusterCommandExecutor is managed through Lettuce/JedisClusterConnection and disposed on connection close.

Previously ClusterCommandExecutor created through a ClusterConnection was not disposed.

Original pull request: #247.
This commit is contained in:
Mark Paluch
2017-04-28 14:31:23 +02:00
parent 4e2fb1b7a0
commit 1216134ce2
2 changed files with 55 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java
View File

@@ -29,6 +29,8 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
@@ -58,12 +60,15 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new PassThroughExceptionTranslationStrategy(
JedisConverters.exceptionConverter());
private final Log log = LogFactory.getLog(getClass());
private final JedisCluster cluster;
private boolean closed;
private final JedisClusterTopologyProvider topologyProvider;
private ClusterCommandExecutor clusterCommandExecutor;
private final boolean disposeClusterCommandExecutorOnClose;
private volatile JedisSubscription subscription;
@@ -82,6 +87,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
topologyProvider = new JedisClusterTopologyProvider(cluster);
clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider, new JedisClusterNodeResourceProvider(cluster),
EXCEPTION_TRANSLATION);
disposeClusterCommandExecutorOnClose = true;
try {
DirectFieldAccessor dfa = new DirectFieldAccessor(cluster);
@@ -108,6 +114,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
this.cluster = cluster;
this.topologyProvider = new JedisClusterTopologyProvider(cluster);
this.clusterCommandExecutor = executor;
this.disposeClusterCommandExecutorOnClose = false;
}
/*
@@ -366,9 +373,8 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
@Override
public String ping() {
return !clusterCommandExecutor
.executeCommandOnAllNodes((JedisClusterCommandCallback<String>) BinaryJedis::ping).resultsAsList()
.isEmpty() ? "PONG" : null;
return !clusterCommandExecutor.executeCommandOnAllNodes((JedisClusterCommandCallback<String>) BinaryJedis::ping)
.resultsAsList().isEmpty() ? "PONG" : null;
}
@@ -588,11 +594,10 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
final RedisClusterNode nodeToUse = topologyProvider.getTopology().lookup(master);
return JedisConverters.toSetOfRedisClusterNodes(
clusterCommandExecutor
.executeCommandOnSingleNode(
(JedisClusterCommandCallback<List<String>>) client -> client.clusterSlaves(nodeToUse.getId()), master)
.getValue());
return JedisConverters.toSetOfRedisClusterNodes(clusterCommandExecutor
.executeCommandOnSingleNode(
(JedisClusterCommandCallback<List<String>>) client -> client.clusterSlaves(nodeToUse.getId()), master)
.getValue());
}
/*
@@ -635,10 +640,8 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
@Override
public ClusterInfo clusterGetClusterInfo() {
return new ClusterInfo(JedisConverters
.toProperties(clusterCommandExecutor
.executeCommandOnArbitraryNode((JedisClusterCommandCallback<String>) Jedis::clusterInfo)
.getValue()));
return new ClusterInfo(JedisConverters.toProperties(clusterCommandExecutor
.executeCommandOnArbitraryNode((JedisClusterCommandCallback<String>) Jedis::clusterInfo).getValue()));
}
/*
@@ -655,6 +658,15 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
*/
@Override
public void close() throws DataAccessException {
if (!closed && disposeClusterCommandExecutorOnClose) {
try {
clusterCommandExecutor.destroy();
} catch (Exception ex) {
log.warn("Cannot properly close cluster command executor", ex);
}
}
closed = true;
}

40
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnection.java
View File

@@ -34,8 +34,11 @@ import java.util.List;
import java.util.Map;
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.factory.DisposableBean;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.ExceptionTranslationStrategy;
import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
@@ -59,9 +62,12 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
new LettuceExceptionConverter());
static final RedisCodec<byte[], byte[]> CODEC = ByteArrayCodec.INSTANCE;
private final Log log = LogFactory.getLog(getClass());
private final RedisClusterClient clusterClient;
private ClusterCommandExecutor clusterCommandExecutor;
private ClusterTopologyProvider topologyProvider;
private final boolean disposeClusterCommandExecutorOnClose;
/**
* Creates new {@link LettuceClusterConnection} using {@link RedisClusterClient}.
@@ -78,6 +84,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
topologyProvider = new LettuceClusterTopologyProvider(clusterClient);
clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider,
new LettuceClusterNodeResourceProvider(clusterClient), exceptionConverter);
disposeClusterCommandExecutorOnClose = true;
}
/**
@@ -97,6 +104,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
this.clusterClient = clusterClient;
topologyProvider = new LettuceClusterTopologyProvider(clusterClient);
clusterCommandExecutor = executor;
disposeClusterCommandExecutorOnClose = false;
}
/*
@@ -497,12 +505,10 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
@Override
public Map<RedisClusterNode, Collection<RedisClusterNode>> clusterGetMasterSlaveMap() {
List<NodeResult<Collection<RedisClusterNode>>> nodeResults = clusterCommandExecutor
.executeCommandAsyncOnNodes(
(LettuceClusterCommandCallback<Collection<RedisClusterNode>>) client -> Converters
.toSetOfRedisClusterNodes(client.clusterSlaves(client.clusterMyId())),
topologyProvider.getTopology().getActiveMasterNodes())
.getResults();
List<NodeResult<Collection<RedisClusterNode>>> nodeResults = clusterCommandExecutor.executeCommandAsyncOnNodes(
(LettuceClusterCommandCallback<Collection<RedisClusterNode>>) client -> Converters
.toSetOfRedisClusterNodes(client.clusterSlaves(client.clusterMyId())),
topologyProvider.getTopology().getActiveMasterNodes()).getResults();
Map<RedisClusterNode, Collection<RedisClusterNode>> result = new LinkedHashMap<>();
@@ -517,6 +523,24 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
return clusterCommandExecutor;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisConnection#close()
*/
@Override
public void close() throws DataAccessException {
if (!isClosed() && disposeClusterCommandExecutorOnClose) {
try {
clusterCommandExecutor.destroy();
} catch (Exception ex) {
log.warn("Cannot properly close cluster command executor", ex);
}
}
super.close();
}
/**
* Lettuce specific implementation of {@link ClusterCommandCallback}.
*
@@ -535,9 +559,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
* @since 1.7
*/
protected interface LettuceMultiKeyClusterCommandCallback<T>
extends MultiKeyClusterCommandCallback<RedisClusterCommands<byte[], byte[]>, T> {
}
extends MultiKeyClusterCommandCallback<RedisClusterCommands<byte[], byte[]>, T> {}
/**
* Lettuce specific implementation of {@link ClusterNodeResourceProvider}.