diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java
index 4131a5230..8444aa256 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveGeoOperations.java
@@ -15,8 +15,6 @@
  */
 package org.springframework.data.redis.core;
 
-import org.springframework.data.redis.domain.geo.GeoReference;
-import org.springframework.data.redis.domain.geo.GeoReference.GeoMemberReference;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -28,6 +26,7 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.data.geo.Circle;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResult;
@@ -37,6 +36,8 @@ import org.springframework.data.redis.connection.ReactiveGeoCommands;
 import org.springframework.data.redis.connection.RedisGeoCommands;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
 import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
+import org.springframework.data.redis.domain.geo.GeoReference;
+import org.springframework.data.redis.domain.geo.GeoReference.GeoMemberReference;
 import org.springframework.data.redis.domain.geo.GeoShape;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.util.Assert;
@@ -320,7 +321,7 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 		Assert.notEmpty(members, "Members must not be null or empty!");
 		Assert.noNullElements(members, "Members must not contain null elements!");
 
-		return template.createMono(connection -> Flux.fromArray(members) //
+		return template.doCreateMono(connection -> Flux.fromArray(members) //
 				.map(this::rawValue) //
 				.collectList() //
 				.flatMap(serialized -> connection.zSetCommands().zRem(rawKey(key), serialized)));
@@ -335,10 +336,10 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveGeoOperations#search(K, RedisGeoCommands.GeoReference, GeoShape, GeoSearchCommandArgs)
 	 */
@@ -350,11 +351,11 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 		Assert.notNull(reference, "GeoReference must not be null!");
 		GeoReference<ByteBuffer> rawReference = getGeoReference(reference);
 
-		return template.createFlux(connection -> connection.geoCommands()
+		return template.doCreateFlux(connection -> connection.geoCommands()
 				.geoSearch(rawKey(key), rawReference, geoPredicate, args).map(this::readGeoResult));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveGeoOperations#searchAndStore(K, K, RedisGeoCommands.GeoReference, GeoShape, GeoSearchStoreCommandArgs)
 	 */
@@ -366,7 +367,7 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 		Assert.notNull(reference, "GeoReference must not be null!");
 		GeoReference<ByteBuffer> rawReference = getGeoReference(reference);
 
-		return template.createMono(connection -> connection.geoCommands().geoSearchStore(rawKey(destKey), rawKey(key),
+		return template.doCreateMono(connection -> connection.geoCommands().geoSearchStore(rawKey(destKey), rawKey(key),
 				rawReference, geoPredicate, args));
 	}
 
@@ -374,14 +375,14 @@ class DefaultReactiveGeoOperations<K, V> implements ReactiveGeoOperations<K, V>
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.geoCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.geoCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveGeoCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.geoCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.geoCommands()));
 	}
 
 	@SuppressWarnings("unchecked")
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java
index 08b3ba083..54f863770 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveHashOperations.java
@@ -15,20 +15,20 @@
  */
 package org.springframework.data.redis.core;
 
-import org.springframework.data.redis.connection.convert.Converters;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.data.redis.connection.ReactiveHashCommands;
+import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.util.Assert;
 
@@ -125,7 +125,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(hashKey, "Hash key must not be null!");
 
-		return template.createMono(connection -> connection //
+		return template.doCreateMono(connection -> connection //
 				.numberCommands() //
 				.hIncrBy(rawKey(key), rawHashKey(hashKey), delta));
 	}
@@ -140,7 +140,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(hashKey, "Hash key must not be null!");
 
-		return template.createMono(connection -> connection //
+		return template.doCreateMono(connection -> connection //
 				.numberCommands() //
 				.hIncrBy(rawKey(key), rawHashKey(hashKey), delta));
 	}
@@ -154,7 +154,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection //
+		return template.doCreateMono(connection -> connection //
 				.hashCommands().hRandField(rawKey(key))).map(this::readHashKey);
 	}
 
@@ -167,7 +167,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection //
+		return template.doCreateMono(connection -> connection //
 				.hashCommands().hRandFieldWithValues(rawKey(key))).map(this::deserializeHashEntry);
 	}
 
