Remove connection wrapping in StringRedisTemplate.

Wrapping is not required due to applying proper serializers. Also, the applied StringRedisConnection type cannot be easily consumed from StringRedisTemplate.

See: #3191
Original pull request: #3193

Author: christophstrobl

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

@@ -55,9 +55,4 @@ public StringRedisTemplate(RedisConnectionFactory connectionFactory) {
 		setConnectionFactory(connectionFactory);
 		afterPropertiesSet();
 	}
-
-	protected RedisConnection preProcessConnection(RedisConnection connection, boolean existingConnection) {
-		return new DefaultStringRedisConnection(connection);
-	}
-
 }

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

@@ -37,6 +37,7 @@
 import org.springframework.data.redis.Person;
 import org.springframework.data.redis.SettingsUtils;
 import org.springframework.data.redis.connection.DataType;
+import org.springframework.data.redis.connection.DefaultStringRedisConnection;
 import org.springframework.data.redis.connection.ExpirationOptions;
 import org.springframework.data.redis.connection.RedisClusterConnection;
 import org.springframework.data.redis.connection.RedisConnection;
@@ -161,7 +162,7 @@ void testTemplateNotInitialized() throws Exception {
 	void testStringTemplateExecutesWithStringConn() {
 		assumeThat(redisTemplate instanceof StringRedisTemplate).isTrue();
 		String value = redisTemplate.execute((RedisCallback<String>) connection -> {
-			StringRedisConnection stringConn = (StringRedisConnection) connection;
+			StringRedisConnection stringConn = new DefaultStringRedisConnection(connection);
 			stringConn.set("test", "it");
 			return stringConn.get("test");
 		});
@@ -307,7 +308,7 @@ void testExecutePipelinedCustomSerializer() {
 		assumeThat(redisTemplate instanceof StringRedisTemplate).isTrue();
 
 		List<Object> results = redisTemplate.executePipelined((RedisCallback) connection -> {
-			StringRedisConnection stringRedisConn = (StringRedisConnection) connection;
+			StringRedisConnection stringRedisConn = new DefaultStringRedisConnection(connection);
 			stringRedisConn.set("foo", "5");
 			stringRedisConn.get("foo");
 			stringRedisConn.rPush("foolist", "10");

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

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2025 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.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import javax.sql.DataSource;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.AfterParameterizedClassInvocation;
+import org.junit.jupiter.params.BeforeParameterizedClassInvocation;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.aggregator.ArgumentsAccessor;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mockito;
+import org.springframework.context.Lifecycle;
+import org.springframework.data.redis.SettingsUtils;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+
+/**
+ * @author Christoph Strobl
+ */
+@ParameterizedClass
+@MethodSource("argumentsStream")
+public class TransactionalStringRedisTemplateTests {
+
+	RedisConnectionFactory redisConnectionFactory;
+	StringRedisTemplate stringTemplate;
+
+	TransactionalStringRedisTemplateTests(RedisConnectionFactory redisConnectionFactory) {
+		this.redisConnectionFactory = redisConnectionFactory;
+
+		if (redisConnectionFactory instanceof Lifecycle lifecycleBean) {
+			lifecycleBean.start();
+		}
+	}
+
+	@BeforeEach
+	void beforeEach() {
+
+		stringTemplate = new StringRedisTemplate(redisConnectionFactory);
+
+		// explicitly enable transaction support
+		stringTemplate.setEnableTransactionSupport(true);
+		stringTemplate.afterPropertiesSet();
+
+		stringTemplate.execute((RedisCallback) con -> {
+			con.flushDb();
+			return null;
+		});
+	}
+
+	@AfterEach
+	void afterEach() {
+		redisConnectionFactory.getConnection().flushAll();
+	}
+
+	@Test // GH-3191
+	void visibilityDuringManagedTransaction() throws SQLException {
+
+		stringTemplate.opsForSet().add("myset", "outside");
+
+		DataSource ds = Mockito.mock(DataSource.class);
+		Mockito.when(ds.getConnection()).thenReturn(Mockito.mock(Connection.class));
+
+		DataSourceTransactionManager txMgr = new DataSourceTransactionManager(ds);
+
+		TransactionTemplate txTemplate = new TransactionTemplate(txMgr);
+		txTemplate.afterPropertiesSet();
+		Map<String, Object> result = txTemplate.execute(x -> {
+
+			Map<String, Object> operationAndOutput = new LinkedHashMap<>();
+			// visible since set outside of tx
+			operationAndOutput.put("isMember(outside)", stringTemplate.opsForSet().isMember("myset", "outside"));
+
+			// add happens inside multi/exec
+			operationAndOutput.put("add", stringTemplate.opsForSet().add("myset", "inside"));
+
+			// changes not visible though inside of tx, but command is not part of multi/exec block
+			operationAndOutput.put("isMember(inside)", stringTemplate.opsForSet().isMember("myset", "inside"));
+
+			return operationAndOutput;
+		});
+
+		assertThat(result).containsEntry("isMember(outside)", true).containsEntry("add", null)
+				.containsEntry("isMember(inside)", false);
+	}
+
+	static Stream<Arguments> argumentsStream() {
+
+		LettuceConnectionFactory lcf = new LettuceConnectionFactory(SettingsUtils.standaloneConfiguration());
+		lcf.afterPropertiesSet();
+
+		JedisConnectionFactory jcf = new JedisConnectionFactory(SettingsUtils.standaloneConfiguration());
+		jcf.afterPropertiesSet();
+
+		return Stream.of(Arguments.of(lcf), Arguments.of(jcf));
+	}
+
+	@AfterParameterizedClassInvocation
+	static void afterInvocation(ArgumentsAccessor accessor) {
+		Object o = accessor.get(0);
+		if (o instanceof Lifecycle lifecycle) {
+			lifecycle.stop();
+		}
+	}
+
+	@BeforeParameterizedClassInvocation
+	static void beforeInvocation(ArgumentsAccessor accessor) {
+		Object o = accessor.get(0);
+		if (o instanceof Lifecycle lifecycle) {
+			lifecycle.start();
+		}
+	}
+}