diff --git a/src/main/java/org/springframework/data/redis/aot/RedisRuntimeHints.java b/src/main/java/org/springframework/data/redis/aot/RedisRuntimeHints.java
index 127b0d76e7..639d1906ba 100644
--- a/src/main/java/org/springframework/data/redis/aot/RedisRuntimeHints.java
+++ b/src/main/java/org/springframework/data/redis/aot/RedisRuntimeHints.java
@@ -106,6 +106,7 @@ public void registerHints(RuntimeHints hints, @Nullable ClassLoader classLoader)
 						TypeReference.of(ReactiveClusterScriptingCommands.class),
 						TypeReference.of(ReactiveClusterGeoCommands.class),
 						TypeReference.of(ReactiveClusterHyperLogLogCommands.class), TypeReference.of(ReactiveRedisOperations.class),
+						TypeReference.of(ReactiveRedisConnectionFactory.class),
 						TypeReference.of(ReactiveRedisTemplate.class), TypeReference.of(RedisOperations.class),
 						TypeReference.of(RedisTemplate.class), TypeReference.of(StringRedisTemplate.class),
 						TypeReference.of(KeyspaceConfiguration.class), TypeReference.of(MappingConfiguration.class),
diff --git a/src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java b/src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java
index 101f4864b2..37195ef05d 100644
--- a/src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java
+++ b/src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java
@@ -15,14 +15,16 @@
  */
 package org.springframework.data.redis.cache;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
-import java.util.function.BiFunction;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
-import java.util.function.Supplier;
 
 import org.springframework.dao.PessimisticLockingFailureException;
 import org.springframework.data.redis.connection.ReactiveRedisConnection;
@@ -34,9 +36,8 @@
 import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
-
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ObjectUtils;
 
 /**
  * {@link RedisCacheWriter} implementation capable of reading/writing binary data from/to Redis in {@literal standalone}
@@ -44,8 +45,8 @@
  * {@link RedisConnection}.
  * <p>
  * {@link DefaultRedisCacheWriter} can be used in
- * {@link RedisCacheWriter#lockingRedisCacheWriter(RedisConnectionFactory) locking}
- * or {@link RedisCacheWriter#nonLockingRedisCacheWriter(RedisConnectionFactory) non-locking} mode. While
+ * {@link RedisCacheWriter#lockingRedisCacheWriter(RedisConnectionFactory) locking} or
+ * {@link RedisCacheWriter#nonLockingRedisCacheWriter(RedisConnectionFactory) non-locking} mode. While
  * {@literal non-locking} aims for maximum performance it may result in overlapping, non-atomic, command execution for
  * operations spanning multiple Redis interactions like {@code putIfAbsent}. The {@literal locking} counterpart prevents
  * command overlap by setting an explicit lock key and checking against presence of this key which leads to additional
@@ -59,6 +60,9 @@
  */
 class DefaultRedisCacheWriter implements RedisCacheWriter {
 
+	private static final boolean REACTIVE_REDIS_CONNECTION_FACTORY_PRESENT = ClassUtils
+			.isPresent("org.springframework.data.redis.connection.ReactiveRedisConnectionFactory", null);
+
 	private final BatchStrategy batchStrategy;
 
 	private final CacheStatisticsCollector statistics;
@@ -69,6 +73,8 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 
 	private final TtlFunction lockTtl;
 
+	private final AsyncCacheWriter asyncCacheWriter;
+
 	/**
 	 * @param connectionFactory must not be {@literal null}.
 	 * @param batchStrategy must not be {@literal null}.
@@ -109,6 +115,12 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 		this.lockTtl = lockTtl;
 		this.statistics = cacheStatisticsCollector;
 		this.batchStrategy = batchStrategy;
+
+		if (REACTIVE_REDIS_CONNECTION_FACTORY_PRESENT && this.connectionFactory instanceof ReactiveRedisConnectionFactory) {
+			asyncCacheWriter = new AsynchronousCacheWriterDelegate();
+		} else {
+			asyncCacheWriter = UnsupportedAsyncCacheWriter.INSTANCE;
+		}
 	}
 
 	@Override
@@ -138,8 +150,8 @@ public byte[] get(String name, byte[] key, @Nullable Duration ttl) {
 	}
 
 	@Override
-	public boolean isRetrieveSupported() {
-		return isReactive();
+	public boolean supportsAsyncRetrieve() {
+		return asyncCacheWriter.isSupported();
 	}
 
 	@Override
@@ -148,68 +160,19 @@ public CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Dur
 		Assert.notNull(name, "Name must not be null");
 		Assert.notNull(key, "Key must not be null");
 
-		CompletableFuture<byte[]> result = nonBlockingRetrieveFunction(name).apply(key, ttl);
-
-		result = result.thenApply(cachedValue -> {
-
-			statistics.incGets(name);
-
-			if (cachedValue != null) {
-				statistics.incHits(name);
-			} else {
-				statistics.incMisses(name);
-			}
-
-			return cachedValue;
-		});
-
-		return result;
-	}
-
-	private BiFunction<byte[], Duration, CompletableFuture<byte[]>> nonBlockingRetrieveFunction(String cacheName) {
-		return isReactive() ? reactiveRetrieveFunction(cacheName) : asyncRetrieveFunction(cacheName);
-	}
-
-	// TODO: Possibly remove if we rely on the default Cache.retrieve(..) behavior
-	//  after assessing RedisCacheWriter.isRetrieveSupported().
-	// Function applied for Cache.retrieve(key) when a non-reactive Redis driver is used, such as Jedis.
-	private BiFunction<byte[], Duration, CompletableFuture<byte[]>> asyncRetrieveFunction(String cacheName) {
-
-		return (key, ttl) -> {
-
-			Supplier<byte[]> getKey = () -> execute(cacheName, connection -> connection.stringCommands().get(key));
-
-			Supplier<byte[]> getKeyWithExpiration = () -> execute(cacheName, connection ->
-					connection.stringCommands().getEx(key, Expiration.from(ttl)));
-
-			return shouldExpireWithin(ttl)
-				? CompletableFuture.supplyAsync(getKeyWithExpiration)
-				: CompletableFuture.supplyAsync(getKey);
-
-		};
-	}
-
-	// Function applied for Cache.retrieve(key) when a reactive Redis driver is used, such as Lettuce.
-	private BiFunction<byte[], Duration, CompletableFuture<byte[]>> reactiveRetrieveFunction(String cacheName) {
-
-		return (key, ttl) -> {
-
-			ByteBuffer wrappedKey = ByteBuffer.wrap(key);
+		return asyncCacheWriter.retrieve(name, key, ttl) //
+				.thenApply(cachedValue -> {
 
-			Flux<?> cacheLockCheckFlux = Flux.interval(Duration.ZERO, this.sleepTime).takeUntil(count ->
-					executeLockFree(connection -> !doCheckLock(cacheName, connection)));
+					statistics.incGets(name);
 
-			Mono<ByteBuffer> getMono = shouldExpireWithin(ttl)
-					? executeReactively(connection -> connection.stringCommands().getEx(wrappedKey, Expiration.from(ttl)))
-					: executeReactively(connection -> connection.stringCommands().get(wrappedKey));
+					if (cachedValue != null) {
+						statistics.incHits(name);
+					} else {
+						statistics.incMisses(name);
+					}
 
-			Mono<ByteBuffer> result = cacheLockCheckFlux.then(getMono);
-
-			@SuppressWarnings("all")
-			Mono<byte[]> byteArrayResult = result.map(DefaultRedisCacheWriter::nullSafeGetBytes);
-
-			return byteArrayResult.toFuture();
-		};
+					return cachedValue;
+				});
 	}
 
 	@Override
@@ -222,8 +185,8 @@ public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
 		execute(name, connection -> {
 
 			if (shouldExpireWithin(ttl)) {
-				connection.stringCommands()
-						.set(key, value, Expiration.from(ttl.toMillis(), TimeUnit.MILLISECONDS), SetOption.upsert());
+				connection.stringCommands().set(key, value, Expiration.from(ttl.toMillis(), TimeUnit.MILLISECONDS),
+						SetOption.upsert());
 			} else {
 				connection.stringCommands().set(key, value);
 			}
@@ -234,6 +197,17 @@ public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
 		statistics.incPuts(name);
 	}
 
+	@Override
+	public CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
+
+		Assert.notNull(name, "Name must not be null");
+		Assert.notNull(key, "Key must not be null");
+		Assert.notNull(value, "Value must not be null");
+
+		return asyncCacheWriter.store(name, key, value, ttl) //
+				.thenRun(() -> statistics.incPuts(name));
+	}
+
 	@Override
 	public byte[] putIfAbsent(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
 
@@ -252,9 +226,10 @@ public byte[] putIfAbsent(String name, byte[] key, byte[] value, @Nullable Durat
 				boolean put;
 
 				if (shouldExpireWithin(ttl)) {
-					put = isTrue(connection.stringCommands().set(key, value, Expiration.from(ttl), SetOption.ifAbsent()));
+					put = ObjectUtils.nullSafeEquals(
+							connection.stringCommands().set(key, value, Expiration.from(ttl), SetOption.ifAbsent()), true);
 				} else {
-					put = isTrue(connection.stringCommands().setNX(key, value));
+					put = ObjectUtils.nullSafeEquals(connection.stringCommands().setNX(key, value), true);
 				}
 
 				if (put) {
@@ -348,8 +323,7 @@ private Boolean doLock(String name, Object contextualKey, @Nullable Object conte
 
 		Expiration expiration = Expiration.from(this.lockTtl.getTimeToLive(contextualKey, contextualValue));
 
-		return connection.stringCommands()
-				.set(createCacheLockKey(name), new byte[0], expiration, SetOption.SET_IF_ABSENT);
+		return connection.stringCommands().set(createCacheLockKey(name), new byte[0], expiration, SetOption.SET_IF_ABSENT);
 	}
 
 	/**
@@ -381,18 +355,6 @@ private <T> T executeLockFree(Function<RedisConnection, T> callback) {
 		}
 	}
 
-	private <T> T executeReactively(Function<ReactiveRedisConnection, T> callback) {
-
-		ReactiveRedisConnection connection = getReactiveRedisConnectionFactory().getReactiveConnection();
-
-		try {
-			return callback.apply(connection);
-		}
-		finally {
-			connection.closeLater();
-		}
-	}
-
 	/**
 	 * Determines whether this {@link RedisCacheWriter} uses locks during caching operations.
 	 *
@@ -419,40 +381,192 @@ private void checkAndPotentiallyWaitUntilUnlocked(String name, RedisConnection c
 			// Re-interrupt current Thread to allow other participants to react.
 			Thread.currentThread().interrupt();
 
-			String message = String.format("Interrupted while waiting to unlock cache %s", name);
-
-			throw new PessimisticLockingFailureException(message, cause);
+			throw new PessimisticLockingFailureException(String.format("Interrupted while waiting to unlock cache %s", name),
+					cause);
 		} finally {
 			this.statistics.incLockTime(name, System.nanoTime() - lockWaitTimeNs);
 		}
 	}
 
 	boolean doCheckLock(String name, RedisConnection connection) {
-		return isTrue(connection.keyCommands().exists(createCacheLockKey(name)));
+		return ObjectUtils.nullSafeEquals(connection.keyCommands().exists(createCacheLockKey(name)), true);
 	}
 
-	private boolean isReactive() {
-		return this.connectionFactory instanceof ReactiveRedisConnectionFactory;
+	byte[] createCacheLockKey(String name) {
+		return (name + "~lock").getBytes(StandardCharsets.UTF_8);
 	}
 
-	private ReactiveRedisConnectionFactory getReactiveRedisConnectionFactory() {
-		return (ReactiveRedisConnectionFactory) this.connectionFactory;
+	private static boolean shouldExpireWithin(@Nullable Duration ttl) {
+		return ttl != null && !ttl.isZero() && !ttl.isNegative();
 	}
 
-	private static byte[] createCacheLockKey(String name) {
-		return (name + "~lock").getBytes(StandardCharsets.UTF_8);
+	/**
+	 * Interface for asynchronous cache retrieval.
+	 *
+	 * @since 3.2
+	 */
+	interface AsyncCacheWriter {
+
+		/**
+		 * @return {@code true} if async cache operations are supported; {@code false} otherwise.
+		 */
+		boolean isSupported();
+
+		/**
+		 * Retrieve a cache entry asynchronously.
+		 *
+		 * @param name the cache name from which to retrieve the cache entry.
+		 * @param key the cache entry key.
+		 * @param ttl optional TTL to set for Time-to-Idle eviction.
+		 * @return a future that completes either with a value if the value exists or completing with {@code null} if the
+		 *         cache does not contain an entry.
+		 */
+		CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Duration ttl);
+
+		/**
+		 * Store a cache entry asynchronously.
+		 *
+		 * @param name the cache name which to store the cache entry to.
+		 * @param key the key for the cache entry. Must not be {@literal null}.
+		 * @param value the value stored for the key. Must not be {@literal null}.
+		 * @param ttl optional expiration time. Can be {@literal null}.
+		 * @return a future that signals completion.
+		 */
+		CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nullable Duration ttl);
 	}
 