@@ -180,7 +180,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createFlux(connection -> connection //
+		return template.doCreateFlux(connection -> connection //
 				.hashCommands().hRandField(rawKey(key), count)).map(this::readHashKey);
 	}
 
@@ -193,7 +193,7 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createFlux(connection -> connection //
+		return template.doCreateFlux(connection -> connection //
 				.hashCommands().hRandFieldWithValues(rawKey(key), count)).map(this::deserializeHashEntry);
 	}
 
@@ -314,21 +314,21 @@ class DefaultReactiveHashOperations<H, HK, HV> implements ReactiveHashOperations
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveHashCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.hashCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.hashCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveHashCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.hashCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.hashCommands()));
 	}
 
 	private ByteBuffer rawKey(H key) {
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveHyperLogLogOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveHyperLogLogOperations.java
index 3fdb8af4d..98f899365 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveHyperLogLogOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveHyperLogLogOperations.java
@@ -103,14 +103,14 @@ class DefaultReactiveHyperLogLogOperations<K, V> implements ReactiveHyperLogLogO
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveHyperLogLogCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.hyperLogLogCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.hyperLogLogCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveListOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveListOperations.java
index adca75fb8..248178591 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveListOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveListOperations.java
@@ -200,7 +200,7 @@ class DefaultReactiveListOperations<K, V> implements ReactiveListOperations<K, V
 		return createMono(connection -> connection.lInsert(rawKey(key), Position.AFTER, rawValue(pivot), rawValue(value)));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveListOperations#move(K, Direction, K, Direction)
 	 */
@@ -216,7 +216,7 @@ class DefaultReactiveListOperations<K, V> implements ReactiveListOperations<K, V
 				connection -> connection.lMove(rawKey(sourceKey), rawKey(destinationKey), from, to).map(this::readValue));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveListOperations#move(K, Direction, K, Direction, Duration)
 	 */
@@ -378,21 +378,21 @@ class DefaultReactiveListOperations<K, V> implements ReactiveListOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveListCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.listCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.listCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveListCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.listCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.listCommands()));
 	}
 
 	private boolean isZeroOrGreater1Second(Duration timeout) {
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java
index 1f1ba2adf..6d287c36e 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveSetOperations.java
@@ -508,21 +508,21 @@ class DefaultReactiveSetOperations<K, V> implements ReactiveSetOperations<K, V>
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveSetCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.setCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.setCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveSetCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.setCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.setCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java
index 0fc811c99..1ecbd97c9 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java
@@ -15,7 +15,6 @@
  */
 package org.springframework.data.redis.core;
 
-import org.springframework.data.redis.connection.convert.Converters;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -28,10 +27,12 @@ import java.util.Map.Entry;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.core.convert.ConversionService;
 import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.ReactiveStreamCommands;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
+import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.connection.stream.ByteBufferRecord;
 import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.MapRecord;
@@ -372,14 +373,14 @@ class DefaultReactiveStreamOperations<K, HK, HV> implements ReactiveStreamOperat
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.streamCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.streamCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveStreamCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.streamCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.streamCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java
index 31c8f0849..a64e50116 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java
@@ -244,7 +244,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().incr(rawKey(key)));
+		return template.doCreateMono(connection -> connection.numberCommands().incr(rawKey(key)));
 	}
 
 	/* (non-Javadoc)
@@ -255,7 +255,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().incrBy(rawKey(key), delta));
+		return template.doCreateMono(connection -> connection.numberCommands().incrBy(rawKey(key), delta));
 	}
 
 	/* (non-Javadoc)
@@ -266,7 +266,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().incrBy(rawKey(key), delta));
+		return template.doCreateMono(connection -> connection.numberCommands().incrBy(rawKey(key), delta));
 	}
 
 	/* (non-Javadoc)
@@ -277,7 +277,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().decr(rawKey(key)));
+		return template.doCreateMono(connection -> connection.numberCommands().decr(rawKey(key)));
 	}
 
 	/* (non-Javadoc)
@@ -288,7 +288,7 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.numberCommands().decrBy(rawKey(key), delta));
+		return template.doCreateMono(connection -> connection.numberCommands().decrBy(rawKey(key), delta));
 	}
 
 	/* (non-Javadoc)
@@ -380,14 +380,14 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveStringCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.stringCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.stringCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java
index 7b57c80ea..ced2d5afa 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java
@@ -120,7 +120,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createMono(connection -> connection.zIncrBy(rawKey(key), delta, rawValue(value)));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#randomMember(K)
 	 */
