DATAREDIS-531 - Scan commands should use a dedicated bound connection.

SCAN, SSCAN, HSCAN and ZSCAN now reference a dedicated connection that is bound to the Cursor. This allows creation of multiple cursors for different threads without the risk of potentially sharing a single connection. As before the caller is responsible for closing the Cursor correctly after usage.

Original pull request: #218.

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -3790,6 +3790,10 @@ public class JedisConnection extends AbstractRedisConnection {
 						JedisConverters.stringListToByteList().convert(result.getResult()));
 			}
 
+			protected void doClose() {
+				JedisConnection.this.close();
+			};
+
 		}.open();
 
 	}
@@ -3828,6 +3832,10 @@ public class JedisConnection extends AbstractRedisConnection {
 						JedisConverters.tuplesToTuples().convert(result.getResult()));
 			}
 
+			protected void doClose() {
+				JedisConnection.this.close();
+			};
+
 		}.open();
 	}
 
@@ -3863,6 +3871,10 @@ public class JedisConnection extends AbstractRedisConnection {
 				redis.clients.jedis.ScanResult<byte[]> result = jedis.sscan(key, JedisConverters.toBytes(cursorId), params);
 				return new ScanIteration<byte[]>(Long.valueOf(result.getStringCursor()), result.getResult());
 			}
+
+			protected void doClose() {
+				JedisConnection.this.close();
+			};
 		}.open();
 	}
 
@@ -3898,6 +3910,11 @@ public class JedisConnection extends AbstractRedisConnection {
 				ScanResult<Entry<byte[], byte[]>> result = jedis.hscan(key, JedisConverters.toBytes(cursorId), params);
 				return new ScanIteration<Map.Entry<byte[], byte[]>>(Long.valueOf(result.getStringCursor()), result.getResult());
 			}
+
+			protected void doClose() {
+				JedisConnection.this.close();
+			};
+
 		}.open();
 	}
 

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

@@ -3719,6 +3719,11 @@ public class LettuceConnection extends AbstractRedisConnection {
 
 				return new ScanIteration<byte[]>(Long.valueOf(nextCursorId), (keys));
 			}
+
+			protected void doClose() {
+				LettuceConnection.this.close();
+			}
+
 		}.open();
 
 	}
@@ -3759,6 +3764,10 @@ public class LettuceConnection extends AbstractRedisConnection {
 				return new ScanIteration<Entry<byte[], byte[]>>(Long.valueOf(nextCursorId), values.entrySet());
 			}
 
+			protected void doClose() {
+				LettuceConnection.this.close();
+			}
+
 		}.open();
 	}
 
@@ -3798,6 +3807,11 @@ public class LettuceConnection extends AbstractRedisConnection {
 				List<byte[]> values = failsafeReadScanValues(valueScanCursor.getValues(), null);
 				return new ScanIteration<byte[]>(Long.valueOf(nextCursorId), values);
 			}
+
+			protected void doClose() {
+				LettuceConnection.this.close();
+			}
+
 		}.open();
 	}
 
@@ -3839,6 +3853,11 @@ public class LettuceConnection extends AbstractRedisConnection {
 				List<Tuple> values = failsafeReadScanValues(result, LettuceConverters.scoredValuesToTupleList());
 				return new ScanIteration<Tuple>(Long.valueOf(nextCursorId), values);
 			}
+
+			protected void doClose() {
+				LettuceConnection.this.close();
+			}
+
 		}.open();
 	}
 

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

@@ -235,7 +235,7 @@ class DefaultHashOperations<K, HK, HV> extends AbstractOperations<K, Object> imp
 	public Cursor<Entry<HK, HV>> scan(K key, final ScanOptions options) {
 
 		final byte[] rawKey = rawKey(key);
-		return execute(new RedisCallback<Cursor<Map.Entry<HK, HV>>>() {
+		return template.executeWithStickyConnection(new RedisCallback<Cursor<Map.Entry<HK, HV>>>() {
 
 			@Override
 			public Cursor<Entry<HK, HV>> doInRedis(RedisConnection connection) throws DataAccessException {
@@ -268,7 +268,7 @@ class DefaultHashOperations<K, HK, HV> extends AbstractOperations<K, Object> imp
 						});
 			}
 
-		}, true);
+		});
 
 	}
 }

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

