Revise RedisCache and documentation

src/main/asciidoc/reference/redis-cache.adoc

@@ -26,7 +26,7 @@ RedisCacheManager cm = RedisCacheManager.builder(connectionFactory)
 
 As shown in the preceding example, `RedisCacheManager` allows definition of configurations on a per-cache basis.
 
-The behavior of `RedisCache` created with `RedisCacheManager` is defined with `RedisCacheConfiguration`. The configuration lets you set key expiration times, prefixes, and ``RedisSerializer`` implementations for converting to and from the binary storage format, as shown in the following example:
+The behavior of `RedisCache` created with `RedisCacheManager` is defined with `RedisCacheConfiguration`. The configuration lets you set key expiration times, prefixes, and `RedisSerializer` implementations for converting to and from the binary storage format, as shown in the following example:
 
 [source,java]
 ----
@@ -70,7 +70,7 @@ The cache implementation defaults to use `KEYS` and `DEL` to clear the cache. `K
 
 [source,java]
 ----
-RedisCacheManager cm = RedisCacheManager.build(RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory, BatchStrategy.scan(1000)))
+RedisCacheManager cm = RedisCacheManager.build(RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory, BatchStrategies.scan(1000)))
 	.cacheDefaults(defaultCacheConfig())
 	...
 ----

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

@@ -0,0 +1,158 @@
+/*
+ * Copyright 2021 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 java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.util.Assert;
+
+/**
+ * A collection of predefined {@link BatchStrategy} implementations using {@code KEYS} or {@code SCAN} command.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 2.6
+ */
+public abstract class BatchStrategies {
+
+	/**
+	 * A {@link BatchStrategy} using a single {@code KEYS} and {@code DEL} command to remove all matching keys.
+	 * {@code KEYS} scans the entire keyspace of the Redis database and can block the Redis worker thread for a long time
+	 * on large keyspaces.
+	 * <p/>
+	 * {@code KEYS} is supported for standalone and clustered (sharded) Redis operation modes.
+	 *
+	 * @return batching strategy using {@code KEYS}.
+	 */
+	public static BatchStrategy keys() {
+		return Keys.INSTANCE;
+	}
+
+	/**
+	 * A {@link BatchStrategy} using a {@code SCAN} cursors and potentially multiple {@code DEL} commands to remove all
+	 * matching keys. This strategy allows a configurable batch size to optimize for scan batching.
+	 * <p/>
+	 * Note that using the {@code SCAN} strategy might be not supported on all drivers and Redis operation modes.
+	 *
+	 * @return batching strategy using {@code SCAN}.
+	 */
+	public static BatchStrategy scan(int batchSize) {
+
+		Assert.isTrue(batchSize > 0, "Batch size must be greater than zero!");
+
+		return new Scan(batchSize);
+	}
+
+	/**
+	 * {@link BatchStrategy} using {@code KEYS}.
+	 */
+	static class Keys implements BatchStrategy {
+
+		static Keys INSTANCE = new Keys();
+
+		@Override
+		public long cleanCache(RedisConnection connection, String name, byte[] pattern) {
+
+			byte[][] keys = Optional.ofNullable(connection.keys(pattern)).orElse(Collections.emptySet())
+					.toArray(new byte[0][]);
+
+			if (keys.length > 0) {
+				connection.del(keys);
+			}
+
+			return keys.length;
+		}
+	}
+
+	/**
+	 * {@link BatchStrategy} using {@code SCAN}.
+	 */
+	static class Scan implements BatchStrategy {
+
+		private final int batchSize;
+
+		Scan(int batchSize) {
+			this.batchSize = batchSize;
+		}
+
+		@Override
+		public long cleanCache(RedisConnection connection, String name, byte[] pattern) {
+
+			Cursor<byte[]> cursor = connection.scan(ScanOptions.scanOptions().count(batchSize).match(pattern).build());
+
+			long count = 0;
+
+			PartitionIterator<byte[]> partitions = new PartitionIterator<>(cursor, batchSize);
+			while (partitions.hasNext()) {
+
+				List<byte[]> keys = partitions.next();
+				count += keys.size();
+
+				if (keys.size() > 0) {
+					connection.del(keys.toArray(new byte[0][]));
+				}
+			}
+
+			return count;
+		}
+	}
+
+	/**
+	 * Utility to split and buffer outcome from a {@link Iterator} into {@link List lists} of {@code T} with a maximum
+	 * chunks {@code size}.
+	 *
+	 * @param <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() {
+			return iterator.hasNext();
+		}
+
+		@Override
+		public List<T> next() {
+
+			if (!hasNext()) {
+				throw new NoSuchElementException();
+			}
+
+			List<T> list = new ArrayList<>(size);
+			while (list.size() < size && iterator.hasNext()) {
+				list.add(iterator.next());
+			}
+
+			return list;
+		}
+	}
+}

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

@@ -15,156 +15,30 @@
  */
 package org.springframework.data.redis.cache;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.Optional;
-
 import org.springframework.data.redis.connection.RedisConnection;
-import org.springframework.data.redis.core.Cursor;
-import org.springframework.data.redis.core.ScanOptions;
-import org.springframework.util.Assert;
 
 /**
- * Batch strategies to be used with {@link RedisCacheWriter}.
+ * A {@link BatchStrategy} to be used with {@link RedisCacheWriter}.
+ * <p/>
+ * Mainly used to clear the cache.
  * <p/>
- * Primarily used to clear the cache.
+ * Predefined strategies using the {@link BatchStrategies#keys() KEYS} or {@link BatchStrategies#scan(int) SCAN}
+ * commands can be found in {@link BatchStrategies}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.6
  */
-public abstract class BatchStrategy {
-
-	/**
-	 * Batching strategy using a single {@code KEYS} and {@code DEL} command to remove all matching keys. {@code KEYS}
-	 * scans the entire keyspace of the Redis database and can block the Redis worker thread for a long time on large
-	 * keyspaces.
-	 * <p/>
-	 * {@code KEYS} is supported for standalone and clustered (sharded) Redis operation modes.
-	 *
-	 * @return batching strategy using {@code KEYS}.
-	 */
-	public static BatchStrategy keys() {
-		return Keys.INSTANCE;
-	}
-
-	/**
-	 * Batching strategy using a {@code SCAN} cursors and potentially multiple {@code DEL} commands to remove all matching
-	 * keys. This strategy allows a configurable batch size to optimize for scan batching.
-	 * <p/>
-	 * Note that using the {@code SCAN} strategy might be not supported on all drivers and Redis operation modes.
-	 *
-	 * @return batching strategy using {@code SCAN}.
-	 */
-	public static BatchStrategy scan(int batchSize) {
-
-		Assert.isTrue(batchSize > 0, "Batch size must be greater than zero");
-
-		return new Scan(batchSize);
-	}
+public interface BatchStrategy {
 
 	/**
 	 * Remove all keys following the given pattern.
 	 *
-	 * @param the connection to use.
-	 * @param name The cache name must not be {@literal null}.
+	 * @param connection the connection to use. Must not be {@literal null}.
+	 * @param name The cache name. Must not be {@literal null}.
 	 * @param pattern The pattern for the keys to remove. Must not be {@literal null}.
 	 * @return number of removed keys.
 	 */
-	abstract int cleanCache(RedisConnection connection, String name, byte[] pattern);
-
-	/**
-	 * {@link BatchStrategy} using {@code KEYS}.
-	 */
-	static class Keys extends BatchStrategy {
-
-		static Keys INSTANCE = new Keys();
-
-		@Override
-		int cleanCache(RedisConnection connection, String name, byte[] pattern) {
-
-			byte[][] keys = Optional.ofNullable(connection.keys(pattern)).orElse(Collections.emptySet())
-					.toArray(new byte[0][]);
-
-			if (keys.length > 0) {
-				connection.del(keys);
-			}
-
-			return keys.length;
-		}
-	}
-
-	/**
-	 * {@link BatchStrategy} using {@code SCAN}.
-	 */
-	static class Scan extends BatchStrategy {
-
-		private final int batchSize;
-
-		public Scan(int batchSize) {
-			this.batchSize = batchSize;
-		}
-
-		@Override
-		int cleanCache(RedisConnection connection, String name, byte[] pattern) {
-
-			Cursor<byte[]> cursor = connection.scan(ScanOptions.scanOptions().count(batchSize).match(pattern).build());
-
-			PartitionIterator<byte[]> partitions = new PartitionIterator<>(cursor, batchSize);
-
-			int count = 0;
-
-			while (partitions.hasNext()) {
-
-				List<byte[]> keys = partitions.next();
-				count += keys.size();
-
-				if (keys.size() > 0) {
-					connection.del(keys.toArray(new byte[0][]));
-				}
-			}
-
-			return count;
-		}
-	}
-
-	/**
-	 * Utility to split and buffer outcome from a {@link Iterator} into {@link List lists} of {@code T} with a maximum
-	 * chunks {@code size}.
-	 *
-	 * @param <T>
-	 */
-	static class PartitionIterator<T> implements Iterator<List<T>> {
-
-		private final Iterator<T> iterator;
-		private final int size;
-
-		public PartitionIterator(Iterator<T> iterator, int size) {
-			this.iterator = iterator;
-			this.size = size;
-		}
-
-		@Override
-		public boolean hasNext() {
-			return iterator.hasNext();
-		}
-
-		@Override
-		public List<T> next() {
-
-			if (!hasNext()) {
-				throw new NoSuchElementException();
-			}
-
-			List<T> list = new ArrayList<>(size);
-			while (list.size() < size && iterator.hasNext()) {
-				list.add(iterator.next());
-			}
-
-			return list;
-		}
-	}
+	long cleanCache(RedisConnection connection, String name, byte[] pattern);
 
 }

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

@@ -217,8 +217,12 @@ public void clean(String name, byte[] pattern) {
 					wasLocked = true;
 				}
 
-
-				statistics.incDeletesBy(name, batchStrategy.cleanCache(connection, name, pattern));
+				long deleteCount = batchStrategy.cleanCache(connection, name, pattern);
+				while (deleteCount > Integer.MAX_VALUE) {
+					statistics.incDeletesBy(name, Integer.MAX_VALUE);
+					deleteCount -= Integer.MAX_VALUE;
+				}
+				statistics.incDeletesBy(name, (int) deleteCount);
 
 			} finally {
 

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

@@ -43,7 +43,7 @@ public interface RedisCacheWriter extends CacheStatisticsProvider {
 	 * @return new instance of {@link DefaultRedisCacheWriter}.
 	 */
 	static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
-		return nonLockingRedisCacheWriter(connectionFactory, BatchStrategy.keys());
+		return nonLockingRedisCacheWriter(connectionFactory, BatchStrategies.keys());
 	}
 
 	/**
@@ -70,7 +70,7 @@ static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connec
 	 * @return new instance of {@link DefaultRedisCacheWriter}.
 	 */
 	static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
-		return lockingRedisCacheWriter(connectionFactory, BatchStrategy.keys());
+		return lockingRedisCacheWriter(connectionFactory, BatchStrategies.keys());
 	}
 
 	/**

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

@@ -307,7 +307,7 @@ void lockingCacheWriterShouldExitWhenInterruptedWaitForLockRelease() throws Inte
 		Thread th = new Thread(() -> {
 
 			DefaultRedisCacheWriter writer = new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50),
-					BatchStrategy.keys()) {
+					BatchStrategies.keys()) {
 
 				@Override
 				boolean doCheckLock(String name, RedisConnection connection) {

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

@@ -262,7 +262,7 @@ void clearWithScanShouldClearCache() {
 		}
 
 		RedisCache cache = new RedisCache("cache",
-				RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory, BatchStrategy.scan(25)),
+				RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory, BatchStrategies.scan(25)),
 				RedisCacheConfiguration.defaultCacheConfig().serializeValuesWith(SerializationPair.fromSerializer(serializer)));
 
 		doWithConnection(connection -> {