@@ -132,7 +132,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createMono(connection -> connection.zRandMember(rawKey(key))).map(this::readValue);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#distinctRandomMembers(K, long)
 	 */
@@ -145,7 +145,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createFlux(connection -> connection.zRandMember(rawKey(key), count)).map(this::readValue);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#randomMembers(K, long)
 	 */
@@ -158,7 +158,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createFlux(connection -> connection.zRandMember(rawKey(key), -count)).map(this::readValue);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#randomMemberWithScore(K)
 	 */
@@ -170,7 +170,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createMono(connection -> connection.zRandMemberWithScore(rawKey(key))).map(this::readTypedTuple);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#distinctRandomMembersWithScore(K, long)
 	 */
@@ -183,7 +183,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createFlux(connection -> connection.zRandMemberWithScore(rawKey(key), count)).map(this::readTypedTuple);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#randomMembersWithScore(K, long)
 	 */
@@ -576,7 +576,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createMono(connection -> connection.zRemRangeByScore(rawKey(key), range));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#difference(K, Collection)
 	 */
@@ -592,7 +592,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zDiff).map(this::readValue));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#differenceWithScores(K, Collection)
 	 */
@@ -608,7 +608,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zDiffWithScores).map(this::readTypedTuple));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#differenceAndStore(K, Collection, K)
 	 */
@@ -626,7 +626,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersect(K, Collection)
 	 */
@@ -642,7 +642,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zInter).map(this::readValue));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectWithScores(K, Collection)
 	 */
@@ -658,7 +658,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zInterWithScores).map(this::readTypedTuple));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#intersectWithScores(K, Collection, Aggregate, Weights)
 	 */
@@ -714,7 +714,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMap(serialized -> connection.zInterStore(rawKey(destKey), serialized, weights, aggregate)));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#union(K, Collection)
 	 */
@@ -730,7 +730,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zUnion).map(this::readValue));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionWithScores(K, Collection)
 	 */
@@ -746,7 +746,7 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 				.flatMapMany(connection::zUnionWithScores).map(this::readTypedTuple));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#unionWithScores(K, Collection, Aggregate, Weights)
 	 */
@@ -877,21 +877,21 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return template.createMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
+		return template.doCreateMono(connection -> connection.keyCommands().del(rawKey(key))).map(l -> l != 0);
 	}
 
 	private <T> Mono<T> createMono(Function<ReactiveZSetCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createMono(connection -> function.apply(connection.zSetCommands()));
+		return template.doCreateMono(connection -> function.apply(connection.zSetCommands()));
 	}
 
 	private <T> Flux<T> createFlux(Function<ReactiveZSetCommands, Publisher<T>> function) {
 
 		Assert.notNull(function, "Function must not be null!");
 
-		return template.createFlux(connection -> function.apply(connection.zSetCommands()));
+		return template.doCreateFlux(connection -> function.apply(connection.zSetCommands()));
 	}
 
 	private ByteBuffer rawKey(K key) {
diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
index 6ae24d0fc..48b576458 100644
--- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
+++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
@@ -173,6 +173,21 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		return Flux.from(doInConnection(callback, exposeConnection));
 	}
 
