DATAREDIS-988 - Consider transaction participation when releasing connections.

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-redis</artifactId>
-	<version>2.1.9.BUILD-SNAPSHOT</version>
+	<version>2.1.9.DATAREDIS-988-SNAPSHOT</version>
 
 	<name>Spring Data Redis</name>
 

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java

@@ -424,7 +424,7 @@ public Object execute(String command, @Nullable CommandOutput commandOutputTypeH
 		}
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.AbstractRedisConnection#close()
 	 */
@@ -457,7 +457,7 @@ public void close() throws DataAccessException {
 		this.dbIndex = defaultDbIndex;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisConnection#isClosed()
 	 */
@@ -466,7 +466,7 @@ public boolean isClosed() {
 		return isClosed && !isSubscribed();
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisConnection#getNativeConnection()
 	 */
@@ -477,7 +477,7 @@ public RedisClusterAsyncCommands<byte[], byte[]> getNativeConnection() {
 		return (subscription != null ? subscription.getNativeConnection().async() : getAsyncConnection());
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisConnection#isQueueing()
 	 */
@@ -486,7 +486,7 @@ public boolean isQueueing() {
 		return isMulti;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisConnection#isPipelined()
 	 */
@@ -495,7 +495,7 @@ public boolean isPipelined() {
 		return isPipelined;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisConnection#openPipeline()
 	 */
@@ -507,7 +507,7 @@ public void openPipeline() {
 		}
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisConnection#closePipeline()
 	 */
@@ -743,7 +743,7 @@ public void watch(byte[]... keys) {
 	// Pub/Sub functionality
 	//
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisPubSubCommands#publish(byte[], byte[])
 	 */
@@ -768,7 +768,7 @@ public Long publish(byte[] channel, byte[] message) {
 		}
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisPubSubCommands#getSubscription()
 	 */
@@ -777,7 +777,7 @@ public Subscription getSubscription() {
 		return subscription;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisPubSubCommands#isSubscribed()
 	 */
@@ -786,7 +786,7 @@ public boolean isSubscribed() {
 		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 void pSubscribe(MessageListener listener, byte[]... patterns) {
 		}
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisPubSubCommands#subscribe(org.springframework.data.redis.connection.MessageListener, byte[][])
 	 */
@@ -1254,6 +1254,13 @@ private class LettucePoolConnectionProvider implements LettuceConnectionProvider
 		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<byte[], byte[]>) connection);
 			} else {
 				pool.returnBrokenResource((StatefulConnection<byte[], byte[]>) connection);

src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java

@@ -17,6 +17,7 @@
 
 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 @@ public void release(StatefulConnection<?, ?> connection) {
 					+ " 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);
 	}
 

src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java

@@ -187,8 +187,24 @@ private static RedisConnection createConnectionProxy(RedisConnection connection,
 	 *
 	 * @param conn the Redis connection to close.
 	 * @param factory the Redis factory that the connection was created with.
+	 * @deprecated since 2.1.9, use {@link #releaseConnection(RedisConnection, RedisConnectionFactory, boolean)}
 	 */
+	@Deprecated
 	public static void releaseConnection(@Nullable RedisConnection conn, RedisConnectionFactory factory) {
+		releaseConnection(conn, factory, false);
+	}
+
+	/**
+	 * Closes the given connection, created via the given factory if not managed externally (i.e. not bound to the
+	 * thread).
+	 *
+	 * @param conn the Redis connection to close.
+	 * @param factory the Redis factory that the connection was created with.
+	 * @param enableTransactionSupport whether transaction support is enabled.
+	 * @since 2.1.9
+	 */
+	public static void releaseConnection(@Nullable RedisConnection conn, RedisConnectionFactory factory,
+			boolean enableTransactionSupport) {
 
 		if (conn == null) {
 			return;
@@ -203,11 +219,25 @@ public static void releaseConnection(@Nullable RedisConnection conn, RedisConnec
 			return;
 		}
 
-		// release transactional/read-only and non-transactional/non-bound connections.
-		// transactional connections for read-only transactions get no synchronizer registered
-		if (isConnectionTransactional(conn, factory) && TransactionSynchronizationManager.isCurrentTransactionReadOnly()) {
-			unbindConnection(factory);
-		} else if (!isConnectionTransactional(conn, factory)) {
+		if (isConnectionTransactional(conn, factory)) {
+
+			// release transactional/read-only and non-transactional/non-bound connections.
+			// transactional connections for read-only transactions get no synchronizer registered
+			if (enableTransactionSupport && TransactionSynchronizationManager.isCurrentTransactionReadOnly()) {
+				if (log.isDebugEnabled()) {
+					log.debug("Unbinding Redis Connection");
+				}
+				unbindConnection(factory);
+			} else {
+
+				// Not participating in transaction management.
+				// Connection could have been attached via session callback.
+				if (log.isDebugEnabled()) {
+					log.debug("Leaving bound Redis Connection attached");
+				}
+			}
+
+		} else {
 			if (log.isDebugEnabled()) {
 				log.debug("Closing Redis Connection");
 			}

src/main/java/org/springframework/data/redis/core/RedisTemplate.java

@@ -231,7 +231,7 @@ public <T> T execute(RedisCallback<T> action, boolean exposeConnection, boolean
 			// TODO: any other connection processing?
 			return postProcessResult(result, connToUse, existingConnection);
 		} finally {
-			RedisConnectionUtils.releaseConnection(conn, factory);
+			RedisConnectionUtils.releaseConnection(conn, factory, enableTransactionSupport);
 		}
 	}
 

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<byte[], byte[]> connectionMock;
+	@Mock RedisAsyncCommands<byte[], byte[]> 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();
+	}
+}

src/test/java/org/springframework/data/redis/core/RedisTemplateUnitTests.java

@@ -28,14 +28,20 @@
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
+
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
 import org.springframework.instrument.classloading.ShadowingClassLoader;
+import org.springframework.lang.Nullable;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
 
 /**
+ * Unit tests for {@link RedisTemplate}.
+ *
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 @RunWith(MockitoJUnitRunner.class)
 public class RedisTemplateUnitTests {
@@ -47,6 +53,8 @@ public class RedisTemplateUnitTests {
 	@Before
 	public void setUp() {
 
+		TransactionSynchronizationManager.clear();
+
 		template = new RedisTemplate<>();
 		template.setConnectionFactory(connectionFactoryMock);
 		when(connectionFactoryMock.getConnection()).thenReturn(redisConnectionMock);
@@ -96,6 +104,60 @@ public void executeWithStickyConnectionShouldNotCloseConnectionWhenDone() {
 		verify(redisConnectionMock, never()).close();
 	}
 
+	@Test // DATAREDIS-988
+	public void executeSessionShouldReuseConnection() {
+
+		template.execute(new SessionCallback<Object>() {
+			@Nullable
+			@Override
+			public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
+				template.multi();
+				template.multi();
+				return null;
+			}
+		});
+
+		verify(connectionFactoryMock).getConnection();
+		verify(redisConnectionMock).close();
+	}
+
+	@Test // DATAREDIS-988
+	public void executeSessionInTransactionShouldReuseConnection() {
+
+		TransactionSynchronizationManager.setCurrentTransactionReadOnly(true);
+
+		template.execute(new SessionCallback<Object>() {
+			@Override
+			public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
+				template.multi();
+				template.multi();
+				return null;
+			}
+		});
+
+		verify(connectionFactoryMock).getConnection();
+		verify(redisConnectionMock).close();
+	}
+
+	@Test // DATAREDIS-988
+	public void transactionAwareTemplateShouldReleaseConnection() {
+
+		template.setEnableTransactionSupport(true);
+		TransactionSynchronizationManager.setCurrentTransactionReadOnly(true);
+
+		template.execute(new SessionCallback<Object>() {
+			@Override
+			public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
+				template.multi();
+				template.multi();
+				return null;
+			}
+		});
+
+		verify(connectionFactoryMock, times(2)).getConnection();
+		verify(redisConnectionMock, times(2)).close();
+	}
+
 	static class SomeArbitrarySerializableObject implements Serializable {
 		private static final long serialVersionUID = -5973659324040506423L;
 	}