-	private static boolean isTrue(@Nullable Boolean value) {
-		return Boolean.TRUE.equals(value);
-	}
+	/**
+	 * Unsupported variant of a {@link AsyncCacheWriter}.
+	 *
+	 * @since 3.2
+	 */
+	enum UnsupportedAsyncCacheWriter implements AsyncCacheWriter {
+		INSTANCE;
 
-	@Nullable
-	private static byte[] nullSafeGetBytes(@Nullable ByteBuffer value) {
-		return value != null ? ByteUtils.getBytes(value) : null;
+		@Override
+		public boolean isSupported() {
+			return false;
+		}
+
+		@Override
+		public CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Duration ttl) {
+			throw new UnsupportedOperationException("async retrieve not supported");
+		}
+
+		@Override
+		public CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
+			throw new UnsupportedOperationException("async store not supported");
+		}
 	}
 
-	private static boolean shouldExpireWithin(@Nullable Duration ttl) {
-		return ttl != null && !ttl.isZero() && !ttl.isNegative();
+	/**
+	 * Delegate implementing {@link AsyncCacheWriter} to provide asynchronous cache retrieval and storage operations using
+	 * {@link ReactiveRedisConnectionFactory}.
+	 *
+	 * @since 3.2
+	 */
+	class AsynchronousCacheWriterDelegate implements AsyncCacheWriter {
+
+		@Override
+		public boolean isSupported() {
+			return true;
+		}
+
+		@Override
+		public CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Duration ttl) {
+
+			return doWithConnection(connection -> {
+
+				ByteBuffer wrappedKey = ByteBuffer.wrap(key);
+				Mono<?> cacheLockCheckFlux;
+
+				if (isLockingCacheWriter())
+					cacheLockCheckFlux = waitForLock(connection, name);
+				else {
+					cacheLockCheckFlux = Mono.empty();
+				}
+
+				Mono<ByteBuffer> get = shouldExpireWithin(ttl)
+						? connection.stringCommands().getEx(wrappedKey, Expiration.from(ttl))
+						: connection.stringCommands().get(wrappedKey);
+
+				return cacheLockCheckFlux.then(get).map(ByteUtils::getBytes).toFuture();
+			});
+		}
+
+		@Override
+		public CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
+
+			return doWithConnection(connection -> {
+
+				Mono<?> mono;
+
+				if (isLockingCacheWriter()) {
+
+					mono = Mono.usingWhen(doLock(name, key, value, connection), unused -> doStore(key, value, ttl, connection),
+							unused -> doUnlock(name, connection));
+				} else {
+					mono = doStore(key, value, ttl, connection);
+				}
+
+				return mono.then().toFuture();
+			});
+		}
+
+		private Mono<Boolean> doStore(byte[] cacheKey, byte[] value, @Nullable Duration ttl,
+				ReactiveRedisConnection connection) {
+
+			ByteBuffer wrappedKey = ByteBuffer.wrap(cacheKey);
+			ByteBuffer wrappedValue = ByteBuffer.wrap(value);
+
+			if (shouldExpireWithin(ttl)) {
+				return connection.stringCommands().set(wrappedKey, wrappedValue,
+						Expiration.from(ttl.toMillis(), TimeUnit.MILLISECONDS), SetOption.upsert());
+			} else {
+				return connection.stringCommands().set(wrappedKey, wrappedValue);
+			}
+		}
+
+		private Mono<Object> doLock(String name, Object contextualKey, @Nullable Object contextualValue,
+				ReactiveRedisConnection connection) {
+
+			Expiration expiration = Expiration.from(lockTtl.getTimeToLive(contextualKey, contextualValue));
+
+			return connection.stringCommands()
+					.set(ByteBuffer.wrap(createCacheLockKey(name)), ByteBuffer.wrap(new byte[0]), expiration,
+							SetOption.SET_IF_ABSENT) //
+					.thenReturn(new Object()); // Ensure we emit an object, otherwise, the Mono.usingWhen operator doesn't run
+																			// the inner resource function.
+		}
+
+		private Mono<Void> doUnlock(String name, ReactiveRedisConnection connection) {
+			return connection.keyCommands().del(ByteBuffer.wrap(createCacheLockKey(name))).then();
+		}
+
+		private Mono<Void> waitForLock(ReactiveRedisConnection connection, String cacheName) {
+
+			AtomicLong lockWaitTimeNs = new AtomicLong();
+			byte[] cacheLockKey = createCacheLockKey(cacheName);
+
+			Flux<Long> wait = Flux.interval(Duration.ZERO, sleepTime);
+			Mono<Boolean> exists = connection.keyCommands().exists(ByteBuffer.wrap(cacheLockKey)).filter(it -> !it);
+
+			return wait.doOnSubscribe(subscription -> lockWaitTimeNs.set(System.nanoTime())) //
+					.flatMap(it -> exists) //
+					.doFinally(signalType -> statistics.incLockTime(cacheName, System.nanoTime() - lockWaitTimeNs.get())) //
+					.next() //
+					.then();
+		}
+
+		private <T> CompletableFuture<T> doWithConnection(
+				Function<ReactiveRedisConnection, CompletableFuture<T>> callback) {
+
+			ReactiveRedisConnectionFactory cf = (ReactiveRedisConnectionFactory) connectionFactory;
+
+			return Mono.usingWhen(Mono.fromSupplier(cf::getReactiveConnection), //
+					it -> Mono.fromCompletionStage(callback.apply(it)), //
+					ReactiveRedisConnection::closeLater) //
+					.toFuture();
+		}
 	}
 }
diff --git a/src/main/java/org/springframework/data/redis/cache/RedisCache.java b/src/main/java/org/springframework/data/redis/cache/RedisCache.java
index d153fd449b..22e1cb9912 100644
--- a/src/main/java/org/springframework/data/redis/cache/RedisCache.java
+++ b/src/main/java/org/springframework/data/redis/cache/RedisCache.java
@@ -55,7 +55,6 @@
  * @author Piotr Mionskowski
  * @author Jos Roseboom
  * @author John Blum
- * @see org.springframework.cache.support.AbstractValueAdaptingCache
  * @since 2.0
  */
 @SuppressWarnings("unused")
