From 204d4b102baf008b264bf50d895114f4224a2f11 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Mon, 3 Jun 2019 10:51:58 +0200 Subject: [PATCH] DATAREDIS-988 - Discard transactions when releasing pooled connections. We now check on connection release whether a connection is in MULTI state. If so, then we discard (rollback) the transaction to reset the connection to a fresh state. Original Pull Request: #453 --- .../connection/lettuce/LettuceConnection.java | 31 ++++---- .../LettucePoolingConnectionProvider.java | 8 +++ ...ucePoolingConnectionProviderUnitTests.java | 70 +++++++++++++++++++ 3 files changed, 97 insertions(+), 12 deletions(-) create mode 100644 src/test/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProviderUnitTests.java diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 0f37e1186..de5f8d2de 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -424,7 +424,7 @@ public class LettuceConnection extends AbstractRedisConnection { } } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.AbstractRedisConnection#close() */ @@ -457,7 +457,7 @@ public class LettuceConnection extends AbstractRedisConnection { this.dbIndex = defaultDbIndex; } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisConnection#isClosed() */ @@ -466,7 +466,7 @@ public class LettuceConnection extends AbstractRedisConnection { return isClosed && !isSubscribed(); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisConnection#getNativeConnection() */ @@ -477,7 +477,7 @@ public class LettuceConnection extends AbstractRedisConnection { return (subscription != null ? subscription.getNativeConnection().async() : getAsyncConnection()); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisConnection#isQueueing() */ @@ -486,7 +486,7 @@ public class LettuceConnection extends AbstractRedisConnection { return isMulti; } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisConnection#isPipelined() */ @@ -495,7 +495,7 @@ public class LettuceConnection extends AbstractRedisConnection { return isPipelined; } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisConnection#openPipeline() */ @@ -507,7 +507,7 @@ public class LettuceConnection extends AbstractRedisConnection { } } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisConnection#closePipeline() */ @@ -743,7 +743,7 @@ public class LettuceConnection extends AbstractRedisConnection { // Pub/Sub functionality // - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisPubSubCommands#publish(byte[], byte[]) */ @@ -768,7 +768,7 @@ public class LettuceConnection extends AbstractRedisConnection { } } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisPubSubCommands#getSubscription() */ @@ -777,7 +777,7 @@ public class LettuceConnection extends AbstractRedisConnection { return subscription; } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisPubSubCommands#isSubscribed() */ @@ -786,7 +786,7 @@ public class LettuceConnection extends AbstractRedisConnection { return (subscription != null && subscription.isAlive()); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisPubSubCommands#pSubscribe(org.springframework.data.redis.connection.MessageListener, byte[][]) */ @@ -807,7 +807,7 @@ public class LettuceConnection extends AbstractRedisConnection { } } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisPubSubCommands#subscribe(org.springframework.data.redis.connection.MessageListener, byte[][]) */ @@ -1254,6 +1254,13 @@ public class LettuceConnection extends AbstractRedisConnection { public void release(StatefulConnection connection) { if (connection.isOpen()) { + + if (connection instanceof StatefulRedisConnection) { + StatefulRedisConnection redisConnection = (StatefulRedisConnection) connection; + if (redisConnection.isMulti()) { + redisConnection.async().discard(); + } + } pool.returnResource((StatefulConnection) connection); } else { pool.returnBrokenResource((StatefulConnection) connection); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java index b1043b45a..b5c049ba9 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java @@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.AbstractRedisClient; import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.support.ConnectionPoolSupport; import java.util.Map; @@ -117,6 +118,13 @@ class LettucePoolingConnectionProvider implements LettuceConnectionProvider, Red + " was either previously returned or does not belong to this connection provider"); } + if (connection instanceof StatefulRedisConnection) { + StatefulRedisConnection redisConnection = (StatefulRedisConnection) connection; + if (redisConnection.isMulti()) { + redisConnection.async().discard(); + } + } + pool.returnObject(connection); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProviderUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProviderUnitTests.java new file mode 100644 index 000000000..4b1ebbab2 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProviderUnitTests.java @@ -0,0 +1,70 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.lettuce; + +import static org.mockito.Mockito.*; + +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.api.async.RedisAsyncCommands; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +/** + * Unit tests for {@link LettucePoolingConnectionProvider}. + * + * @author Mark Paluch + */ +@RunWith(MockitoJUnitRunner.class) +public class LettucePoolingConnectionProviderUnitTests { + + @Mock LettuceConnectionProvider connectionProviderMock; + @Mock StatefulRedisConnection connectionMock; + @Mock RedisAsyncCommands commandsMock; + + LettucePoolingClientConfiguration config = LettucePoolingClientConfiguration.defaultConfiguration(); + + @Before + public void before() { + + when(connectionMock.async()).thenReturn(commandsMock); + when(connectionProviderMock.getConnection(any())).thenReturn(connectionMock); + } + + @Test // DATAREDIS-988 + public void shouldReturnConnectionOnRelease() { + + LettucePoolingConnectionProvider provider = new LettucePoolingConnectionProvider(connectionProviderMock, config); + + provider.release(provider.getConnection(StatefulRedisConnection.class)); + + verifyZeroInteractions(commandsMock); + } + + @Test // DATAREDIS-988 + public void shouldDiscardTransactionOnReleaseOnActiveTransaction() { + + LettucePoolingConnectionProvider provider = new LettucePoolingConnectionProvider(connectionProviderMock, config); + when(connectionMock.isMulti()).thenReturn(true); + + provider.release(provider.getConnection(StatefulRedisConnection.class)); + + verify(commandsMock).discard(); + } +}