@@ -255,7 +255,7 @@ class DefaultSetOperations<K, V> extends AbstractOperations<K, V> implements Set
 	public Cursor<V> scan(K key, final ScanOptions options) {
 
 		final byte[] rawKey = rawKey(key);
-		return execute(new RedisCallback<Cursor<V>>() {
+		return template.executeWithStickyConnection(new RedisCallback<Cursor<V>>() {
 
 			@Override
 			public Cursor<V> doInRedis(RedisConnection connection) throws DataAccessException {
@@ -267,7 +267,7 @@ class DefaultSetOperations<K, V> extends AbstractOperations<K, V> implements Set
 					}
 				});
 			}
-		}, true);
+		});
 
 	}
 }

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

@@ -409,13 +409,13 @@ class DefaultZSetOperations<K, V> extends AbstractOperations<K, V> implements ZS
 	public Cursor<TypedTuple<V>> scan(K key, final ScanOptions options) {
 
 		final byte[] rawKey = rawKey(key);
-		Cursor<Tuple> cursor = execute(new RedisCallback<Cursor<Tuple>>() {
+		Cursor<Tuple> cursor = template.executeWithStickyConnection(new RedisCallback<Cursor<Tuple>>() {
 
 			@Override
 			public Cursor<Tuple> doInRedis(RedisConnection connection) throws DataAccessException {
 				return connection.zScan(rawKey, options);
 			}
-		}, true);
+		});
 
 		return new ConvertingCursor<Tuple, TypedTuple<V>>(cursor, new Converter<Tuple, TypedTuple<V>>() {
 

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

@@ -150,7 +150,8 @@ public interface HashOperations<H, HK, HV> {
 	RedisOperations<H, ?> getOperations();
 
 	/**
-	 * Use a {@link Cursor} to iterate over entries in hash at {@code key}.
+	 * Use a {@link Cursor} to iterate over entries in hash at {@code key}. <br />
+	 * <strong>Important:</strong> Call {@link Cursor#close()} when done to avoid resource leak.
 	 *
 	 * @since 1.4
 	 * @param key must not be {@literal null}.

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.redis.core;
 
+import java.io.Closeable;
 import java.util.Collection;
 import java.util.Date;
 import java.util.List;
@@ -22,6 +23,7 @@ import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.data.redis.connection.DataType;
+import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.core.query.SortQuery;
 import org.springframework.data.redis.core.script.RedisScript;
 import org.springframework.data.redis.core.types.RedisClientInfo;
@@ -129,6 +131,16 @@ public interface RedisOperations<K, V> {
 	<T> T execute(RedisScript<T> script, RedisSerializer<?> argsSerializer, RedisSerializer<T> resultSerializer,
 			List<K> keys, Object... args);
 
+	/**
+	 * Allocates and binds a new {@link RedisConnection} to the actual return type of the method. It is up to the caller
+	 * to free resources after use.
+	 *
+	 * @param callback must not be {@literal null}.
+	 * @return
+	 * @since 1.8
+	 */
+	<T extends Closeable> T executeWithStickyConnection(RedisCallback<T> callback);
+
 	Boolean hasKey(K key);
 
 	void delete(K key);

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.redis.core;
 
+import java.io.Closeable;
 import java.lang.reflect.Proxy;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -304,6 +305,23 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 		return scriptExecutor.execute(script, argsSerializer, resultSerializer, keys, args);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.RedisOperations#executeWithStickyConnection(org.springframework.data.redis.core.RedisCallback)
+	 */
+	public <T extends Closeable> T executeWithStickyConnection(RedisCallback<T> callback) {
+
+		Assert.isTrue(initialized, "template not initialized; call afterPropertiesSet() before using it");
+		Assert.notNull(callback, "Callback object must not be null");
+
+		RedisConnectionFactory factory = getConnectionFactory();
+
+		RedisConnection connection = preProcessConnection(RedisConnectionUtils.doGetConnection(factory, true, false, false),
+				false);
+
+		return callback.doInRedis(connection);
+	}
+
 	private Object executeSession(SessionCallback<?> session) {
 		return session.execute(this);
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2014 the original author or authors.
+ * Copyright 2011-2016 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.
@@ -75,8 +75,9 @@ public interface SetOperations<K, V> {
 	RedisOperations<K, V> getOperations();
 
 	/**
-	 * Iterate over elements in set at {@code key}.
-	 * 
+	 * Iterate over elements in set at {@code key}. <br />
+	 * <strong>Important:</strong> Call {@link Cursor#close()} when done to avoid resource leak.
+	 *
 	 * @since 1.4
 	 * @param key
 	 * @param options

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

@@ -136,6 +136,9 @@ public interface ZSetOperations<K, V> {
 	RedisOperations<K, V> getOperations();
 
 	/**
+	 * Iterate over elements in zset at {@code key}. <br />
+	 * <strong>Important:</strong> Call {@link Cursor#close()} when done to avoid resource leak.
+	 *
 	 * @since 1.4
 	 * @param key
 	 * @param options

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2016 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.
@@ -20,6 +20,10 @@ import static org.junit.Assert.*;
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map.Entry;
+
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -29,11 +33,16 @@ import org.mockito.ArgumentCaptor;
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.data.redis.connection.AbstractConnectionUnitTestBase;
 import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption;
+import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
 import org.springframework.data.redis.connection.jedis.JedisConnectionUnitTestSuite.JedisConnectionPipelineUnitTests;
 import org.springframework.data.redis.connection.jedis.JedisConnectionUnitTestSuite.JedisConnectionUnitTests;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
 
 import redis.clients.jedis.Client;
 import redis.clients.jedis.Jedis;
+import redis.clients.jedis.ScanParams;
+import redis.clients.jedis.ScanResult;
 
 /**
  * @author Christoph Strobl
@@ -205,6 +214,122 @@ public class JedisConnectionUnitTestSuite {
 			connection.zRangeByScore("foo".getBytes(), "foo", "bar", Integer.MAX_VALUE, new Long(Integer.MAX_VALUE) + 1L);
 		}
 
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test
+		public void scanShouldKeepTheConnectionOpen() {
+
+			doReturn(new ScanResult<String>("0", Collections.<String> emptyList())).when(jedisSpy).scan(anyString(),
+					any(ScanParams.class));
+
+			connection.scan();
+
+			verify(jedisSpy, never()).close();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test
+		public void scanShouldCloseTheConnectionWhenCursorIsClosed() throws IOException {
+
+			doReturn(new ScanResult<String>("0", Collections.<String> emptyList())).when(jedisSpy).scan(anyString(),
+					any(ScanParams.class));
+
+			Cursor<byte[]> cursor = connection.scan();
+			cursor.close();
+
+			verify(jedisSpy, times(1)).quit();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test
+		public void sScanShouldKeepTheConnectionOpen() {
+
+			doReturn(new ScanResult<String>("0", Collections.<String> emptyList())).when(jedisSpy).sscan(any(byte[].class),
+					any(byte[].class), any(ScanParams.class));
+
+			connection.sScan("foo".getBytes(), ScanOptions.NONE);
+
+			verify(jedisSpy, never()).close();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test
+		public void sScanShouldCloseTheConnectionWhenCursorIsClosed() throws IOException {
+
+			doReturn(new ScanResult<String>("0", Collections.<String> emptyList())).when(jedisSpy).sscan(any(byte[].class),
+					any(byte[].class), any(ScanParams.class));
+
+			Cursor<byte[]> cursor = connection.sScan("foo".getBytes(), ScanOptions.NONE);
+			cursor.close();
+
+			verify(jedisSpy, times(1)).quit();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test
+		public void zScanShouldKeepTheConnectionOpen() {
+
+			doReturn(new ScanResult<String>("0", Collections.<String> emptyList())).when(jedisSpy).zscan(any(byte[].class),
+					any(byte[].class), any(ScanParams.class));
+
+			connection.zScan("foo".getBytes(), ScanOptions.NONE);
+
+			verify(jedisSpy, never()).close();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test
+		public void zScanShouldCloseTheConnectionWhenCursorIsClosed() throws IOException {
+
+			doReturn(new ScanResult<String>("0", Collections.<String> emptyList())).when(jedisSpy).zscan(any(byte[].class),
+					any(byte[].class), any(ScanParams.class));
+
+			Cursor<Tuple> cursor = connection.zScan("foo".getBytes(), ScanOptions.NONE);
+			cursor.close();
+
+			verify(jedisSpy, times(1)).quit();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test
+		public void hScanShouldKeepTheConnectionOpen() {
+
+			doReturn(new ScanResult<String>("0", Collections.<String> emptyList())).when(jedisSpy).hscan(any(byte[].class),
+					any(byte[].class), any(ScanParams.class));
+
+			connection.hScan("foo".getBytes(), ScanOptions.NONE);
+
+			verify(jedisSpy, never()).close();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test
+		public void hScanShouldCloseTheConnectionWhenCursorIsClosed() throws IOException {
+
+			doReturn(new ScanResult<String>("0", Collections.<String> emptyList())).when(jedisSpy).hscan(any(byte[].class),
+					any(byte[].class), any(ScanParams.class));
+
+			Cursor<Entry<byte[], byte[]>> cursor = connection.hScan("foo".getBytes(), ScanOptions.NONE);
+			cursor.close();
+
+			verify(jedisSpy, times(1)).quit();
+		}
+
 	}
 
 	public static class JedisConnectionPipelineUnitTests extends JedisConnectionUnitTests {
@@ -267,6 +392,70 @@ public class JedisConnectionUnitTestSuite {
 			super.slaveOfNoOneShouldBeSentCorrectly();
 		}
 
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test(expected = UnsupportedOperationException.class)
+		public void scanShouldKeepTheConnectionOpen() {
+			super.scanShouldKeepTheConnectionOpen();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test(expected = UnsupportedOperationException.class)
+		public void scanShouldCloseTheConnectionWhenCursorIsClosed() throws IOException {
+			super.scanShouldCloseTheConnectionWhenCursorIsClosed();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test(expected = UnsupportedOperationException.class)
+		public void sScanShouldKeepTheConnectionOpen() {
+			super.sScanShouldKeepTheConnectionOpen();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test(expected = UnsupportedOperationException.class)
+		public void sScanShouldCloseTheConnectionWhenCursorIsClosed() throws IOException {
+			super.sScanShouldCloseTheConnectionWhenCursorIsClosed();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test(expected = UnsupportedOperationException.class)
+		public void zScanShouldKeepTheConnectionOpen() {
+			super.zScanShouldKeepTheConnectionOpen();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test(expected = UnsupportedOperationException.class)
+		public void zScanShouldCloseTheConnectionWhenCursorIsClosed() throws IOException {
+			super.zScanShouldCloseTheConnectionWhenCursorIsClosed();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test(expected = UnsupportedOperationException.class)
+		public void hScanShouldKeepTheConnectionOpen() {
+			super.hScanShouldKeepTheConnectionOpen();
+		}
+
+		/**
+		 * @see DATAREDIS-531
+		 */
+		@Test(expected = UnsupportedOperationException.class)
+		public void hScanShouldCloseTheConnectionWhenCursorIsClosed() throws IOException {
+			super.hScanShouldCloseTheConnectionWhenCursorIsClosed();
+		}
+
 	}
 
 	/**

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

@@ -0,0 +1,134 @@
+/*
+ * Copyright 2016 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 static org.hamcrest.MatcherAssert.*;
+import static org.hamcrest.Matchers.*;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.springframework.data.redis.ConnectionFactoryTracker;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.BoundHashOperations;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+/**
+ * @autor Mark Paluch
+ * @author Christoph Strobl
+ */
+@RunWith(Parameterized.class)
+public class ScanTests {
+
+	RedisConnectionFactory factory;
+	RedisTemplate<String, String> redisOperations;
+
+	ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10, 1, TimeUnit.MINUTES,
+			new LinkedBlockingDeque<Runnable>());
+
+	public ScanTests(RedisConnectionFactory factory) {
+
+		this.factory = factory;
+		ConnectionFactoryTracker.add(factory);
+	}
+
+	@Parameters
+	public static List<RedisConnectionFactory> params() {
+
+		JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
+		jedisConnectionFactory.setHostName("127.0.0.1");
+		jedisConnectionFactory.setPort(6379);
+		jedisConnectionFactory.afterPropertiesSet();
+
+		LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory();
+		lettuceConnectionFactory.setHostName("127.0.0.1");
+		lettuceConnectionFactory.setPort(6379);
+		lettuceConnectionFactory.afterPropertiesSet();
+
+		return Arrays.<RedisConnectionFactory> asList(jedisConnectionFactory, lettuceConnectionFactory);
+	}
+
+	@AfterClass
+	public static void cleanUp() {
+		ConnectionFactoryTracker.cleanUp();
+	}
+
+	@Before
+	public void setUp() {
+
+		redisOperations = new StringRedisTemplate(factory);
+		redisOperations.afterPropertiesSet();
+	}
+
+	@Test
+	public void contextLoads() throws InterruptedException {
+
+		BoundHashOperations<String, String, String> hash = redisOperations.boundHashOps("hash");
+		final AtomicReference<Exception> exception = new AtomicReference<Exception>();
+
+		// Create some keys so that SCAN requires a while to return all data.
+		for (int i = 0; i < 10000; i++) {
+			hash.put("key-" + i, "value");
+		}
+
+		// Concurrent access
+		for (int i = 0; i < 10; i++) {
+
+			executor.submit(new Runnable() {
+				@Override
+				public void run() {
+					try {
+
+						Cursor<Entry<Object, Object>> cursorMap = redisOperations.boundHashOps("hash")
+								.scan(ScanOptions.scanOptions().match("*").count(100).build());
+
+						// This line invokes the lazy SCAN invocation
+						while (cursorMap.hasNext()) {
+							cursorMap.next();
+						}
+						cursorMap.close();
+					} catch (Exception e) {
+						exception.set(e);
+					}
+				}
+			});
+		}
+
+		// Wait until work is finished
+		while (executor.getActiveCount() > 0) {
+			Thread.sleep(100);
+		}
+
+		executor.shutdown();
+
+		assertThat(exception.get(), is(nullValue()));
+	}
+}

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

@@ -18,9 +18,9 @@ package org.springframework.data.redis.core;
 import static org.hamcrest.CoreMatchers.*;
 import static org.junit.Assert.*;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Iterator;
 import java.util.Map;
 
 import org.junit.After;
@@ -154,7 +154,7 @@ public class DefaultHashOperationsTests<K, HK, HV> {
 	 */
 	@Test
 	@IfProfileValue(name = "redisVersion", value = "2.8+")
-	public void testHScanReadsValuesFully() {
+	public void testHScanReadsValuesFully() throws IOException {
 
 		K key = keyFactory.instance();
 		HK key1 = hashKeyFactory.instance();
@@ -164,7 +164,7 @@ public class DefaultHashOperationsTests<K, HK, HV> {
 		hashOps.put(key, key1, val1);
 		hashOps.put(key, key2, val2);
 
-		Iterator<Map.Entry<HK, HV>> it = hashOps.scan(key, ScanOptions.scanOptions().count(1).build());
+		Cursor<Map.Entry<HK, HV>> it = hashOps.scan(key, ScanOptions.scanOptions().count(1).build());
 
 		long count = 0;
 		while (it.hasNext()) {
@@ -174,6 +174,7 @@ public class DefaultHashOperationsTests<K, HK, HV> {
 			count++;
 		}
 
+		it.close();
 		assertThat(count, is(hashOps.size(key)));
 	}
 

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

@@ -20,10 +20,10 @@ import static org.junit.Assert.*;
 import static org.junit.Assume.*;
 import static org.springframework.data.redis.matcher.RedisTestMatchers.*;
 
+import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
 
@@ -216,7 +216,7 @@ public class DefaultSetOperationsTests<K, V> {
 	@Test
 	@SuppressWarnings("unchecked")
 	@IfProfileValue(name = "redisVersion", value = "2.8+")
-	public void testSSCanReadsValuesFully() {
+	public void testSSCanReadsValuesFully() throws IOException {
 
 		K key = keyFactory.instance();
 		V v1 = valueFactory.instance();
@@ -224,12 +224,13 @@ public class DefaultSetOperationsTests<K, V> {
 		V v3 = valueFactory.instance();
 
 		setOps.add(key, v1, v2, v3);
-		Iterator<V> it = setOps.scan(key, ScanOptions.scanOptions().count(1).build());
+		Cursor<V> it = setOps.scan(key, ScanOptions.scanOptions().count(1).build());
 		long count = 0;
 		while (it.hasNext()) {
 			assertThat(it.next(), anyOf(equalTo(v1), equalTo(v2), equalTo(v3)));
 			count++;
 		}
+		it.close();
 		assertThat(count, is(setOps.size(key)));
 	}
 

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

@@ -22,10 +22,10 @@ import static org.junit.Assert.*;
 import static org.junit.Assume.*;
 import static org.springframework.data.redis.matcher.RedisTestMatchers.*;
 
+import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
-import java.util.Iterator;
 import java.util.LinkedHashSet;
 import java.util.Set;
 
@@ -355,7 +355,7 @@ public class DefaultZSetOperationsTests<K, V> {
 	 */
 	@Test
 	@IfProfileValue(name = "redisVersion", value = "2.8+")
-	public void testZScanShouldReadEntireValueRange() {
+	public void testZScanShouldReadEntireValueRange() throws IOException {
 
 		K key = keyFactory.instance();
 
@@ -374,12 +374,13 @@ public class DefaultZSetOperationsTests<K, V> {
 		zSetOps.add(key, values);
 
 		int count = 0;
-		Iterator<TypedTuple<V>> it = zSetOps.scan(key, ScanOptions.scanOptions().count(2).build());
+		Cursor<TypedTuple<V>> it = zSetOps.scan(key, ScanOptions.scanOptions().count(2).build());
 		while (it.hasNext()) {
 			assertThat(it.next(), anyOf(equalTo(tuple1), equalTo(tuple2), equalTo(tuple3)));
 			count++;
 		}
 
+		it.close();
 		assertThat(count, equalTo(3));
 	}
 }

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

@@ -23,11 +23,13 @@ import static org.mockito.Mockito.*;
 
 import java.io.Serializable;
 
+import org.hamcrest.core.IsSame;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.runners.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;
@@ -94,8 +96,37 @@ public class RedisTemplateUnitTests {
 		assertThat(deserialized.getClass().getClassLoader(), is((ClassLoader) scl));
 	}
 
+	/**
+	 * @see DATAREDIS-531
+	 */
+	@Test
+	public void executeWithStickyConnectionShouldNotCloseConnectionWhenDone() {
+
+		CapturingCallback callback = new CapturingCallback();
+		template.executeWithStickyConnection(callback);
+
+		assertThat(callback.getConnection(), IsSame.sameInstance(redisConnectionMock));
+		verify(redisConnectionMock, never()).close();
+	}
+
 	static class SomeArbitrarySeriaizableObject implements Serializable {
 		private static final long serialVersionUID = -5973659324040506423L;
 	}
 
+	static class CapturingCallback implements RedisCallback<Cursor<Object>> {
+
+		private RedisConnection connection;
+
+		@Override
+		public Cursor<Object> doInRedis(RedisConnection connection) throws DataAccessException {
+			this.connection = connection;
+			return null;
+		}
+
+		public RedisConnection getConnection() {
+			return connection;
+		}
+
+	}
+
 }

src/test/java/org/springframework/data/redis/support/collections/AbstractRedisMapTests.java

@@ -20,11 +20,11 @@ import static org.junit.Assert.*;
 import static org.junit.Assume.*;
 import static org.springframework.data.redis.matcher.RedisTestMatchers.*;
 
+import java.io.IOException;
 import java.text.DecimalFormat;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.Map;
@@ -47,6 +47,7 @@ import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.ConnectionUtils;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.core.RedisTemplate;
@@ -493,7 +494,7 @@ public abstract class AbstractRedisMapTests<K, V> {
 	@Test
 	@IfProfileValue(name = "redisVersion", value = "2.8+")
 	@WithRedisDriver({ RedisDriver.JEDIS, RedisDriver.LETTUCE })
-	public void testScanWorksCorrectly() {
+	public void testScanWorksCorrectly() throws IOException {
 
 		K k1 = getKey();
 		K k2 = getKey();
@@ -504,11 +505,12 @@ public abstract class AbstractRedisMapTests<K, V> {
 		map.put(k1, v1);
 		map.put(k2, v2);
 
-		Iterator<Entry<K, V>> it = map.scan();
-		while (it.hasNext()) {
-			Entry<K, V> entry = it.next();
+		Cursor<Entry<K, V>> cursor = (Cursor<Entry<K, V>>) map.scan();
+		while (cursor.hasNext()) {
+			Entry<K, V> entry = cursor.next();
 			assertThat(entry.getKey(), anyOf(equalTo(k1), equalTo(k2)));
 			assertThat(entry.getValue(), anyOf(equalTo(v1), equalTo(v2)));
 		}
+		cursor.close();
 	}
 }

src/test/java/org/springframework/data/redis/support/collections/AbstractRedisSetTests.java

@@ -19,6 +19,7 @@ import static org.hamcrest.CoreMatchers.*;
 import static org.junit.Assert.*;
 import static org.springframework.data.redis.matcher.RedisTestMatchers.*;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
@@ -31,6 +32,7 @@ import org.junit.Test;
 import org.springframework.data.redis.ObjectFactory;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.BoundSetOperations;
+import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.test.util.MinimumRedisVersionRule;
 import org.springframework.data.redis.test.util.RedisClientRule;
@@ -308,14 +310,15 @@ public abstract class AbstractRedisSetTests<T> extends AbstractRedisCollectionTe
 	@IfProfileValue(name = "redisVersion", value = "2.8+")
 	@Test
 	@WithRedisDriver({ RedisDriver.JEDIS, RedisDriver.LETTUCE })
-	public void testScanWorksCorrectly() {
+	public void testScanWorksCorrectly() throws IOException {
 
 		Object[] expectedArray = new Object[] { getT(), getT(), getT() };
 		collection.addAll((List<T>) Arrays.asList(expectedArray));
 
-		Iterator<T> it = set.scan();
-		while (it.hasNext()) {
-			assertThat(it.next(), anyOf(equalTo(expectedArray[0]), equalTo(expectedArray[1]), equalTo(expectedArray[2])));
+		Cursor<T> cursor = (Cursor<T>) set.scan();
+		while (cursor.hasNext()) {
+			assertThat(cursor.next(), anyOf(equalTo(expectedArray[0]), equalTo(expectedArray[1]), equalTo(expectedArray[2])));
 		}
+		cursor.close();
 	}
 }

src/test/java/org/springframework/data/redis/support/collections/AbstractRedisZSetTest.java

@@ -20,6 +20,7 @@ import static org.junit.Assert.*;
 import static org.junit.Assume.*;
 import static org.springframework.data.redis.matcher.RedisTestMatchers.*;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
@@ -37,6 +38,7 @@ import org.springframework.data.redis.connection.ConnectionUtils;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisZSetCommands;
 import org.springframework.data.redis.core.BoundZSetOperations;
+import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
 import org.springframework.data.redis.test.util.MinimumRedisVersionRule;
@@ -645,7 +647,7 @@ public abstract class AbstractRedisZSetTest<T> extends AbstractRedisCollectionTe
 	@IfProfileValue(name = "redisVersion", value = "2.8+")
 	@Test
 	@WithRedisDriver({ RedisDriver.JEDIS, RedisDriver.LETTUCE })
-	public void testScanWorksCorrectly() {
+	public void testScanWorksCorrectly() throws IOException {
 
 		T t1 = getT();
 		T t2 = getT();
@@ -657,9 +659,12 @@ public abstract class AbstractRedisZSetTest<T> extends AbstractRedisCollectionTe
 		zSet.add(t3, 3);
 		zSet.add(t4, 4);
 
-		Iterator<T> it = zSet.scan();
-		while (it.hasNext()) {
-			assertThat(it.next(), anyOf(equalTo(t1), equalTo(t2), equalTo(t3), equalTo(t4)));
+		Cursor<T> cursor = (Cursor<T>) zSet.scan();
+		while (cursor.hasNext()) {
+			assertThat(cursor.next(), anyOf(equalTo(t1), equalTo(t2), equalTo(t3), equalTo(t4)));
 		}
+
+		cursor.close();
+
 	}
 }

src/test/java/org/springframework/data/redis/support/collections/RedisPropertiesTests.java

@@ -18,6 +18,7 @@ package org.springframework.data.redis.support.collections;
 import static org.junit.Assert.*;
 
 import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.PrintWriter;
 import java.io.StringWriter;
@@ -225,7 +226,7 @@ public class RedisPropertiesTests extends RedisMapTests {
 
 	@Override
 	@Test(expected = UnsupportedOperationException.class)
-	public void testScanWorksCorrectly() {
+	public void testScanWorksCorrectly() throws IOException {
 		super.testScanWorksCorrectly();
 	}