+	/**
+	 * Internal variant of {@link #createFlux(ReactiveRedisCallback)} bypassing proxy creation. Create a reusable Flux for
+	 * a {@link ReactiveRedisCallback}. Callback is executed within a connection context. The connection is released
+	 * outside the callback.
+	 *
+	 * @param callback must not be {@literal null}
+	 * @return a {@link Flux} wrapping the {@link ReactiveRedisCallback}.
+	 */
+	<T> Flux<T> doCreateFlux(ReactiveRedisCallback<T> callback) {
+
+		Assert.notNull(callback, "ReactiveRedisCallback must not be null!");
+
+		return Flux.from(doInConnection(callback, true));
+	}
+
 	/**
 	 * Create a reusable Mono for a {@link ReactiveRedisCallback}. Callback is executed within a connection context. The
 	 * connection is released outside the callback.
@@ -187,6 +202,21 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		return Mono.from(doInConnection(callback, exposeConnection));
 	}
 
+	/**
+	 * Internal variant of {@link #createMono(ReactiveRedisCallback)} bypassing proxy creation. Create a reusable Mono for
+	 * a {@link ReactiveRedisCallback}. Callback is executed within a connection context. The connection is released
+	 * outside the callback.
+	 *
+	 * @param callback must not be {@literal null}
+	 * @return a {@link Mono} wrapping the {@link ReactiveRedisCallback}.
+	 */
+	<T> Mono<T> doCreateMono(ReactiveRedisCallback<T> callback) {
+
+		Assert.notNull(callback, "ReactiveRedisCallback must not be null!");
+
+		return Mono.from(doInConnection(callback, true));
+	}
+
 	/**
 	 * Executes the given action object within a connection that can be exposed or not. Additionally, the connection can
 	 * be pipelined. Note the results of the pipeline are discarded (making it suitable for write-only scenarios).
@@ -225,7 +255,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.hasText(destination, "Destination channel must not be empty!");
 		Assert.notNull(message, "Message must not be null!");
 
-		return createMono(connection -> connection.pubSubCommands().publish(
+		return doCreateMono(connection -> connection.pubSubCommands().publish(
 				getSerializationContext().getStringSerializationPair().write(destination),
 				getSerializationContext().getValueSerializationPair().write(message)));
 	}
@@ -274,7 +304,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(sourceKey, "Source key must not be null!");
 		Assert.notNull(targetKey, "Target key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().copy(rawKey(sourceKey), rawKey(targetKey), replace));
+		return doCreateMono(connection -> connection.keyCommands().copy(rawKey(sourceKey), rawKey(targetKey), replace));
 	}
 
 	/*
@@ -286,7 +316,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().exists(rawKey(key)));
+		return doCreateMono(connection -> connection.keyCommands().exists(rawKey(key)));
 	}
 
 	/*
@@ -298,7 +328,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().type(rawKey(key)));
+		return doCreateMono(connection -> connection.keyCommands().type(rawKey(key)));
 	}
 
 	/*
@@ -310,7 +340,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(pattern, "Pattern must not be null!");
 
-		return createFlux(connection -> connection.keyCommands().keys(rawKey(pattern))) //
+		return doCreateFlux(connection -> connection.keyCommands().keys(rawKey(pattern))) //
 				.flatMap(Flux::fromIterable) //
 				.map(this::readKey);
 	}
@@ -324,7 +354,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(options, "ScanOptions must not be null!");
 
-		return createFlux(connection -> connection.keyCommands().scan(options)) //
+		return doCreateFlux(connection -> connection.keyCommands().scan(options)) //
 				.map(this::readKey);
 	}
 
@@ -334,7 +364,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 	 */
 	@Override
 	public Mono<K> randomKey() {
-		return createMono(connection -> connection.keyCommands().randomKey()).map(this::readKey);
+		return doCreateMono(connection -> connection.keyCommands().randomKey()).map(this::readKey);
 	}
 
 	/*
@@ -347,7 +377,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(oldKey, "Old key must not be null!");
 		Assert.notNull(newKey, "New Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().rename(rawKey(oldKey), rawKey(newKey)));
+		return doCreateMono(connection -> connection.keyCommands().rename(rawKey(oldKey), rawKey(newKey)));
 	}
 
 	/*
@@ -360,7 +390,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(oldKey, "Old key must not be null!");
 		Assert.notNull(newKey, "New Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().renameNX(rawKey(oldKey), rawKey(newKey)));
+		return doCreateMono(connection -> connection.keyCommands().renameNX(rawKey(oldKey), rawKey(newKey)));
 	}
 
 	/*
@@ -376,11 +406,11 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.noNullElements(keys, "Keys must not contain null elements!");
 
 		if (keys.length == 1) {
-			return createMono(connection -> connection.keyCommands().del(rawKey(keys[0])));
+			return doCreateMono(connection -> connection.keyCommands().del(rawKey(keys[0])));
 		}
 
 		Mono<List<ByteBuffer>> listOfKeys = Flux.fromArray(keys).map(this::rawKey).collectList();
-		return createMono(connection -> listOfKeys.flatMap(rawKeys -> connection.keyCommands().mDel(rawKeys)));
+		return doCreateMono(connection -> listOfKeys.flatMap(rawKeys -> connection.keyCommands().mDel(rawKeys)));
 	}
 
 	/*
@@ -392,7 +422,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(keys, "Keys must not be null!");
 
-		return createFlux(connection -> connection.keyCommands() //
+		return doCreateFlux(connection -> connection.keyCommands() //
 				.mDel(Flux.from(keys).map(this::rawKey).buffer(128)) //
 				.map(CommandResponse::getOutput)) //
 						.collect(Collectors.summingLong(value -> value));
@@ -411,11 +441,11 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.noNullElements(keys, "Keys must not contain null elements!");
 
 		if (keys.length == 1) {
-			return createMono(connection -> connection.keyCommands().unlink(rawKey(keys[0])));
+			return doCreateMono(connection -> connection.keyCommands().unlink(rawKey(keys[0])));
 		}
 
 		Mono<List<ByteBuffer>> listOfKeys = Flux.fromArray(keys).map(this::rawKey).collectList();
-		return createMono(connection -> listOfKeys.flatMap(rawKeys -> connection.keyCommands().mUnlink(rawKeys)));
+		return doCreateMono(connection -> listOfKeys.flatMap(rawKeys -> connection.keyCommands().mUnlink(rawKeys)));
 	}
 
 	/*
@@ -427,7 +457,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(keys, "Keys must not be null!");
 
-		return createFlux(connection -> connection.keyCommands() //
+		return doCreateFlux(connection -> connection.keyCommands() //
 				.mUnlink(Flux.from(keys).map(this::rawKey).buffer(128)) //
 				.map(CommandResponse::getOutput)) //
 						.collect(Collectors.summingLong(value -> value));
@@ -444,11 +474,11 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(timeout, "Timeout must not be null!");
 
 		if (timeout.getNano() == 0) {
-			return createMono(connection -> connection.keyCommands() //
+			return doCreateMono(connection -> connection.keyCommands() //
 					.expire(rawKey(key), timeout));
 		}
 
-		return createMono(connection -> connection.keyCommands().pExpire(rawKey(key), timeout));
+		return doCreateMono(connection -> connection.keyCommands().pExpire(rawKey(key), timeout));
 	}
 
 	/*
@@ -462,11 +492,11 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		Assert.notNull(expireAt, "Expire at must not be null!");
 
 		if (expireAt.getNano() == 0) {
-			return createMono(connection -> connection.keyCommands() //
+			return doCreateMono(connection -> connection.keyCommands() //
 					.expireAt(rawKey(key), expireAt));
 		}
 
-		return createMono(connection -> connection.keyCommands().pExpireAt(rawKey(key), expireAt));
+		return doCreateMono(connection -> connection.keyCommands().pExpireAt(rawKey(key), expireAt));
 	}
 
 	/*
@@ -478,7 +508,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().persist(rawKey(key)));
+		return doCreateMono(connection -> connection.keyCommands().persist(rawKey(key)));
 	}
 
 	/*
@@ -490,7 +520,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().pTtl(rawKey(key)).flatMap(expiry -> {
+		return doCreateMono(connection -> connection.keyCommands().pTtl(rawKey(key)).flatMap(expiry -> {
 
 			if (expiry == -1) {
 				return Mono.just(Duration.ZERO);
@@ -513,7 +543,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 
 		Assert.notNull(key, "Key must not be null!");
 
-		return createMono(connection -> connection.keyCommands().move(rawKey(key), dbIndex));
+		return doCreateMono(connection -> connection.keyCommands().move(rawKey(key), dbIndex));
 	}
 
 	// -------------------------------------------------------------------------