DATAREDIS-332 - @Transactional operation should use the same connection.

We only unbind the current redisConnectionFactory from the current thread if the transaction is not active anymore.
Previously we performed the unbinding in RedisTemplate.execute(RedisCallback, boolean, boolean) overtime which lead to the fact that the next redis operation within the same transaction used a different redis connection, which could lead to exhaustion of the connection pool.
We also ensure that we correctly unbind the RedisConnectionFactory resource from the TransactionSynchronizationManager on transaction completion in RedisTransactionSynchronizer.afterCompletion(int).
Added test case supplied from issue.

Original pull request: #134.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2015 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.
@@ -154,11 +154,12 @@ public abstract class RedisConnectionUtils {
 
 			if (!connHolder.isTransactionSyncronisationActive()) {
 				connHolder.setTransactionSyncronisationActive(true);
-				
+
 				RedisConnection conn = connHolder.getConnection();
 				conn.multi();
 
-				TransactionSynchronizationManager.registerSynchronization(new RedisTransactionSynchronizer(connHolder, conn));
+				TransactionSynchronizationManager.registerSynchronization(new RedisTransactionSynchronizer(connHolder, conn,
+						factory));
 			}
 		}
 	}
@@ -252,26 +253,31 @@ public abstract class RedisConnectionUtils {
 	}
 
 	/**
-	 * A {@link TransactionSynchronizationAdapter} that makes sure that the associated RedisConnection is released after the transaction completes.
+	 * A {@link TransactionSynchronizationAdapter} that makes sure that the associated RedisConnection is released after
+	 * the transaction completes.
 	 * 
 	 * @author Christoph Strobl
 	 * @author Thomas Darimont
 	 */
 	private static class RedisTransactionSynchronizer extends TransactionSynchronizationAdapter {
-		
+
 		private final RedisConnectionHolder connHolder;
 		private final RedisConnection connection;
+		private final RedisConnectionFactory factory;
 
 		/**
 		 * Creates a new {@link RedisTransactionSynchronizer}.
 		 * 
 		 * @param connHolder
 		 * @param connection
+		 * @param factory
 		 */
-		private RedisTransactionSynchronizer(RedisConnectionHolder connHolder, RedisConnection connection) {
-			
+		private RedisTransactionSynchronizer(RedisConnectionHolder connHolder, RedisConnection connection,
+				RedisConnectionFactory factory) {
+
 			this.connHolder = connHolder;
 			this.connection = connection;
+			this.factory = factory;
 		}
 
 		@Override
@@ -294,9 +300,10 @@ public abstract class RedisConnectionUtils {
 				if (log.isDebugEnabled()) {
 					log.debug("Closing bound connection after transaction completed with " + status);
 				}
-				
+
 				connHolder.setTransactionSyncronisationActive(false);
 				connection.close();
+				TransactionSynchronizationManager.unbindResource(factory);
 			}
 		}
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2015 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.
@@ -198,9 +198,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 			return postProcessResult(result, connToUse, existingConnection);
 		} finally {
 
-			if (enableTransactionSupport) {
-				RedisConnectionUtils.unbindConnection(factory);
-			} else {
+			if (!enableTransactionSupport) {
 				RedisConnectionUtils.releaseConnection(conn, factory);
 			}
 		}

src/test/java/org/springframework/data/redis/connection/jedis/JedisTransactionalConnectionStarvationTest.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2015 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
+ *
+ *      http://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.jedis;
+
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.SettingsUtils;
+import org.springframework.data.redis.connection.AbstractTransactionalTestBase;
+import org.springframework.data.redis.connection.jedis.JedisTransactionalConnectionStarvationTest.PooledJedisContextConfiguration;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+import org.springframework.test.annotation.Rollback;
+import org.springframework.test.context.ContextConfiguration;
+
+import redis.clients.jedis.JedisPoolConfig;
+
+/**
+ * @author Thomas Darimont
+ */
+@ContextConfiguration(classes = { PooledJedisContextConfiguration.class })
+public class JedisTransactionalConnectionStarvationTest extends AbstractTransactionalTestBase {
+
+	protected static final int MAX_CONNECTIONS = 5;
+
+	@Autowired StringRedisTemplate template;
+
+	protected void tryOperations(int numOperationsToTry) {
+
+		ValueOperations<String, String> ops = template.opsForValue();
+
+		for (int i = 0; i < numOperationsToTry; i++) {
+			ops.set("test-key-" + i, "test-value-" + i);
+		}
+	}
+
+	/**
+	 * @see DATAREDIS-332
+	 */
+	@Test
+	@Rollback
+	public void testNumberOfOperationsIsOne() {
+		tryOperations(1);
+	}
+
+	/**
+	 * @see DATAREDIS-332
+	 */
+	@Test
+	@Rollback
+	public void testNumberOfOperationsEqualToNumberOfConnections() {
+		tryOperations(MAX_CONNECTIONS);
+	}
+
+	/**
+	 * @see DATAREDIS-332
+	 */
+	@Test
+	@Rollback
+	public void testNumberOfOperationsGreaterThanNumberOfConnections() {
+		tryOperations(MAX_CONNECTIONS + 1);
+	}
+
+	@Configuration
+	public static class PooledJedisContextConfiguration extends RedisContextConfiguration {
+
+		@Bean
+		public JedisConnectionFactory redisConnectionFactory() {
+
+			JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
+			connectionFactory.setHostName(SettingsUtils.getHost());
+			connectionFactory.setPort(SettingsUtils.getPort());
+
+			JedisPoolConfig poolConfig = new JedisPoolConfig();
+			poolConfig.setMaxTotal(MAX_CONNECTIONS);
+			poolConfig.setMaxIdle(MAX_CONNECTIONS);
+			poolConfig.setMaxWaitMillis(2000L);
+			connectionFactory.setPoolConfig(poolConfig);
+			connectionFactory.afterPropertiesSet();
+
+			return connectionFactory;
+		}
+	}
+}