diff --git a/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java b/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java
index 93f9bc900..7f4c0d90b 100644
--- a/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java
+++ b/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java
@@ -187,8 +187,24 @@ public abstract class RedisConnectionUtils {
 	 *
 	 * @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 abstract class RedisConnectionUtils {
 			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");
 			}
diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java
index 8ca4f0f9f..a1a85fe8d 100644
--- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java
+++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java
@@ -234,7 +234,7 @@ public class RedisTemplate<K, V> extends RedisAccessor implements RedisOperation
 			// TODO: any other connection processing?
 			return postProcessResult(result, connToUse, existingConnection);
 		} finally {
-			RedisConnectionUtils.releaseConnection(conn, factory);
+			RedisConnectionUtils.releaseConnection(conn, factory, enableTransactionSupport);
 		}
 	}
 
diff --git a/src/test/java/org/springframework/data/redis/core/RedisTemplateUnitTests.java b/src/test/java/org/springframework/data/redis/core/RedisTemplateUnitTests.java
index 92ddc9168..027823519 100644
--- a/src/test/java/org/springframework/data/redis/core/RedisTemplateUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/core/RedisTemplateUnitTests.java
@@ -28,14 +28,20 @@ import org.junit.Test;
 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 class RedisTemplateUnitTests {
 		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;
 	}