@@ -72,16 +71,16 @@ public class RedisCache extends AbstractValueAdaptingCache {
 	private final String name;
 
 	/**
-	 * Create a new {@link RedisCache} with the given {@link String name} and {@link RedisCacheConfiguration},
-	 * using the {@link RedisCacheWriter} to execute Redis commands supporting the cache operations.
+	 * Create a new {@link RedisCache} with the given {@link String name} and {@link RedisCacheConfiguration}, using the
+	 * {@link RedisCacheWriter} to execute Redis commands supporting the cache operations.
 	 *
 	 * @param name {@link String name} for this {@link Cache}; must not be {@literal null}.
-	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by
-	 * executing the necessary Redis commands; must not be {@literal null}.
-	 * @param cacheConfiguration {@link RedisCacheConfiguration} applied to this {@link RedisCache} on creation;
-	 * must not be {@literal null}.
+	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing the
+	 *          necessary Redis commands; must not be {@literal null}.
+	 * @param cacheConfiguration {@link RedisCacheConfiguration} applied to this {@link RedisCache} on creation; must not
+	 *          be {@literal null}.
 	 * @throws IllegalArgumentException if either the given {@link RedisCacheWriter} or {@link RedisCacheConfiguration}
-	 * are {@literal null} or the given {@link String} name for this {@link RedisCache} is {@literal null}.
+	 *           are {@literal null} or the given {@link String} name for this {@link RedisCache} is {@literal null}.
 	 */
 	protected RedisCache(String name, RedisCacheWriter cacheWriter, RedisCacheConfiguration cacheConfiguration) {
 
@@ -120,7 +119,7 @@ protected RedisCacheWriter getCacheWriter() {
 	 * accessing entries in the cache.
 	 *
 	 * @return the configured {@link ConversionService} used to convert {@link Object cache keys} to a {@link String} when
-	 * accessing entries in the cache.
+	 *         accessing entries in the cache.
 	 * @see RedisCacheConfiguration#getConversionService()
 	 * @see #getCacheConfiguration()
 	 */
@@ -218,16 +217,7 @@ private Duration getTimeToLive(Object key, @Nullable Object value) {
 	@Override
 	public void put(Object key, @Nullable Object value) {
 
-		Object cacheValue = preProcessCacheValue(value);
-
-		if (nullCacheValueIsNotAllowed(cacheValue)) {
-
-			String message = String.format("Cache '%s' does not allow 'null' values; Avoid storing null"
-					+ " via '@Cacheable(unless=\"#result == null\")' or configure RedisCache to allow 'null'"
-					+ " via RedisCacheConfiguration", getName());
-
-			throw new IllegalArgumentException(message);
-		}
+		Object cacheValue = processAndCheckValue(value);
 
 		getCacheWriter().put(getName(), createAndConvertCacheKey(key), serializeCacheValue(cacheValue),
 				getTimeToLive(key, value));
@@ -279,50 +269,71 @@ public void evict(Object key) {
 		getCacheWriter().remove(getName(), createAndConvertCacheKey(key));
 	}
 
-	/**
-	 * Customization hook called before passing object to
-	 * {@link org.springframework.data.redis.serializer.RedisSerializer}.
-	 *
-	 * @param value can be {@literal null}.
-	 * @return preprocessed value. Can be {@literal null}.
-	 */
-	@Nullable
-	protected Object preProcessCacheValue(@Nullable Object value) {
-		return value != null ? value : isAllowNullValues() ? NullValue.INSTANCE : null;
-	}
-
 	@Override
 	public CompletableFuture<?> retrieve(Object key) {
 
-		if (getCacheWriter().isRetrieveSupported()) {
-			return retrieveValue(key).thenApply(this::nullSafeDeserializedStoreValue);
+		if (!getCacheWriter().supportsAsyncRetrieve()) {
+			throw new UnsupportedOperationException(
+					"The Redis driver configured with RedisCache through RedisCacheWriter does not support CompletableFuture-based retrieval");
 		}
 
-		return super.retrieve(key);
+		return retrieveValue(key).thenApply(this::nullSafeDeserializedStoreValue);
 	}
 
 	@Override
 	@SuppressWarnings("unchecked")
 	public <T> CompletableFuture<T> retrieve(Object key, Supplier<CompletableFuture<T>> valueLoader) {
 
-		if (getCacheWriter().isRetrieveSupported()) {
-			return retrieveValue(key)
-				.thenApply(this::nullSafeDeserializedStoreValue)
-				.thenCompose(cachedValue -> cachedValue != null
-					? CompletableFuture.completedFuture((T) cachedValue)
-					: valueLoader.get());
+		if (!getCacheWriter().supportsAsyncRetrieve()) {
+			throw new UnsupportedOperationException(
+					"The Redis driver configured with RedisCache through RedisCacheWriter does not support CompletableFuture-based retrieval");
 		}
 
-		return super.retrieve(key, valueLoader);
+		return retrieveValue(key) //
+				.thenCompose(bytes -> {
+
+					if (bytes != null) {
+						return CompletableFuture.completedFuture((T) nullSafeDeserializedStoreValue(bytes));
+					}
+
+					return valueLoader.get().thenCompose(value -> {
+
+						Object cacheValue = processAndCheckValue(value);
+
+						return getCacheWriter()
+								.store(getName(), createAndConvertCacheKey(key), serializeCacheValue(cacheValue),
+										getTimeToLive(key, cacheValue)) //
+								.thenApply(v -> value);
+					});
+				});
 	}
 
-	CompletableFuture<byte[]> retrieveValue(Object key) {
-		return getCacheWriter().retrieve(getName(), createAndConvertCacheKey(key));
+	private Object processAndCheckValue(@Nullable Object value) {
+
+		Object cacheValue = preProcessCacheValue(value);
+
+		if (nullCacheValueIsNotAllowed(cacheValue)) {
+
+			String message = String.format("Cache '%s' does not allow 'null' values; Avoid storing null"
+					+ " via '@Cacheable(unless=\"#result == null\")' or configure RedisCache to allow 'null'"
+					+ " via RedisCacheConfiguration", getName());
+
+			throw new IllegalArgumentException(message);
+		}
+
+		return cacheValue;
 	}
 
+	/**
+	 * Customization hook called before passing object to
+	 * {@link org.springframework.data.redis.serializer.RedisSerializer}.
+	 *
+	 * @param value can be {@literal null}.
+	 * @return preprocessed value. Can be {@literal null}.
+	 */
 	@Nullable
-	Object nullSafeDeserializedStoreValue(@Nullable byte[] value) {
-		return value != null ? fromStoreValue(deserializeCacheValue(value)) : null;
+	protected Object preProcessCacheValue(@Nullable Object value) {
+		return value != null ? value : isAllowNullValues() ? NullValue.INSTANCE : null;
 	}
 
 	/**
@@ -357,7 +368,7 @@ protected byte[] serializeCacheValue(Object value) {
 	 *
 	 * @param value array of bytes to deserialize; must not be {@literal null}.
 	 * @return an {@link Object} deserialized from the array of bytes using the configured value
-	 * {@link RedisSerializationContext.SerializationPair}; can be {@literal null}.
+	 *         {@link RedisSerializationContext.SerializationPair}; can be {@literal null}.
 	 * @see RedisCacheConfiguration#getValueSerializationPair()
 	 */
 	@Nullable
@@ -418,13 +429,23 @@ protected String convertKey(Object key) {
 			return key.toString();
 		}
 
-		String message = String.format("Cannot convert cache key %s to String; Please register a suitable Converter"
-				+ " via 'RedisCacheConfiguration.configureKeyConverters(...)' or override '%s.toString()'",
+		String message = String.format(
+				"Cannot convert cache key %s to String; Please register a suitable Converter"
+						+ " via 'RedisCacheConfiguration.configureKeyConverters(...)' or override '%s.toString()'",
 				source, key.getClass().getName());
 
 		throw new IllegalStateException(message);
 	}
 
+	private CompletableFuture<byte[]> retrieveValue(Object key) {
+		return getCacheWriter().retrieve(getName(), createAndConvertCacheKey(key));
+	}
+
+	@Nullable
+	private Object nullSafeDeserializedStoreValue(@Nullable byte[] value) {
+		return value != null ? fromStoreValue(deserializeCacheValue(value)) : null;
+	}
+
 	private boolean hasToStringMethod(Object target) {
 		return hasToStringMethod(target.getClass());
 	}
diff --git a/src/main/java/org/springframework/data/redis/cache/RedisCacheManager.java b/src/main/java/org/springframework/data/redis/cache/RedisCacheManager.java
index c64c84ac27..84237293e1 100644
--- a/src/main/java/org/springframework/data/redis/cache/RedisCacheManager.java
+++ b/src/main/java/org/springframework/data/redis/cache/RedisCacheManager.java
@@ -34,21 +34,20 @@
 /**
  * {@link CacheManager} implementation for Redis backed by {@link RedisCache}.
  * <p>
- * This {@link CacheManager} creates {@link Cache caches} on first write, by default. Empty {@link Cache caches}
- * are not visible in Redis due to how Redis represents empty data structures.
+ * This {@link CacheManager} creates {@link Cache caches} on first write, by default. Empty {@link Cache caches} are not
+ * visible in Redis due to how Redis represents empty data structures.
  * <p>
- * {@link Cache Caches} requiring a different {@link RedisCacheConfiguration cache configuration}
- * than the {@link RedisCacheConfiguration#defaultCacheConfig() default cache configuration}
- * can be specified via {@link RedisCacheManagerBuilder#withInitialCacheConfigurations(Map)} or individually
- * using {@link RedisCacheManagerBuilder#withCacheConfiguration(String, RedisCacheConfiguration)}.
+ * {@link Cache Caches} requiring a different {@link RedisCacheConfiguration cache configuration} than the
+ * {@link RedisCacheConfiguration#defaultCacheConfig() default cache configuration} can be specified via
+ * {@link RedisCacheManagerBuilder#withInitialCacheConfigurations(Map)} or individually using
+ * {@link RedisCacheManagerBuilder#withCacheConfiguration(String, RedisCacheConfiguration)}.
  *
  * @author Christoph Strobl
  * @author Mark Paluch
  * @author Yanming Zhou
  * @author John Blum
- * @see org.springframework.cache.Cache
+ * @see RedisCache
  * @see org.springframework.cache.CacheManager
- * @see org.springframework.cache.transaction.AbstractTransactionSupportingCacheManager
  * @see org.springframework.data.redis.connection.RedisConnectionFactory
  * @see org.springframework.data.redis.cache.RedisCacheConfiguration
  * @see org.springframework.data.redis.cache.RedisCacheWriter
@@ -58,84 +57,6 @@ public class RedisCacheManager extends AbstractTransactionSupportingCacheManager
 
 	protected static final boolean DEFAULT_ALLOW_RUNTIME_CACHE_CREATION = true;
 
-	/**
-	 * Factory method returning a {@literal Builder} used to construct and configure a {@link RedisCacheManager}.
-	 *
-	 * @return new {@link RedisCacheManagerBuilder}.
-	 * @since 2.3
-	 */
-	public static RedisCacheManagerBuilder builder() {
-		return new RedisCacheManagerBuilder();
-	}
-
-	/**
-	 * Factory method returning a {@literal Builder} used to construct and configure a {@link RedisCacheManager}
-	 * initialized with the given {@link RedisCacheWriter}.
-	 *
-	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations
-	 * by executing appropriate Redis commands; must not be {@literal null}.
-	 * @return new {@link RedisCacheManagerBuilder}.
-	 * @throws IllegalArgumentException if the given {@link RedisCacheWriter} is {@literal null}.
-	 * @see org.springframework.data.redis.cache.RedisCacheWriter
-	 */
-	public static RedisCacheManagerBuilder builder(RedisCacheWriter cacheWriter) {
-
-		Assert.notNull(cacheWriter, "CacheWriter must not be null");
-
-		return RedisCacheManagerBuilder.fromCacheWriter(cacheWriter);
-	}
-
-	/**
-	 * Factory method returning a {@literal Builder} used to construct and configure a {@link RedisCacheManager}
-	 * initialized with the given {@link RedisConnectionFactory}.
-	 *
-	 * @param connectionFactory {@link RedisConnectionFactory} used by the {@link RedisCacheManager}
-	 * to acquire connections to Redis when performing {@link RedisCache} operations; must not be {@literal null}.
-	 * @return new {@link RedisCacheManagerBuilder}.
-	 * @throws IllegalArgumentException if the given {@link RedisConnectionFactory} is {@literal null}.
-	 * @see org.springframework.data.redis.connection.RedisConnectionFactory
-	 */
-	public static RedisCacheManagerBuilder builder(RedisConnectionFactory connectionFactory) {
-
-		Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
-
-		return RedisCacheManagerBuilder.fromConnectionFactory(connectionFactory);
-	}
-
-	/**
-	 * Factory method used to construct a new {@link RedisCacheManager} initialized with
-	 * the given {@link RedisConnectionFactory} and using {@link RedisCacheConfiguration#defaultCacheConfig() defaults}
-	 * for caching.
-	 *
-	 * <dl>
-	 * <dt>locking</dt>
-	 * <dd>disabled</dd>
-	 * <dt>batch strategy</dt>
-	 * <dd>{@link BatchStrategies#keys()}</dd>
-	 * <dt>cache configuration</dt>
-	 * <dd>{@link RedisCacheConfiguration#defaultCacheConfig()}</dd>
-	 * <dt>initial caches</dt>
-	 * <dd>none</dd>
-	 * <dt>transaction aware</dt>
-	 * <dd>no</dd>
-	 * <dt>in-flight cache creation</dt>
-	 * <dd>enabled</dd>
-	 * </dl>
-	 *
-	 * @param connectionFactory {@link RedisConnectionFactory} used by the {@link RedisCacheManager}
-	 * to acquire connections to Redis when performing {@link RedisCache} operations; must not be {@literal null}.
-	 * @return new {@link RedisCacheManager}.
-	 * @throws IllegalArgumentException if the given {@link RedisConnectionFactory} is {@literal null}.
-	 * @see org.springframework.data.redis.connection.RedisConnectionFactory
-	 */
-	public static RedisCacheManager create(RedisConnectionFactory connectionFactory) {
-
-		Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
-
-		return new RedisCacheManager(org.springframework.data.redis.cache.RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory),
-			RedisCacheConfiguration.defaultCacheConfig());
-	}
-
 	private final boolean allowRuntimeCacheCreation;
 
 	private final RedisCacheConfiguration defaultCacheConfiguration;
@@ -145,17 +66,17 @@ public static RedisCacheManager create(RedisConnectionFactory connectionFactory)
 	private final Map<String, RedisCacheConfiguration> initialCacheConfiguration;
 
 	/**
-	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter}
-	 * and a default {@link RedisCacheConfiguration}.
+	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter} and a default
+	 * {@link RedisCacheConfiguration}.
 	 * <p>
 	 * Allows {@link RedisCache cache} creation at runtime.
 	 *
-	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations
-	 * by executing appropriate Redis commands; must not be {@literal null}.
-	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches}
-	 * by default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
+	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing appropriate
+	 *          Redis commands; must not be {@literal null}.
+	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches} by
+	 *          default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
 	 * @throws IllegalArgumentException if either the given {@link RedisCacheWriter} or {@link RedisCacheConfiguration}
-	 * are {@literal null}.
+	 *           are {@literal null}.
 	 * @see org.springframework.data.redis.cache.RedisCacheConfiguration
 	 * @see org.springframework.data.redis.cache.RedisCacheWriter
 	 */
@@ -164,17 +85,17 @@ public RedisCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration d
 	}
 
 	/**
-	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter}
-	 * and default {@link RedisCacheConfiguration}, and whether to allow cache creation at runtime.
+	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter} and default
+	 * {@link RedisCacheConfiguration}, and whether to allow cache creation at runtime.
 	 *
-	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations
-	 * by executing appropriate Redis commands; must not be {@literal null}.
-	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches}
-	 * by default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
+	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing appropriate
+	 *          Redis commands; must not be {@literal null}.
+	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches} by
+	 *          default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
 	 * @param allowRuntimeCacheCreation boolean specifying whether to allow creation of undeclared caches at runtime;
-	 * {@literal true} by default. Maybe just use {@link RedisCacheConfiguration#defaultCacheConfig()}.
+	 *          {@literal true} by default. Maybe just use {@link RedisCacheConfiguration#defaultCacheConfig()}.
 	 * @throws IllegalArgumentException if either the given {@link RedisCacheWriter} or {@link RedisCacheConfiguration}
-	 * are {@literal null}.
+	 *           are {@literal null}.
 	 * @see org.springframework.data.redis.cache.RedisCacheConfiguration
 	 * @see org.springframework.data.redis.cache.RedisCacheWriter
 	 * @since 2.0.4
@@ -191,20 +112,20 @@ private RedisCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration
 	}
 
 	/**
-	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter}
-	 * and a default {@link RedisCacheConfiguration}, along with an optional, initial set of {@link String cache names}
-	 * used to create {@link RedisCache Redis caches} on startup.
+	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter} and a default
+	 * {@link RedisCacheConfiguration}, along with an optional, initial set of {@link String cache names} used to create
+	 * {@link RedisCache Redis caches} on startup.
 	 * <p>
 	 * Allows {@link RedisCache cache} creation at runtime.
 	 *
-	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations
-	 * by executing appropriate Redis commands; must not be {@literal null}.
-	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches}
-	 * by default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
+	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing appropriate
+	 *          Redis commands; must not be {@literal null}.
+	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches} by
+	 *          default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
 	 * @param initialCacheNames optional set of {@link String cache names} used to create {@link RedisCache Redis caches}
-	 * on startup. The default {@link RedisCacheConfiguration} will be applied to each cache.
+	 *          on startup. The default {@link RedisCacheConfiguration} will be applied to each cache.
 	 * @throws IllegalArgumentException if either the given {@link RedisCacheWriter} or {@link RedisCacheConfiguration}
-	 * are {@literal null}.
+	 *           are {@literal null}.
 	 * @see org.springframework.data.redis.cache.RedisCacheConfiguration
 	 * @see org.springframework.data.redis.cache.RedisCacheWriter
 	 */
@@ -215,22 +136,22 @@ public RedisCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration d
 	}
 
 	/**
-	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter}
-	 * and default {@link RedisCacheConfiguration}, and whether to allow cache creation at runtime.
+	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter} and default
+	 * {@link RedisCacheConfiguration}, and whether to allow cache creation at runtime.
 	 * <p>
-	 * Additionally, the optional, initial set of {@link String cache names} will be used to create
-	 * {@link RedisCache Redis caches} on startup.
+	 * Additionally, the optional, initial set of {@link String cache names} will be used to create {@link RedisCache
+	 * Redis caches} on startup.
 	 *
-	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations
-	 * by executing appropriate Redis commands; must not be {@literal null}.
-	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches}
-	 * by default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
+	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing appropriate
+	 *          Redis commands; must not be {@literal null}.
+	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches} by
+	 *          default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
 	 * @param allowRuntimeCacheCreation boolean specifying whether to allow creation of undeclared caches at runtime;
-	 * {@literal true} by default. Maybe just use {@link RedisCacheConfiguration#defaultCacheConfig()}.
+	 *          {@literal true} by default. Maybe just use {@link RedisCacheConfiguration#defaultCacheConfig()}.
 	 * @param initialCacheNames optional set of {@link String cache names} used to create {@link RedisCache Redis caches}
-	 * on startup. The default {@link RedisCacheConfiguration} will be applied to each cache.
+	 *          on startup. The default {@link RedisCacheConfiguration} will be applied to each cache.
 	 * @throws IllegalArgumentException if either the given {@link RedisCacheWriter} or {@link RedisCacheConfiguration}
-	 * are {@literal null}.
+	 *           are {@literal null}.
 	 * @see org.springframework.data.redis.cache.RedisCacheConfiguration
 	 * @see org.springframework.data.redis.cache.RedisCacheWriter
 	 * @since 2.0.4
@@ -254,15 +175,15 @@ public RedisCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration d
 	 * <p>
 	 * Allows {@link RedisCache cache} creation at runtime.
 	 *
-	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations
-	 * by executing appropriate Redis commands; must not be {@literal null}.
-	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches}
-	 * by default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
+	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing appropriate
+	 *          Redis commands; must not be {@literal null}.
+	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches} by
+	 *          default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
 	 * @param initialCacheConfigurations {@link Map} of declared, known {@link String cache names} along with associated
-	 * {@link RedisCacheConfiguration} used to create and configure {@link RedisCache Reds caches} on startup;
-	 * must not be {@literal null}.
+	 *          {@link RedisCacheConfiguration} used to create and configure {@link RedisCache Reds caches} on startup;
+	 *          must not be {@literal null}.
 	 * @throws IllegalArgumentException if either the given {@link RedisCacheWriter} or {@link RedisCacheConfiguration}
-	 * are {@literal null}.
+	 *           are {@literal null}.
 	 * @see org.springframework.data.redis.cache.RedisCacheConfiguration
 	 * @see org.springframework.data.redis.cache.RedisCacheWriter
 	 */
@@ -273,23 +194,23 @@ public RedisCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration d
 	}
 
 	/**
-	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter}
-	 * and a default {@link RedisCacheConfiguration}, and whether to allow {@link RedisCache} creation at runtime.
+	 * Creates a new {@link RedisCacheManager} initialized with the given {@link RedisCacheWriter} and a default
+	 * {@link RedisCacheConfiguration}, and whether to allow {@link RedisCache} creation at runtime.
 	 * <p>
 	 * Additionally, an initial {@link RedisCache} will be created and configured using the associated
 	 * {@link RedisCacheConfiguration} for each {@link String named} {@link RedisCache} in the given {@link Map}.
 	 *
-	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations
-	 * by executing appropriate Redis commands; must not be {@literal null}.
-	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches}
-	 * by default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
+	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing appropriate
+	 *          Redis commands; must not be {@literal null}.
+	 * @param defaultCacheConfiguration {@link RedisCacheConfiguration} applied to new {@link RedisCache Redis caches} by
+	 *          default when no cache-specific {@link RedisCacheConfiguration} is provided; must not be {@literal null}.
 	 * @param allowRuntimeCacheCreation boolean specifying whether to allow creation of undeclared caches at runtime;
-	 * {@literal true} by default. Maybe just use {@link RedisCacheConfiguration#defaultCacheConfig()}.
-	 * @param initialCacheConfigurations {@link Map} of declared, known {@link String cache names} along with
-	 * the associated {@link RedisCacheConfiguration} used to create and configure {@link RedisCache Redis caches}
-	 * on startup; must not be {@literal null}.
+	 *          {@literal true} by default. Maybe just use {@link RedisCacheConfiguration#defaultCacheConfig()}.
+	 * @param initialCacheConfigurations {@link Map} of declared, known {@link String cache names} along with the
+	 *          associated {@link RedisCacheConfiguration} used to create and configure {@link RedisCache Redis caches} on
+	 *          startup; must not be {@literal null}.
 	 * @throws IllegalArgumentException if either the given {@link RedisCacheWriter} or {@link RedisCacheConfiguration}
-	 * are {@literal null}.
+	 *           are {@literal null}.
 	 * @see org.springframework.data.redis.cache.RedisCacheConfiguration
 	 * @see org.springframework.data.redis.cache.RedisCacheWriter
 	 * @since 2.0.4
@@ -305,16 +226,93 @@ public RedisCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration d
 	}
 
 	/**
-	 * @deprecated use {@link org.springframework.data.redis.cache.RedisCacheManager#RedisCacheManager(RedisCacheWriter, RedisCacheConfiguration, boolean, Map)}
-	 * instead.
+	 * @deprecated since 3.2. Use
+	 *             {@link RedisCacheManager#RedisCacheManager(RedisCacheWriter, RedisCacheConfiguration, boolean, Map)}
+	 *             instead.
 	 */
-	@Deprecated
+	@Deprecated(since = "3.2")
 	public RedisCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration defaultCacheConfiguration,
 			Map<String, RedisCacheConfiguration> initialCacheConfigurations, boolean allowRuntimeCacheCreation) {
-
 		this(cacheWriter, defaultCacheConfiguration, allowRuntimeCacheCreation, initialCacheConfigurations);
 	}
 
+	/**
+	 * Factory method returning a {@literal Builder} used to construct and configure a {@link RedisCacheManager}.
+	 *
+	 * @return new {@link RedisCacheManagerBuilder}.
+	 * @since 2.3
+	 */
+	public static RedisCacheManagerBuilder builder() {
+		return new RedisCacheManagerBuilder();
+	}
+
+	/**
+	 * Factory method returning a {@literal Builder} used to construct and configure a {@link RedisCacheManager}
+	 * initialized with the given {@link RedisCacheWriter}.
+	 *
+	 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing appropriate
+	 *          Redis commands; must not be {@literal null}.
+	 * @return new {@link RedisCacheManagerBuilder}.
+	 * @throws IllegalArgumentException if the given {@link RedisCacheWriter} is {@literal null}.
+	 * @see org.springframework.data.redis.cache.RedisCacheWriter
+	 */
+	public static RedisCacheManagerBuilder builder(RedisCacheWriter cacheWriter) {
+
+		Assert.notNull(cacheWriter, "CacheWriter must not be null");
+
+		return RedisCacheManagerBuilder.fromCacheWriter(cacheWriter);
+	}
+
+	/**
+	 * Factory method returning a {@literal Builder} used to construct and configure a {@link RedisCacheManager}
+	 * initialized with the given {@link RedisConnectionFactory}.
+	 *
+	 * @param connectionFactory {@link RedisConnectionFactory} used by the {@link RedisCacheManager} to acquire
+	 *          connections to Redis when performing {@link RedisCache} operations; must not be {@literal null}.
+	 * @return new {@link RedisCacheManagerBuilder}.
+	 * @throws IllegalArgumentException if the given {@link RedisConnectionFactory} is {@literal null}.
+	 * @see org.springframework.data.redis.connection.RedisConnectionFactory
+	 */
+	public static RedisCacheManagerBuilder builder(RedisConnectionFactory connectionFactory) {
+
+		Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
+
+		return RedisCacheManagerBuilder.fromConnectionFactory(connectionFactory);
+	}
+
+	/**
+	 * Factory method used to construct a new {@link RedisCacheManager} initialized with the given
+	 * {@link RedisConnectionFactory} and using {@link RedisCacheConfiguration#defaultCacheConfig() defaults} for caching.
+	 * <dl>
+	 * <dt>locking</dt>
+	 * <dd>disabled</dd>
+	 * <dt>batch strategy</dt>
+	 * <dd>{@link BatchStrategies#keys()}</dd>
+	 * <dt>cache configuration</dt>
+	 * <dd>{@link RedisCacheConfiguration#defaultCacheConfig()}</dd>
+	 * <dt>initial caches</dt>
+	 * <dd>none</dd>
+	 * <dt>transaction aware</dt>
+	 * <dd>no</dd>
+	 * <dt>in-flight cache creation</dt>
+	 * <dd>enabled</dd>
+	 * </dl>
+	 *
+	 * @param connectionFactory {@link RedisConnectionFactory} used by the {@link RedisCacheManager} to acquire
+	 *          connections to Redis when performing {@link RedisCache} operations; must not be {@literal null}.
+	 * @return new {@link RedisCacheManager}.
+	 * @throws IllegalArgumentException if the given {@link RedisConnectionFactory} is {@literal null}.
+	 * @see org.springframework.data.redis.connection.RedisConnectionFactory
+	 */
+	public static RedisCacheManager create(RedisConnectionFactory connectionFactory) {
+
+		Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
+
+		return new RedisCacheManager(
+				org.springframework.data.redis.cache.RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory),
+				RedisCacheConfiguration.defaultCacheConfig());
+	}
+
 	/**
 	 * Determines whether {@link RedisCache Redis caches} are allowed to be created at runtime.
 	 *
@@ -325,11 +323,11 @@ public boolean isAllowRuntimeCacheCreation() {
 	}
 
 	/**
-	 * Return an {@link Collections#unmodifiableMap(Map) unmodifiable Map} containing {@link String caches name}
-	 * mapped to the {@link RedisCache} {@link RedisCacheConfiguration configuration}.
+	 * Return an {@link Collections#unmodifiableMap(Map) unmodifiable Map} containing {@link String caches name} mapped to
+	 * the {@link RedisCache} {@link RedisCacheConfiguration configuration}.
 	 *
-	 * @return unmodifiable {@link Map} containing {@link String cache name}
-	 * / {@link RedisCacheConfiguration configuration} pairs.
+	 * @return unmodifiable {@link Map} containing {@link String cache name} / {@link RedisCacheConfiguration
+	 *         configuration} pairs.
 	 */
 	public Map<String, RedisCacheConfiguration> getCacheConfigurations() {
 
@@ -345,8 +343,8 @@ public Map<String, RedisCacheConfiguration> getCacheConfigurations() {
 	}
 
 	/**
-	 * Gets the default {@link RedisCacheConfiguration} applied to new {@link RedisCache} instances on creation
-	 * when custom, non-specific {@link RedisCacheConfiguration} was not provided.
+	 * Gets the default {@link RedisCacheConfiguration} applied to new {@link RedisCache} instances on creation when
+	 * custom, non-specific {@link RedisCacheConfiguration} was not provided.
 	 *
 	 * @return the default {@link RedisCacheConfiguration}.
 	 */
@@ -355,8 +353,8 @@ protected RedisCacheConfiguration getDefaultCacheConfiguration() {
 	}
 
 	/**
-	 * Gets a {@link Map} of {@link String cache names} to {@link RedisCacheConfiguration} objects as the initial set
-	 * of {@link RedisCache Redis caches} to create on startup.
+	 * Gets a {@link Map} of {@link String cache names} to {@link RedisCacheConfiguration} objects as the initial set of
+	 * {@link RedisCache Redis caches} to create on startup.
 	 *
 	 * @return a {@link Map} of {@link String cache names} to {@link RedisCacheConfiguration} objects.
 	 */
@@ -365,8 +363,8 @@ protected Map<String, RedisCacheConfiguration> getInitialCacheConfiguration() {
 	}
 
 	/**
-	 * Returns a reference to the configured {@link RedisCacheWriter} used to perform {@link RedisCache} operations,
-	 * such as reading from and writing to the cache.
+	 * Returns a reference to the configured {@link RedisCacheWriter} used to perform {@link RedisCache} operations, such
+	 * as reading from and writing to the cache.
 	 *
 	 * @return a reference to the configured {@link RedisCacheWriter}.
 	 * @see org.springframework.data.redis.cache.RedisCacheWriter
@@ -384,8 +382,8 @@ protected RedisCache getMissingCache(String name) {
 	 * Creates a new {@link RedisCache} with given {@link String name} and {@link RedisCacheConfiguration}.
 	 *
 	 * @param name {@link String name} for the {@link RedisCache}; must not be {@literal null}.
-	 * @param cacheConfiguration {@link RedisCacheConfiguration} used to configure the {@link RedisCache};
-	 * resolves to the {@link #getDefaultCacheConfiguration()} if {@literal null}.
+	 * @param cacheConfiguration {@link RedisCacheConfiguration} used to configure the {@link RedisCache}; resolves to the
+	 *          {@link #getDefaultCacheConfiguration()} if {@literal null}.
 	 * @return a new {@link RedisCache} instance; never {@literal null}.
 	 */
 	protected RedisCache createRedisCache(String name, @Nullable RedisCacheConfiguration cacheConfiguration) {
@@ -396,8 +394,7 @@ protected RedisCache createRedisCache(String name, @Nullable RedisCacheConfigura
 	protected Collection<RedisCache> loadCaches() {
 
 		return getInitialCacheConfiguration().entrySet().stream()
-			.map(entry -> createRedisCache(entry.getKey(), entry.getValue()))
-			.toList();
+				.map(entry -> createRedisCache(entry.getKey(), entry.getValue())).toList();
 	}
 
 	private RedisCacheConfiguration resolveCacheConfiguration(@Nullable RedisCacheConfiguration cacheConfiguration) {
@@ -416,26 +413,25 @@ private RedisCacheConfiguration resolveCacheConfiguration(@Nullable RedisCacheCo
 	public static class RedisCacheManagerBuilder {
 
 		/**
-		 * Factory method returning a new {@literal Builder} used to create and configure a {@link RedisCacheManager}
-		 * using the given {@link RedisCacheWriter}.
+		 * Factory method returning a new {@literal Builder} used to create and configure a {@link RedisCacheManager} using
+		 * the given {@link RedisCacheWriter}.
 		 *
 		 * @param cacheWriter {@link RedisCacheWriter} used to perform {@link RedisCache} operations by executing
-		 * appropriate Redis commands; must not be {@literal null}.
+		 *          appropriate Redis commands; must not be {@literal null}.
 		 * @return new {@link RedisCacheManagerBuilder}.
 		 * @throws IllegalArgumentException if the given {@link RedisCacheWriter} is {@literal null}.
 		 * @see org.springframework.data.redis.cache.RedisCacheWriter
 		 */
 		public static RedisCacheManagerBuilder fromCacheWriter(RedisCacheWriter cacheWriter) {
-			return new RedisCacheManagerBuilder(RedisAssertions.requireNonNull(cacheWriter,
-					"CacheWriter must not be null"));
+			return new RedisCacheManagerBuilder(RedisAssertions.requireNonNull(cacheWriter, "CacheWriter must not be null"));
 		}
 
 		/**
-		 * Factory method returning a new {@literal Builder} used to create and configure a {@link RedisCacheManager}
-		 * using the given {@link RedisConnectionFactory}.
+		 * Factory method returning a new {@literal Builder} used to create and configure a {@link RedisCacheManager} using
+		 * the given {@link RedisConnectionFactory}.
 		 *
-		 * @param connectionFactory {@link RedisConnectionFactory} used by the {@link RedisCacheManager}
-		 * to acquire connections to Redis when performing {@link RedisCache} operations; must not be {@literal null}.
+		 * @param connectionFactory {@link RedisConnectionFactory} used by the {@link RedisCacheManager} to acquire
+		 *          connections to Redis when performing {@link RedisCache} operations; must not be {@literal null}.
 		 * @return new {@link RedisCacheManagerBuilder}.
 		 * @throws IllegalArgumentException if the given {@link RedisConnectionFactory} is {@literal null}.
 		 * @see org.springframework.data.redis.connection.RedisConnectionFactory
@@ -468,8 +464,8 @@ private RedisCacheManagerBuilder(RedisCacheWriter cacheWriter) {
 		/**
 		 * Configure whether to allow cache creation at runtime.
 		 *
-		 * @param allowRuntimeCacheCreation boolean to allow creation of undeclared caches at runtime;
-		 * {@literal true} by default.
+		 * @param allowRuntimeCacheCreation boolean to allow creation of undeclared caches at runtime; {@literal true} by
+		 *          default.
 		 * @return this {@link RedisCacheManagerBuilder}.
 		 */
 		public RedisCacheManagerBuilder allowCreateOnMissingCache(boolean allowRuntimeCacheCreation) {
@@ -577,8 +573,8 @@ public RedisCacheManagerBuilder transactionAware() {
 		}
 
 		/**
-		 * Registers the given {@link String cache name} and {@link RedisCacheConfiguration} used to
-		 * create and configure a {@link RedisCache} on startup.
+		 * Registers the given {@link String cache name} and {@link RedisCacheConfiguration} used to create and configure a
+		 * {@link RedisCache} on startup.
 		 *
 		 * @param cacheName {@link String name} of the cache to register for creation on startup.
 		 * @param cacheConfiguration {@link RedisCacheConfiguration} used to configure the new cache on startup.
@@ -644,7 +640,7 @@ public Set<String> getConfiguredCaches() {
 		public RedisCacheManager build() {
 
 			Assert.state(cacheWriter != null, "CacheWriter must not be null;"
-				+ " You can provide one via 'RedisCacheManagerBuilder#cacheWriter(RedisCacheWriter)'");
+					+ " You can provide one via 'RedisCacheManagerBuilder#cacheWriter(RedisCacheWriter)'");
 
 			RedisCacheWriter resolvedCacheWriter = !CacheStatisticsCollector.none().equals(this.statisticsCollector)
 					? this.cacheWriter.withStatisticsCollector(this.statisticsCollector)
diff --git a/src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java b/src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java
index 57e8e5b0f4..f6da33eaf7 100644
--- a/src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java
+++ b/src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java
@@ -88,8 +88,7 @@ static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectio
 	static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory,
 			BatchStrategy batchStrategy) {
 
-		return lockingRedisCacheWriter(connectionFactory, Duration.ofMillis(50), TtlFunction.persistent(),
-				batchStrategy);
+		return lockingRedisCacheWriter(connectionFactory, Duration.ofMillis(50), TtlFunction.persistent(), batchStrategy);
 	}
 
 	/**
@@ -108,8 +107,8 @@ static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectio
 
 		Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
 
-		return new DefaultRedisCacheWriter(connectionFactory, sleepTime, lockTtlFunction,
-				CacheStatisticsCollector.none(), batchStrategy);
+		return new DefaultRedisCacheWriter(connectionFactory, sleepTime, lockTtlFunction, CacheStatisticsCollector.none(),
+				batchStrategy);
 	}
 
 	/**
@@ -124,8 +123,8 @@ static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectio
 	byte[] get(String name, byte[] key);
 
 	/**
-	 * Get the binary value representation from Redis stored for the given key and set the given
-	 * {@link Duration TTL expiration} for the cache entry.
+	 * Get the binary value representation from Redis stored for the given key and set the given {@link Duration TTL
+	 * expiration} for the cache entry.
 	 *
 	 * @param name must not be {@literal null}.
 	 * @param key must not be {@literal null}.
@@ -138,18 +137,19 @@ default byte[] get(String name, byte[] key, @Nullable Duration ttl) {
 	}
 
 	/**
-	 * Determines whether the asynchronous {@link #retrieve(String, byte[])}
-	 * and {@link #retrieve(String, byte[], Duration)} cache operations are supported by the implementation.
+	 * Determines whether the asynchronous {@link #retrieve(String, byte[])} and
+	 * {@link #retrieve(String, byte[], Duration)} cache operations are supported by the implementation.
 	 * <p>
-	 * The main factor for whether the {@literal retrieve} operation can be supported will primarily be determined by
-	 * the Redis driver in use at runtime.
+	 * The main factor for whether the {@literal retrieve} operation can be supported will primarily be determined by the
+	 * Redis driver in use at runtime.
 	 * <p>
-	 * Returns {@literal false} by default. This will have an effect of {@link RedisCache#retrieve(Object)}
-	 * and {@link RedisCache#retrieve(Object, Supplier)} throwing an {@link UnsupportedOperationException}.
+	 * Returns {@literal false} by default. This will have an effect of {@link RedisCache#retrieve(Object)} and
+	 * {@link RedisCache#retrieve(Object, Supplier)} throwing an {@link UnsupportedOperationException}.
 	 *
 	 * @return {@literal true} if asynchronous {@literal retrieve} operations are supported by the implementation.
+	 * @since 3.2
 	 */
-	default boolean isRetrieveSupported() {
+	default boolean supportsAsyncRetrieve() {
 		return false;
 	}
 
@@ -162,7 +162,7 @@ default boolean isRetrieveSupported() {
 	 * @param key {@link byte[] key} mapped to the {@link CompletableFuture value} in the {@link RedisCache}.
 	 * @return the {@link CompletableFuture value} to which the {@link RedisCache} maps the given {@link byte[] key}.
 	 * @see #retrieve(String, byte[], Duration)
-	 * @since 3.2.0
+	 * @since 3.2
 	 */
 	default CompletableFuture<byte[]> retrieve(String name, byte[] key) {
 		return retrieve(name, key, null);
@@ -178,7 +178,7 @@ default CompletableFuture<byte[]> retrieve(String name, byte[] key) {
 	 * @param key {@link byte[] key} mapped to the {@link CompletableFuture value} in the {@link RedisCache}.
 	 * @param ttl {@link Duration} specifying the {@literal expiration timeout} for the cache entry.
 	 * @return the {@link CompletableFuture value} to which the {@link RedisCache} maps the given {@link byte[] key}.
-	 * @since 3.2.0
+	 * @since 3.2
 	 */
 	CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Duration ttl);
 
@@ -192,6 +192,19 @@ default CompletableFuture<byte[]> retrieve(String name, byte[] key) {
 	 */
 	void put(String name, byte[] key, byte[] value, @Nullable Duration ttl);
 
+	/**
+	 * Store the given key/value pair asynchronously to Redis and set the expiration time if defined.
+	 * <p>
+	 * This operation is non-blocking.
+	 *
+	 * @param name The cache name must not be {@literal null}.
+	 * @param key The key for the cache entry. Must not be {@literal null}.
+	 * @param value The value stored for the key. Must not be {@literal null}.
+	 * @param ttl Optional expiration time. Can be {@literal null}.
+	 * @since 3.2
+	 */
+	CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nullable Duration ttl);
+
 	/**
 	 * Write the given value to Redis if the key does not already exist.
 	 *
@@ -272,15 +285,15 @@ static TtlFunction persistent() {
 		/**
 		 * Compute a {@link Duration time-to-live (TTL)} using the cache {@code key} and {@code value}.
 		 * <p>
-		 * The {@link Duration time-to-live (TTL)} is computed on each write operation. Redis uses millisecond
-		 * granularity for timeouts. Any more granular values (e.g. micros or nanos) are not considered
-		 * and will be truncated due to rounding. Returning {@link Duration#ZERO}, or a value less than
-		 * {@code Duration.ofMillis(1)}, results in a persistent value that does not expire.
+		 * The {@link Duration time-to-live (TTL)} is computed on each write operation. Redis uses millisecond granularity
+		 * for timeouts. Any more granular values (e.g. micros or nanos) are not considered and will be truncated due to
+		 * rounding. Returning {@link Duration#ZERO}, or a value less than {@code Duration.ofMillis(1)}, results in a
+		 * persistent value that does not expire.
 		 *
 		 * @param key the cache key.
 		 * @param value the cache value. Can be {@code null} if the cache supports {@code null} value caching.
 		 * @return the computed {@link Duration time-to-live (TTL)}. Can be {@link Duration#ZERO} for persistent values
-		 * (i.e. cache entry does not expire).
+		 *         (i.e. cache entry does not expire).
 		 */
 		Duration getTimeToLive(Object key, @Nullable Object value);
 
diff --git a/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterTests.java b/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterTests.java
index 7ec58b7c03..b2eec5b2de 100644
--- a/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterTests.java
+++ b/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterTests.java
@@ -16,6 +16,7 @@
 package org.springframework.data.redis.cache;
 
 import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.Assumptions.*;
 import static org.springframework.data.redis.cache.RedisCacheWriter.*;
 
 import java.nio.charset.Charset;
@@ -23,6 +24,7 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
@@ -31,10 +33,8 @@
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
-import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
-import org.springframework.data.redis.connection.jedis.extension.JedisConnectionFactoryExtension;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.core.types.Expiration;
-import org.springframework.data.redis.test.extension.RedisStanalone;
 import org.springframework.data.redis.test.extension.parametrized.MethodSource;
 import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest;
 
@@ -67,12 +67,6 @@ public static Collection<Object[]> testParams() {
 
 	@BeforeEach
 	void setUp() {
-
-		JedisConnectionFactory connectionFactory =
-				JedisConnectionFactoryExtension.getConnectionFactory(RedisStanalone.class);
-
-		this.connectionFactory = connectionFactory;
-
 		doWithConnection(RedisConnection::flushAll);
 	}
 
@@ -152,6 +146,49 @@ void getShouldReturnNullWhenKeyDoesNotExist() {
 		assertThat(nonLockingRedisCacheWriter(connectionFactory).get(CACHE_NAME, binaryCacheKey)).isNull();
 	}
 
+	@ParameterizedRedisTest // GH-2650
+	void cacheHitRetrieveShouldIncrementStatistics() throws ExecutionException, InterruptedException {
+
+		assumeThat(connectionFactory).isInstanceOf(LettuceConnectionFactory.class);
+
+		doWithConnection(connection -> connection.set(binaryCacheKey, binaryCacheValue));
+
+		RedisCacheWriter writer = nonLockingRedisCacheWriter(connectionFactory)
+				.withStatisticsCollector(CacheStatisticsCollector.create());
+
+		writer.retrieve(CACHE_NAME, binaryCacheKey).get();
+
+		assertThat(writer.getCacheStatistics(CACHE_NAME).getGets()).isOne();
+		assertThat(writer.getCacheStatistics(CACHE_NAME).getHits()).isOne();
+	}
+
+	@ParameterizedRedisTest // GH-2650
+	void storeShouldIncrementStatistics() throws ExecutionException, InterruptedException {
+
+		assumeThat(connectionFactory).isInstanceOf(LettuceConnectionFactory.class);
+
+		RedisCacheWriter writer = nonLockingRedisCacheWriter(connectionFactory)
+				.withStatisticsCollector(CacheStatisticsCollector.create());
+
+		writer.store(CACHE_NAME, binaryCacheKey, binaryCacheValue, null).get();
+
+		assertThat(writer.getCacheStatistics(CACHE_NAME).getPuts()).isOne();
+	}
+
+	@ParameterizedRedisTest // GH-2650
+	void cacheMissRetrieveWithLoaderAsyncShouldIncrementStatistics() throws ExecutionException, InterruptedException {
+
+		assumeThat(connectionFactory).isInstanceOf(LettuceConnectionFactory.class);
+
+		RedisCacheWriter writer = nonLockingRedisCacheWriter(connectionFactory)
+				.withStatisticsCollector(CacheStatisticsCollector.create());
+
+		writer.retrieve(CACHE_NAME, binaryCacheKey).get();
+
+		assertThat(writer.getCacheStatistics(CACHE_NAME).getGets()).isOne();
+		assertThat(writer.getCacheStatistics(CACHE_NAME).getMisses()).isOne();
+	}
+
 	@ParameterizedRedisTest // DATAREDIS-481, DATAREDIS-1082
 	void putIfAbsentShouldAddEternalEntryWhenKeyDoesNotExist() {
 
@@ -253,8 +290,8 @@ void lockingCacheWriterShouldIgnoreExistingLockOnDifferenceCache() {
 
 		((DefaultRedisCacheWriter) lockingRedisCacheWriter(connectionFactory)).lock(CACHE_NAME);
 
-		lockingRedisCacheWriter(connectionFactory).put(CACHE_NAME + "-no-the-other-cache", binaryCacheKey,
-				binaryCacheValue, Duration.ZERO);
+		lockingRedisCacheWriter(connectionFactory).put(CACHE_NAME + "-no-the-other-cache", binaryCacheKey, binaryCacheValue,
+				Duration.ZERO);
 
 		doWithConnection(connection -> {
 			assertThat(connection.exists(binaryCacheKey)).isTrue();
@@ -341,8 +378,7 @@ boolean doCheckLock(String name, RedisConnection connection) {
 
 		afterWrite.await();
 
-		assertThat(exceptionRef.get()).hasMessageContaining("Interrupted while waiting to unlock")
-				.hasCauseInstanceOf(InterruptedException.class);
+		assertThat(exceptionRef.get()).hasRootCauseInstanceOf(InterruptedException.class);
 	}
 
 	@ParameterizedRedisTest // GH-2300
diff --git a/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterUnitTests.java b/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterUnitTests.java
deleted file mode 100644
index 46a41433d6..0000000000
--- a/src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterUnitTests.java
+++ /dev/null
@@ -1,324 +0,0 @@
-/*
- * Copyright 2023 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.cache;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoInteractions;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-
-import java.nio.ByteBuffer;
-import java.time.Duration;
-import java.util.concurrent.CompletableFuture;
-
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-import org.springframework.data.redis.connection.ReactiveRedisConnection;
-import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
-import org.springframework.data.redis.connection.ReactiveStringCommands;
-import org.springframework.data.redis.connection.RedisConnection;
-import org.springframework.data.redis.connection.RedisConnectionFactory;
-import org.springframework.data.redis.connection.RedisKeyCommands;
-import org.springframework.data.redis.connection.RedisStringCommands;
-import org.springframework.data.redis.core.types.Expiration;
-
-import reactor.core.publisher.Mono;
-
-/**
- * Unit tests for {@link DefaultRedisCacheWriter}
- *
- * @author John Blum
- */
-@ExtendWith(MockitoExtension.class)
-class DefaultRedisCacheWriterUnitTests {
-
-	@Mock
-	private CacheStatisticsCollector mockCacheStatisticsCollector = mock(CacheStatisticsCollector.class);
-
-	@Mock
-	private RedisConnection mockConnection;
-
-	@Mock(strictness = Mock.Strictness.LENIENT)
-	private RedisConnectionFactory mockConnectionFactory;
-
-	@Mock
-	private ReactiveRedisConnection mockReactiveConnection;
-
-	@Mock(strictness = Mock.Strictness.LENIENT)
-	private TestReactiveRedisConnectionFactory mockReactiveConnectionFactory;
-
-	@BeforeEach
-	void setup() {
-		doReturn(this.mockConnection).when(this.mockConnectionFactory).getConnection();
-		doReturn(this.mockConnection).when(this.mockReactiveConnectionFactory).getConnection();
-		doReturn(this.mockReactiveConnection).when(this.mockReactiveConnectionFactory).getReactiveConnection();
-	}
-
-	private RedisCacheWriter newRedisCacheWriter() {
-		return new DefaultRedisCacheWriter(this.mockConnectionFactory, mock(BatchStrategy.class))
-				.withStatisticsCollector(this.mockCacheStatisticsCollector);
-	}
-
-	private RedisCacheWriter newReactiveRedisCacheWriter() {
-		return newReactiveRedisCacheWriter(Duration.ZERO);
-	}
-
-	private RedisCacheWriter newReactiveRedisCacheWriter(Duration sleepTime) {
-		return new DefaultRedisCacheWriter(this.mockReactiveConnectionFactory, sleepTime, mock(BatchStrategy.class))
-				.withStatisticsCollector(this.mockCacheStatisticsCollector);
-	}
-
-	@Test // GH-2351
-	void getWithNonNullTtl() {
-
-		byte[] key = "TestKey".getBytes();
-		byte[] value = "TestValue".getBytes();
-
-		Duration ttl = Duration.ofSeconds(15);
-		Expiration expiration = Expiration.from(ttl);
-
-		RedisStringCommands mockStringCommands = mock(RedisStringCommands.class);
-
-		doReturn(mockStringCommands).when(this.mockConnection).stringCommands();
-		doReturn(value).when(mockStringCommands).getEx(any(), any());
-
-		RedisCacheWriter cacheWriter = newRedisCacheWriter();
-
-		assertThat(cacheWriter.get("TestCache", key, ttl)).isEqualTo(value);
-
-		verify(this.mockConnection, times(1)).stringCommands();
-		verify(mockStringCommands, times(1)).getEx(eq(key), eq(expiration));
-		verify(this.mockConnection).close();
-		verifyNoMoreInteractions(this.mockConnection, mockStringCommands);
-	}
-
-	@Test // GH-2351
-	void getWithNullTtl() {
-
-		byte[] key = "TestKey".getBytes();
-		byte[] value = "TestValue".getBytes();
-
-		RedisStringCommands mockStringCommands = mock(RedisStringCommands.class);
-
-		doReturn(mockStringCommands).when(this.mockConnection).stringCommands();
-		doReturn(value).when(mockStringCommands).get(any());
-
-		RedisCacheWriter cacheWriter = newRedisCacheWriter();
-
-		assertThat(cacheWriter.get("TestCache", key, null)).isEqualTo(value);
-
-		verify(this.mockConnection, times(1)).stringCommands();
-		verify(mockStringCommands, times(1)).get(eq(key));
-		verify(this.mockConnection).close();
-		verifyNoMoreInteractions(this.mockConnection, mockStringCommands);
-	}
-
-	@Test // GH-2650
-	@SuppressWarnings("all")
-	void retrieveWithNoCacheName() {
-
-		byte[] key = "TestKey".getBytes();
-
-		RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter();
-
-		assertThatIllegalArgumentException()
-			.isThrownBy(() -> cacheWriter.retrieve(null, key))
-			.withMessage("Name must not be null")
-			.withNoCause();
-
-		verifyNoInteractions(this.mockReactiveConnectionFactory);
-	}
-
-	@Test // GH-2650
-	@SuppressWarnings("all")
-	void retrieveWithNoKey() {
-
-		RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter();
-
-		assertThatIllegalArgumentException()
-			.isThrownBy(() -> cacheWriter.retrieve("TestCacheName", null))
-			.withMessage("Key must not be null")
-			.withNoCause();
-
-		verifyNoInteractions(this.mockReactiveConnectionFactory);
-	}
-
-	@Test // GH-2650
-	void retrieveReturnsAsyncFutureWithValue() throws Exception {
-
-		byte[] key = "TestKey".getBytes();
-
-		RedisStringCommands mockStringCommands = mock(RedisStringCommands.class);
-
-		doReturn(mockStringCommands).when(this.mockConnection).stringCommands();
-		doReturn("test".getBytes()).when(mockStringCommands).get(any(byte[].class));
-
-		RedisCacheWriter cacheWriter = newRedisCacheWriter();
-
-		CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key);
-
-		assertThat(result).isNotNull();
-		verifyNoInteractions(this.mockCacheStatisticsCollector);
-
-		byte[] value = result.get();
-
-		assertThat(value).isNotNull();
-		assertThat(new String(value)).isEqualTo("test");
-
-		verify(mockStringCommands, times(1)).get(eq(key));
-		verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
-		verify(this.mockCacheStatisticsCollector, times(1)).incHits(eq("TestCacheName"));
-		verifyNoMoreInteractions(mockStringCommands, this.mockCacheStatisticsCollector);
-	}
-
-	@Test // GH-2650
-	void retrieveWithExpirationReturnsAsyncFutureWithValue() throws Exception {
-
-		byte[] key = "TestKey".getBytes();
-
-		Duration thirtySeconds = Duration.ofSeconds(30L);
-
-		RedisStringCommands mockStringCommands = mock(RedisStringCommands.class);
-
-		doReturn(mockStringCommands).when(this.mockConnection).stringCommands();
-		doReturn("test".getBytes()).when(mockStringCommands).getEx(any(byte[].class), any(Expiration.class));
-
-		RedisCacheWriter cacheWriter = newRedisCacheWriter();
-
-		CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key, thirtySeconds);
-
-		assertThat(result).isNotNull();
-
-		byte[] value = result.get();
-
-		assertThat(value).isNotNull();
-		assertThat(new String(value)).isEqualTo("test");
-
-		verify(mockStringCommands, times(1)).getEx(eq(key), eq(Expiration.from(thirtySeconds)));
-		verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
-		verify(this.mockCacheStatisticsCollector, times(1)).incHits(eq("TestCacheName"));
-		verifyNoMoreInteractions(mockStringCommands, this.mockCacheStatisticsCollector);
-	}
-
-	@Test // GH-2650
-	void retrieveReturnsReactiveFutureWithValue() throws Exception {
-
-		byte[] key = "TestKey".getBytes();
-
-		Duration sixtySeconds = Duration.ofMillis(60L);
-
-		RedisKeyCommands mockKeyCommands = mock(RedisKeyCommands.class);
-		ReactiveStringCommands mockStringCommands = mock(ReactiveStringCommands.class);
-
-		doReturn(mockKeyCommands).when(this.mockConnection).keyCommands();
-		doReturn(false).when(mockKeyCommands).exists(any(byte[].class));
-		doReturn(mockStringCommands).when(this.mockReactiveConnection).stringCommands();
-		doReturn(Mono.just(ByteBuffer.wrap("test".getBytes()))).when(mockStringCommands).get(any(ByteBuffer.class));
-
-		RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter(sixtySeconds);
-
-		CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key);
-
-		assertThat(result).isNotNull();
-
-		byte[] value = result.get();
-
-		assertThat(value).isNotNull();
-		assertThat(new String(value)).isEqualTo("test");
-
-		verify(mockKeyCommands, times(1)).exists(any(byte[].class));
-		verify(mockStringCommands, times(1)).get(eq(ByteBuffer.wrap(key)));
-		verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
-		verify(this.mockCacheStatisticsCollector, times(1)).incHits(eq("TestCacheName"));
-		verifyNoMoreInteractions(mockKeyCommands, mockStringCommands, this.mockCacheStatisticsCollector);
-	}
-
-	@Test // GH-2650
-	void retrieveReturnsReactiveFutureWithNoValue() throws Exception {
-
-		byte[] key = "TestKey".getBytes();
-
-		RedisKeyCommands mockKeyCommands = mock(RedisKeyCommands.class);
-		ReactiveStringCommands mockStringCommands = mock(ReactiveStringCommands.class);
-
-		doReturn(mockKeyCommands).when(this.mockConnection).keyCommands();
-		doReturn(false).when(mockKeyCommands).exists(any(byte[].class));
-		doReturn(mockStringCommands).when(this.mockReactiveConnection).stringCommands();
-		doReturn(Mono.empty()).when(mockStringCommands).get(any(ByteBuffer.class));
-
-		RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter();
-
-		CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key);
-
-		assertThat(result).isNotNull();
-
-		byte[] value = result.get();
-
-		assertThat(value).isNull();
-
-		verify(mockKeyCommands, times(1)).exists(any(byte[].class));
-		verify(mockStringCommands, times(1)).get(eq(ByteBuffer.wrap(key)));
-		verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
-		verify(this.mockCacheStatisticsCollector, times(1)).incMisses(eq("TestCacheName"));
-		verifyNoMoreInteractions(mockKeyCommands, mockStringCommands, this.mockCacheStatisticsCollector);
-	}
-
-	@Test // GH-2650
-	void retrieveWithExpirationReturnsReactiveFutureWithValue() throws Exception {
-
-		byte[] key = "TestKey".getBytes();
-
-		Duration twoMinutes = Duration.ofMinutes(2L);
-
-		RedisKeyCommands mockKeyCommands = mock(RedisKeyCommands.class);
-		ReactiveStringCommands mockStringCommands = mock(ReactiveStringCommands.class);
-
-		doReturn(mockKeyCommands).when(this.mockConnection).keyCommands();
-		doReturn(false).when(mockKeyCommands).exists(any(byte[].class));
-		doReturn(mockStringCommands).when(this.mockReactiveConnection).stringCommands();
-		doReturn(Mono.just(ByteBuffer.wrap("test".getBytes()))).when(mockStringCommands).getEx(any(ByteBuffer.class), any());
-
-		RedisCacheWriter cacheWriter = newReactiveRedisCacheWriter();
-
-		CompletableFuture<byte[]> result = cacheWriter.retrieve("TestCacheName", key, twoMinutes);
-
-		assertThat(result).isNotNull();
-
-		byte[] value = result.get();
-
-		assertThat(value).isNotNull();
-		assertThat(new String(value)).isEqualTo("test");
-
-		verify(mockKeyCommands, times(1)).exists(any(byte[].class));
-		verify(mockStringCommands, times(1)).getEx(eq(ByteBuffer.wrap(key)), eq(Expiration.from(twoMinutes)));
-		verify(this.mockCacheStatisticsCollector, times(1)).incGets(eq("TestCacheName"));
-		verify(this.mockCacheStatisticsCollector, times(1)).incHits(eq("TestCacheName"));
-		verifyNoMoreInteractions(mockKeyCommands, mockStringCommands, this.mockCacheStatisticsCollector);
-	}
-
-	interface TestReactiveRedisConnectionFactory extends ReactiveRedisConnectionFactory, RedisConnectionFactory { }
-
-}
diff --git a/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java b/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java
index 722a123f13..a927ee457a 100644
--- a/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java
+++ b/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java
@@ -15,18 +15,16 @@
  */
 package org.springframework.data.redis.cache;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
-import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
-import static org.assertj.core.api.Assumptions.assumeThat;
-import static org.awaitility.Awaitility.await;
+import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.Assumptions.*;
+import static org.awaitility.Awaitility.*;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
 
 import java.io.Serializable;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
-import java.time.LocalDateTime;
-import java.time.Month;
-import java.time.ZoneOffset;
+import java.time.Instant;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
@@ -45,7 +43,6 @@
 import java.util.stream.IntStream;
 
 import org.junit.jupiter.api.BeforeEach;
-
 import org.springframework.cache.Cache.ValueWrapper;
 import org.springframework.cache.interceptor.SimpleKey;
 import org.springframework.cache.interceptor.SimpleKeyGenerator;
@@ -57,14 +54,10 @@
 import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.test.condition.EnabledOnCommand;
-import org.springframework.data.redis.test.condition.EnabledOnRedisDriver;
-import org.springframework.data.redis.test.condition.RedisDriver;
 import org.springframework.data.redis.test.extension.parametrized.MethodSource;
 import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest;
 import org.springframework.lang.Nullable;
 
-import io.netty.util.concurrent.DefaultThreadFactory;
-
 /**
  * Tests for {@link RedisCache} with {@link DefaultRedisCacheWriter} using different {@link RedisSerializer} and
  * {@link RedisConnectionFactory} pairs.
@@ -353,8 +346,9 @@ void computePrefixCreatesCacheKeyCorrectly() {
 
 		cacheWithCustomPrefix.put("key-1", sample);
 
-		doWithConnection(connection -> assertThat(connection.stringCommands()
-				.get("_cache_key-1".getBytes(StandardCharsets.UTF_8))).isEqualTo(binarySample));
+		doWithConnection(
+				connection -> assertThat(connection.stringCommands().get("_cache_key-1".getBytes(StandardCharsets.UTF_8)))
+						.isEqualTo(binarySample));
 	}
 
 	@ParameterizedRedisTest // DATAREDIS-1041
@@ -366,8 +360,9 @@ void prefixCacheNameCreatesCacheKeyCorrectly() {
 
 		cacheWithCustomPrefix.put("key-1", sample);
 
-		doWithConnection(connection -> assertThat(connection.stringCommands()
-				.get("redis::cache::key-1".getBytes(StandardCharsets.UTF_8))).isEqualTo(binarySample));
+		doWithConnection(connection -> assertThat(
+				connection.stringCommands().get("redis::cache::key-1".getBytes(StandardCharsets.UTF_8)))
+						.isEqualTo(binarySample));
 	}
 
 	@ParameterizedRedisTest // DATAREDIS-715
@@ -429,8 +424,8 @@ void cacheShouldAllowMapCacheKeys() {
 				.generateKey(Collections.singletonMap("map-key", new ComplexKey(sample.getFirstname(), sample.getBirthdate())));
 		cache.put(key, sample);
 
-		ValueWrapper target = cache.get(SimpleKeyGenerator
-				.generateKey(Collections.singletonMap("map-key", new ComplexKey(sample.getFirstname(), sample.getBirthdate()))));
+		ValueWrapper target = cache.get(SimpleKeyGenerator.generateKey(
+				Collections.singletonMap("map-key", new ComplexKey(sample.getFirstname(), sample.getBirthdate()))));
 
 		assertThat(target.get()).isEqualTo(sample);
 	}
@@ -481,6 +476,11 @@ public CompletableFuture<byte[]> retrieve(String name, byte[] key, @Nullable Dur
 				return CompletableFuture.completedFuture(value);
 			}
 
+			@Override
+			public CompletableFuture<Void> store(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
+				return null;
+			}
+
 			@Override
 			public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
 				storage.set(value);
@@ -579,21 +579,11 @@ void retrieveCacheValueUsingJedis() {
 		assumeThat(this.connectionFactory instanceof JedisConnectionFactory).isTrue();
 
 		assertThatExceptionOfType(UnsupportedOperationException.class)
-			.isThrownBy(() -> this.cache.retrieve(this.binaryCacheKey))
-			.withMessageContaining(RedisCache.class.getName())
-			.withNoCause();
-	}
-
-	@ParameterizedRedisTest
-	void retrieveCacheValueWithLoaderUsingJedis() {
-
-		// TODO: Is there a better way to do this? @EnableOnRedisDriver(RedisDriver.JEDIS) does not work!
-		assumeThat(this.connectionFactory instanceof JedisConnectionFactory).isTrue();
+				.isThrownBy(() -> this.cache.retrieve(this.binaryCacheKey)).withMessageContaining("RedisCache");
 
 		assertThatExceptionOfType(UnsupportedOperationException.class)
-			.isThrownBy(() -> this.cache.retrieve(this.binaryCacheKey, () -> CompletableFuture.completedFuture("TEST")))
-			.withMessageContaining(RedisCache.class.getName())
-			.withNoCause();
+				.isThrownBy(() -> this.cache.retrieve(this.binaryCacheKey, () -> CompletableFuture.completedFuture("TEST")))
+				.withMessageContaining("RedisCache");
 	}
 
 	@ParameterizedRedisTest // GH-2650
@@ -631,19 +621,13 @@ void retrieveReturnsCachedValueWhenLockIsReleased() throws Exception {
 		RedisCache cache = new RedisCache("cache", usingLockingRedisCacheWriter(Duration.ofMillis(5L)),
 				usingRedisCacheConfiguration());
 
-		RedisCacheWriter cacheWriter = cache.getCacheWriter();
-
-		assertThat(cacheWriter).isInstanceOf(DefaultRedisCacheWriter.class);
-
-		((DefaultRedisCacheWriter) cacheWriter).lock("cache");
+		DefaultRedisCacheWriter cacheWriter = (DefaultRedisCacheWriter) cache.getCacheWriter();
+		cacheWriter.lock("cache");
 
 		CompletableFuture<String> value = (CompletableFuture<String>) cache.retrieve(this.key);
-
-		assertThat(value).isNotNull();
-		assertThat(value.getNow(mockValue)).isEqualTo(mockValue);
 		assertThat(value).isNotDone();
 
-		((DefaultRedisCacheWriter) cacheWriter).unlock("cache");
+		cacheWriter.unlock("cache");
 
 		assertThat(value.get(15L, TimeUnit.MILLISECONDS)).isEqualTo(testValue);
 		assertThat(value).isDone();
@@ -656,14 +640,8 @@ void retrieveReturnsLoadedValue() throws Exception {
 		assumeThat(this.connectionFactory instanceof LettuceConnectionFactory).isTrue();
 
 		RedisCache cache = new RedisCache("cache", usingLockingRedisCacheWriter(), usingRedisCacheConfiguration());
-
 		AtomicBoolean loaded = new AtomicBoolean(false);
-
-		Date birthdate = Date.from(LocalDateTime.of(2023, Month.SEPTEMBER, 22, 17, 3)
-				.toInstant(ZoneOffset.UTC));
-
-		Person jon = new Person("Jon", birthdate);
-
+		Person jon = new Person("Jon", Date.from(Instant.now()));
 		CompletableFuture<Person> valueLoader = CompletableFuture.completedFuture(jon);
 
 		Supplier<CompletableFuture<Person>> valueLoaderSupplier = () -> {
@@ -673,13 +651,29 @@ void retrieveReturnsLoadedValue() throws Exception {
 
 		CompletableFuture<Person> value = cache.retrieve(this.key, valueLoaderSupplier);
 
-		assertThat(value).isNotNull();
 		assertThat(loaded.get()).isFalse();
 		assertThat(value.get()).isEqualTo(jon);
 		assertThat(loaded.get()).isTrue();
 		assertThat(value).isDone();
 	}
 
+	@ParameterizedRedisTest // GH-2650
+	void retrieveStoresLoadedValue() throws Exception {
+
+		// TODO: Is there a better way to do this? @EnableOnRedisDriver(RedisDriver.LETTUCE) does not work!
+		assumeThat(this.connectionFactory instanceof LettuceConnectionFactory).isTrue();
+
+		RedisCache cache = new RedisCache("cache", usingLockingRedisCacheWriter(), usingRedisCacheConfiguration());
+		Person jon = new Person("Jon", Date.from(Instant.now()));
+		Supplier<CompletableFuture<Person>> valueLoaderSupplier = () -> CompletableFuture.completedFuture(jon);
+
+		cache.retrieve(this.key, valueLoaderSupplier).get();
+
+		doWithConnection(
+				connection -> assertThat(connection.keyCommands().exists("cache::key-1".getBytes(StandardCharsets.UTF_8)))
+						.isTrue());
+	}
+
 	@ParameterizedRedisTest // GH-2650
 	void retrieveReturnsNull() throws Exception {
 
@@ -732,8 +726,8 @@ private Object unwrap(@Nullable Object value) {
 
 	private Function<RedisCacheConfiguration, RedisCacheConfiguration> withTtiExpiration() {
 
-		Function<RedisCacheConfiguration, RedisCacheConfiguration> entryTtlFunction =
-			cacheConfiguration -> cacheConfiguration.entryTtl(Duration.ofMillis(100));
+		Function<RedisCacheConfiguration, RedisCacheConfiguration> entryTtlFunction = cacheConfiguration -> cacheConfiguration
+				.entryTtl(Duration.ofMillis(100));
 
 		return entryTtlFunction.andThen(RedisCacheConfiguration::enableTimeToIdle);
 	}
@@ -752,7 +746,7 @@ static class Person implements Serializable {
 		private String firstname;
 		private Date birthdate;
 
-		public Person() { }
+		public Person() {}
 
 		public Person(String firstname, Date birthdate) {
 			this.firstname = firstname;
@@ -787,7 +781,7 @@ public boolean equals(Object obj) {
 			}
 
 			return Objects.equals(this.getFirstname(), that.getFirstname())
-				&& Objects.equals(this.getBirthdate(), that.getBirthdate());
+					&& Objects.equals(this.getBirthdate(), that.getBirthdate());
 		}
 
 		@Override
@@ -797,8 +791,7 @@ public int hashCode() {
 
 		@Override
 		public String toString() {
-			return "RedisCacheTests.Person(firstname=" + this.getFirstname()
-				+ ", birthdate=" + this.getBirthdate() + ")";
+			return "RedisCacheTests.Person(firstname=" + this.getFirstname() + ", birthdate=" + this.getBirthdate() + ")";
 		}
 	}
 
@@ -844,7 +837,7 @@ public boolean equals(final Object obj) {
 			}
 
 			return Objects.equals(this.getFirstname(), that.getFirstname())
-				&& Objects.equals(this.getBirthdate(), that.getBirthdate());
+					&& Objects.equals(this.getBirthdate(), that.getBirthdate());
 		}
 
 		@Override
@@ -854,8 +847,7 @@ public int hashCode() {
 
 		@Override
 		public String toString() {
-			return "RedisCacheTests.ComplexKey(firstame=" + this.getFirstname()
-				+ ", birthdate=" + this.getBirthdate() + ")";
+			return "RedisCacheTests.ComplexKey(firstame=" + this.getFirstname() + ", birthdate=" + this.getBirthdate() + ")";
 		}
 	}
 }
diff --git a/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java b/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java
index 70d1ce5c48..17663734a3 100644
--- a/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java
@@ -15,29 +15,20 @@
  */
 package org.springframework.data.redis.cache;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.ArgumentMatchers.isA;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoInteractions;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
 
 import java.util.concurrent.CompletableFuture;
 
 import org.junit.jupiter.api.Test;
-
-import org.springframework.cache.support.NullValue;
-import org.springframework.data.redis.util.ByteUtils;
+import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
 
 /**
  * Unit tests for {@link RedisCache}.
  *
  * @author John Blum
+ * @author Mark Paluch
  */
 class RedisCacheUnitTests {
 
@@ -46,76 +37,21 @@ void cacheRetrieveValueCallsCacheWriterRetrieveCorrectly() throws Exception {
 
 		RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
 
-		doReturn(CompletableFuture.completedFuture("TEST".getBytes()))
-				.when(mockCacheWriter).retrieve(anyString(), any(byte[].class));
+		when(mockCacheWriter.supportsAsyncRetrieve()).thenReturn(true);
+		when(mockCacheWriter.retrieve(anyString(), any(byte[].class)))
+				.thenReturn(CompletableFuture.completedFuture("TEST".getBytes()));
 
 		RedisCache cache = new RedisCache("TestCache", mockCacheWriter,
-				RedisCacheConfiguration.defaultCacheConfig());
+				RedisCacheConfiguration.defaultCacheConfig().serializeValuesWith(SerializationPair.byteArray()));
 
-		CompletableFuture<byte[]> value = cache.retrieveValue("TestKey");
+		CompletableFuture<byte[]> value = (CompletableFuture<byte[]>) cache.retrieve("TestKey");
 
 		assertThat(value).isNotNull();
 		assertThat(new String(value.get())).isEqualTo("TEST");
 
 		verify(mockCacheWriter, times(1)).retrieve(eq("TestCache"), isA(byte[].class));
+		verify(mockCacheWriter).supportsAsyncRetrieve();
 		verifyNoMoreInteractions(mockCacheWriter);
 	}
 
-	@Test // GH-2650
-	void nullSafeDeserializedStoreValueWithNullValueIsNullSafe() {
-
-		RedisCacheConfiguration cacheConfiguration = RedisCacheConfiguration.defaultCacheConfig();
-		RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
-		RedisCache cache = new RedisCache("TestCache", mockCacheWriter, cacheConfiguration);
-
-		assertThat(cache.nullSafeDeserializedStoreValue(null)).isNull();
-
-		verifyNoInteractions(mockCacheWriter);
-	}
-
-	@Test // GH-2650
-	void nullSafeDeserializedStoreValueWithBinaryNullValueAllowingNullValues() {
-
-		RedisCacheConfiguration cacheConfiguration = RedisCacheConfiguration.defaultCacheConfig();
-		RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
-		RedisCache cache = new RedisCache("TestCache", mockCacheWriter, cacheConfiguration);
-
-		assertThat(cacheConfiguration.getAllowCacheNullValues()).isTrue();
-		assertThat(cache.nullSafeDeserializedStoreValue(RedisCache.BINARY_NULL_VALUE)).isNull();
-
-		verifyNoInteractions(mockCacheWriter);
-	}
-
-	@Test // GH-2650
-	void nullSafeDeserializedStoreValueWithBinaryNullValueDisablingNullValues() {
-
-		RedisCacheConfiguration cacheConfiguration =
-				RedisCacheConfiguration.defaultCacheConfig().disableCachingNullValues();
-
-		RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
-
-		RedisCache cache = new RedisCache("TestCache", mockCacheWriter, cacheConfiguration);
-
-		assertThat(cacheConfiguration.getAllowCacheNullValues()).isFalse();
-		assertThat(cache.nullSafeDeserializedStoreValue(RedisCache.BINARY_NULL_VALUE)).isEqualTo(NullValue.INSTANCE);
-
-		verifyNoInteractions(mockCacheWriter);
-	}
-
-	@Test // GH-2650
-	void nullSafeDeserializedStoreValueWithNonNullValue() {
-
-		RedisCacheConfiguration cacheConfiguration = RedisCacheConfiguration.defaultCacheConfig();
-
-		byte[] serializedValue = ByteUtils.getBytes(cacheConfiguration.getValueSerializationPair()
-				.write("TestValue"));
-
-		RedisCacheWriter mockCacheWriter = mock(RedisCacheWriter.class);
-
-		RedisCache cache = new RedisCache("TestCache", mockCacheWriter, cacheConfiguration);
-
-		assertThat(cache.nullSafeDeserializedStoreValue(serializedValue)).isEqualTo("TestValue");
-
-		verifyNoInteractions(mockCacheWriter);
-	}
 }