Align method names of RedisCacheWriter with RedisCache.

Also undo change type to record which is leaking the underlying Iterator and add more tests.

Original Pull Request: #3243

Author: christophstrobl

src/main/java/org/springframework/data/redis/cache/BatchStrategies.java

@@ -132,7 +132,16 @@ public long cleanCache(RedisConnection connection, String name, byte[] pattern)
 	 *
 	 * @param <T>
 	 */
-	record PartitionIterator<T>(Iterator<T> iterator, int size) implements Iterator<List<T>> {
+	static class PartitionIterator<T> implements Iterator<List<T>> {
+
+		private final Iterator<T> iterator;
+		private final int size;
+
+		PartitionIterator(Iterator<T> iterator, int size) {
+
+			this.iterator = iterator;
+			this.size = size;
+		}
 
 		@Override
 		public boolean hasNext() {

src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java

@@ -416,20 +416,20 @@ public byte[] putIfAbsent(String name, byte[] key, byte[] value, @Nullable Durat
 	}
 
 	@Override
-	public void remove(String name, byte[] key) {
+	public void evict(String name, byte[] key) {
 
 		Assert.notNull(name, "Name must not be null");
 		Assert.notNull(key, "Key must not be null");
 
 		if (writeAsynchronously()) {
 			asyncCacheWriter.remove(name, key).thenRun(() -> statistics.incDeletes(name));
 		} else {
-			removeIfPresent(name, key);
+			evictIfPresent(name, key);
 		}
 	}
 
 	@Override
-	public boolean removeIfPresent(String name, byte[] key) {
+	public boolean evictIfPresent(String name, byte[] key) {
 
 		Long removals = execute(name, connection -> connection.keyCommands().del(key));
 		statistics.incDeletes(name);
@@ -438,13 +438,13 @@ public boolean removeIfPresent(String name, byte[] key) {
 	}
 
 	@Override
-	public void clean(String name, byte[] pattern) {
+	public void clear(String name, byte[] pattern) {
 
 		Assert.notNull(name, "Name must not be null");
 		Assert.notNull(pattern, "Pattern must not be null");
 
 		if (writeAsynchronously()) {
-			asyncCacheWriter.clean(name, pattern, batchStrategy)
+			asyncCacheWriter.clear(name, pattern, batchStrategy)
 					.thenAccept(deleteCount -> statistics.incDeletesBy(name, deleteCount.intValue()));
 			return;
 		}
@@ -651,8 +651,9 @@ interface AsyncCacheWriter {
 		 * @param pattern {@link String pattern} used to match Redis keys to clear.
 		 * @param batchStrategy strategy to use.
 		 * @return a future that signals completion emitting the number of removed keys.
+		 * @since 4.0
 		 */
-		CompletableFuture<Long> clean(String name, byte[] pattern, BatchStrategy batchStrategy);
+		CompletableFuture<Long> clear(String name, byte[] pattern, BatchStrategy batchStrategy);
 
 	}
 
@@ -686,7 +687,7 @@ public CompletableFuture<Void> remove(String name, byte[] key) {
 		}
 
 		@Override
-		public CompletableFuture<Long> clean(String name, byte[] pattern, BatchStrategy batchStrategy) {
+		public CompletableFuture<Long> clear(String name, byte[] pattern, BatchStrategy batchStrategy) {
 			throw new UnsupportedOperationException("async clean not supported");
 		}
 
@@ -701,10 +702,10 @@ public CompletableFuture<Long> clean(String name, byte[] pattern, BatchStrategy
 	class AsynchronousCacheWriterDelegate implements AsyncCacheWriter {
 
 		private static final int DEFAULT_SCAN_BATCH_SIZE = 64;
-		private final int cleanBatchSize;
+		private final int clearBatchSize;
 
 		public AsynchronousCacheWriterDelegate() {
-			this.cleanBatchSize = batchStrategy instanceof BatchStrategies.Scan scan ? scan.batchSize()
+			this.clearBatchSize = batchStrategy instanceof BatchStrategies.Scan scan ? scan.batchSize()
 					: DEFAULT_SCAN_BATCH_SIZE;
 		}
 
@@ -765,14 +766,14 @@ public CompletableFuture<Void> remove(String name, byte[] key) {
 		}
 
 		@Override
-		public CompletableFuture<Long> clean(String name, byte[] pattern, BatchStrategy batchStrategy) {
+		public CompletableFuture<Long> clear(String name, byte[] pattern, BatchStrategy batchStrategy) {
 
 			return doWithConnection(connection -> {
-				return doWithLocking(name, pattern, null, connection, () -> doClean(pattern, connection));
+				return doWithLocking(name, pattern, null, connection, () -> doClear(pattern, connection));
 			});
 		}
 
-		private Mono<Long> doClean(byte[] pattern, ReactiveRedisConnection connection) {
+		private Mono<Long> doClear(byte[] pattern, ReactiveRedisConnection connection) {
 
 			ReactiveKeyCommands commands = connection.keyCommands();
 
@@ -781,11 +782,11 @@ private Mono<Long> doClean(byte[] pattern, ReactiveRedisConnection connection) {
 			if (batchStrategy instanceof BatchStrategies.Keys) {
 				keys = commands.keys(ByteBuffer.wrap(pattern)).flatMapMany(Flux::fromIterable);
 			} else {
-				keys = commands.scan(ScanOptions.scanOptions().count(cleanBatchSize).match(pattern).build());
+				keys = commands.scan(ScanOptions.scanOptions().count(clearBatchSize).match(pattern).build());
 			}
 
 			return keys
-					.buffer(cleanBatchSize) //
+					.buffer(clearBatchSize) //
 					.flatMap(commands::mUnlink) //
 					.collect(Collectors.summingLong(Long::longValue));
 		}

src/main/java/org/springframework/data/redis/cache/RedisCache.java

@@ -265,12 +265,12 @@ public void clearStatistics() {
 
 	@Override
 	public void evict(Object key) {
-		getCacheWriter().remove(getName(), createAndConvertCacheKey(key));
+		getCacheWriter().evict(getName(), createAndConvertCacheKey(key));
 	}
 
 	@Override
 	public boolean evictIfPresent(Object key) {
-		return getCacheWriter().removeIfPresent(getName(), createAndConvertCacheKey(key));
+		return getCacheWriter().evictIfPresent(getName(), createAndConvertCacheKey(key));
 	}
 
 	@Override

src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java

@@ -260,8 +260,24 @@ default CompletableFuture<byte[]> retrieve(String name, byte[] key) {
 	 *
 	 * @param name cache name must not be {@literal null}.
 	 * @param key key for the cache entry. Must not be {@literal null}.
+	 * @deprecated since 4.0 in favor of {@link #evict(String, byte[])}
 	 */
-	void remove(String name, byte[] key);
+	@Deprecated(since = "4.0", forRemoval = true)
+	default void remove(String name, byte[] key) {
+		evict(name, key);
+	}
+
+	/**
+	 * Remove the given key from Redis.
+	 * <p>
+	 * Actual eviction may be performed in an asynchronous or deferred fashion, with subsequent lookups possibly still
+	 * seeing the entry.
+	 *
+	 * @param name cache name must not be {@literal null}.
+	 * @param key key for the cache entry. Must not be {@literal null}.
+	 * @since 4.0
+	 */
+	void evict(String name, byte[] key);
 
 	/**
 	 * Remove the given key from Redis if it is present, expecting the key to be immediately invisible for subsequent
@@ -271,9 +287,10 @@ default CompletableFuture<byte[]> retrieve(String name, byte[] key) {
 	 * @param key key for the cache entry. Must not be {@literal null}.
 	 * @return {@code true} if the cache was known to have a mapping for this key before, {@code false} if it did not (or
 	 *         if prior presence could not be determined).
+	 * @since 4.0
 	 */
-	default boolean removeIfPresent(String name, byte[] key) {
-		remove(name, key);
+	default boolean evictIfPresent(String name, byte[] key) {
+		evict(name, key);
 		return false;
 	}
 
@@ -285,8 +302,24 @@ default boolean removeIfPresent(String name, byte[] key) {
 	 *
 	 * @param name cache name must not be {@literal null}.
 	 * @param pattern pattern for the keys to remove. Must not be {@literal null}.
+	 * @deprecated since 4.0 in favor of {@link #clear(String, byte[])}
+	 */
+	@Deprecated(since = "4.0", forRemoval = true)
+	default void clean(String name, byte[] pattern) {
+		clear(name, pattern);
+	}
+
+	/**
+	 * Remove all keys following the given pattern.
+	 * <p>
+	 * Actual clearing may be performed in an asynchronous or deferred fashion, with subsequent lookups possibly still
+	 * seeing the entries.
+	 *
+	 * @param name cache name must not be {@literal null}.
+	 * @param pattern pattern for the keys to remove. Must not be {@literal null}.
+	 * @since 4.0
 	 */
-	void clean(String name, byte[] pattern);
+	void clear(String name, byte[] pattern);
 
 	/**
 	 * Remove all keys following the given pattern expecting all entries to be immediately invisible for subsequent
@@ -298,7 +331,7 @@ default boolean removeIfPresent(String name, byte[] key) {
 	 *         presence of entries could not be determined).
 	 */
 	default boolean invalidate(String name, byte[] pattern) {
-		clean(name, pattern);
+		clear(name, pattern);
 		return false;
 	}
 
@@ -379,7 +412,7 @@ default RedisCacheWriterConfigurer enableLocking(Consumer<CacheLockingConfigurat
 
 		/**
 		 * Use immediate writes (i.e. write operations such as
-		 * {@link RedisCacheWriter#put(String, byte[], byte[], Duration)} or {@link #clean(String, byte[])}) shall apply
+		 * {@link RedisCacheWriter#put(String, byte[], byte[], Duration)} or {@link #clear(String, byte[])}) shall apply
 		 * immediately.
 		 * <p>
 		 * Several {@link org.springframework.cache.Cache} operations can be performed asynchronously or deferred and this
@@ -395,7 +428,7 @@ default RedisCacheWriterConfigurer immediateWrites() {
 
 		/**
 		 * Configure whether to use immediate writes (i.e. write operations such as
-		 * {@link RedisCacheWriter#put(String, byte[], byte[], Duration)} or {@link #clean(String, byte[])}) shall apply
+		 * {@link RedisCacheWriter#put(String, byte[], byte[], Duration)} or {@link #clear(String, byte[])}) shall apply
 		 * immediately.
 		 * <p>
 		 * Several {@link org.springframework.cache.Cache} operations can be performed asynchronously or deferred and this

src/test/java/org/springframework/data/redis/cache/DefaultRedisCachWriterUnitTests.java

@@ -19,15 +19,22 @@
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
+
+import reactor.core.publisher.Mono;
+
+import java.nio.ByteBuffer;
 import java.time.Duration;
 
 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.mockito.quality.Strictness;
 import org.springframework.dao.PessimisticLockingFailureException;
+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;
@@ -127,4 +134,62 @@ void mustNotUnlockWhenLockingFails() {
 
 		verify(mockKeyCommands, never()).del(any());
 	}
+
+	@Test // GH-3236
+	void usesAsyncPutIfPossible() {
+
+		byte[] key = "TestKey".getBytes();
+		byte[] value = "TestValue".getBytes();
+
+		RedisConnectionFactory connectionFactory = mock(RedisConnectionFactory.class,
+				withSettings().extraInterfaces(ReactiveRedisConnectionFactory.class));
+		ReactiveRedisConnection mockConnection = mock(ReactiveRedisConnection.class);
+		ReactiveStringCommands mockStringCommands = mock(ReactiveStringCommands.class);
+
+		doReturn(mockConnection).when((ReactiveRedisConnectionFactory) connectionFactory).getReactiveConnection();
+		doReturn(mockStringCommands).when(mockConnection).stringCommands();
+		doReturn(Mono.just(value)).when(mockStringCommands).set(any(), any(), any(), any());
+
+		RedisCacheWriter cacheWriter = RedisCacheWriter.create(connectionFactory, cfg -> {
+			cfg.immediateWrites(false);
+		});
+
+		cacheWriter.put("TestCache", key, value, null);
+
+		verify(mockConnection, times(1)).stringCommands();
+		verify(mockStringCommands, times(1)).set(eq(ByteBuffer.wrap(key)), any());
+	}
+
+	@Test // GH-3236
+	void usesBlockingWritesIfConfiguredWithImmediateWritesEnabled() {
+
+		byte[] key = "TestKey".getBytes();
+		byte[] value = "TestValue".getBytes();
+
+		RedisConnectionFactory connectionFactory = mock(RedisConnectionFactory.class,
+				withSettings().strictness(Strictness.LENIENT).extraInterfaces(ReactiveRedisConnectionFactory.class));
+		ReactiveRedisConnection reactiveMockConnection = mock(ReactiveRedisConnection.class,
+				withSettings().strictness(Strictness.LENIENT));
+		ReactiveStringCommands reactiveMockStringCommands = mock(ReactiveStringCommands.class,
+				withSettings().strictness(Strictness.LENIENT));
+
+		doReturn(reactiveMockConnection).when((ReactiveRedisConnectionFactory) connectionFactory).getReactiveConnection();
+		doReturn(reactiveMockStringCommands).when(reactiveMockConnection).stringCommands();
+
+		RedisStringCommands mockStringCommands = mock(RedisStringCommands.class);
+
+		doReturn(mockStringCommands).when(this.mockConnection).stringCommands();
+		doReturn(this.mockConnection).when(connectionFactory).getConnection();
+
+		RedisCacheWriter cacheWriter = RedisCacheWriter.create(connectionFactory, cfg -> {
+			cfg.immediateWrites(true);
+		});
+
+		cacheWriter.put("TestCache", key, value, null);
+
+		verify(this.mockConnection, times(1)).stringCommands();
+		verify(mockStringCommands, times(1)).set(eq(key), any());
+		verify(reactiveMockConnection, never()).stringCommands();
+		verify(reactiveMockStringCommands, never()).set(eq(ByteBuffer.wrap(key)), any());
+	}
 }