Add support for configurable batch strategies using RedisCache.

We now support a configurable BatchStrategy for RedisCache. The implementations consist of KEYS (default) and SCAN. Since SCAN is not supported with Jedis Cluster and SCAN requires a batch size, we default to KEYS.

Closes: #1721
Original Pull Request: #2051

Author: mp911de

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

@@ -0,0 +1,170 @@
+/*
+ * 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;
+
+/**
+ * Batch strategies to be used with {@link RedisCacheWriter}.
+ * <p/>
+ * Primarily used to clear the cache.
+ *
+ * @author Mark Paluch
+ * @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 when the
+	 * keyspace has a significant size.
+	 * <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);
+	}
+
+	/**
+	 * Remove all keys following the given pattern.
+	 *
+	 * @param the connection to use.
+	 * @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;
+		}
+	}
+
+}

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

@@ -17,8 +17,6 @@
 
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
-import java.util.Collections;
-import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -53,39 +51,45 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 	private final RedisConnectionFactory connectionFactory;
 	private final Duration sleepTime;
 	private final CacheStatisticsCollector statistics;
+	private final BatchStrategy batchStrategy;
 
 	/**
 	 * @param connectionFactory must not be {@literal null}.
+	 * @param batchStrategy must not be {@literal null}.
 	 */
-	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory) {
-		this(connectionFactory, Duration.ZERO);
+	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, BatchStrategy batchStrategy) {
+		this(connectionFactory, Duration.ZERO, batchStrategy);
 	}
 
 	/**
 	 * @param connectionFactory must not be {@literal null}.
 	 * @param sleepTime sleep time between lock request attempts. Must not be {@literal null}. Use {@link Duration#ZERO}
 	 *          to disable locking.
+	 * @param batchStrategy must not be {@literal null}.
 	 */
-	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime) {
-		this(connectionFactory, sleepTime, CacheStatisticsCollector.none());
+	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime, BatchStrategy batchStrategy) {
+		this(connectionFactory, sleepTime, CacheStatisticsCollector.none(), batchStrategy);
 	}
 
 	/**
 	 * @param connectionFactory must not be {@literal null}.
 	 * @param sleepTime sleep time between lock request attempts. Must not be {@literal null}. Use {@link Duration#ZERO}
 	 *          to disable locking.
 	 * @param cacheStatisticsCollector must not be {@literal null}.
+	 * @param batchStrategy must not be {@literal null}.
 	 */
 	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime,
-			CacheStatisticsCollector cacheStatisticsCollector) {
+			CacheStatisticsCollector cacheStatisticsCollector, BatchStrategy batchStrategy) {
 
 		Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
 		Assert.notNull(sleepTime, "SleepTime must not be null!");
 		Assert.notNull(cacheStatisticsCollector, "CacheStatisticsCollector must not be null!");
+		Assert.notNull(batchStrategy, "BatchStrategy must not be null!");
 
 		this.connectionFactory = connectionFactory;
 		this.sleepTime = sleepTime;
 		this.statistics = cacheStatisticsCollector;
+		this.batchStrategy = batchStrategy;
 	}
 
 	/*
@@ -213,13 +217,9 @@ public void clean(String name, byte[] pattern) {
 					wasLocked = true;
 				}
 
-				byte[][] keys = Optional.ofNullable(connection.keys(pattern)).orElse(Collections.emptySet())
-						.toArray(new byte[0][]);
 
-				if (keys.length > 0) {
-					statistics.incDeletesBy(name, keys.length);
-					connection.del(keys);
-				}
+				statistics.incDeletesBy(name, batchStrategy.cleanCache(connection, name, pattern));
+
 			} finally {
 
 				if (wasLocked && isLockingCacheWriter()) {
@@ -255,7 +255,7 @@ public void clearStatistics(String name) {
 	 */
 	@Override
 	public RedisCacheWriter withStatisticsCollector(CacheStatisticsCollector cacheStatisticsCollector) {
-		return new DefaultRedisCacheWriter(connectionFactory, sleepTime, cacheStatisticsCollector);
+		return new DefaultRedisCacheWriter(connectionFactory, sleepTime, cacheStatisticsCollector, this.batchStrategy);
 	}
 
 	/**

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

@@ -186,7 +186,7 @@ public static RedisCacheManager create(RedisConnectionFactory connectionFactory)
 
 		Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
 
-		return new RedisCacheManager(new DefaultRedisCacheWriter(connectionFactory),
+		return new RedisCacheManager(RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory),
 				RedisCacheConfiguration.defaultCacheConfig());
 	}
 
@@ -311,7 +311,7 @@ public static RedisCacheManagerBuilder fromConnectionFactory(RedisConnectionFact
 
 			Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
 
-			return new RedisCacheManagerBuilder(new DefaultRedisCacheWriter(connectionFactory));
+			return new RedisCacheManagerBuilder(RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory));
 		}
 
 		/**

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

@@ -26,6 +26,9 @@
  * caching. <br />
  * The {@link RedisCacheWriter} may be shared by multiple cache implementations and is responsible for writing / reading
  * binary data to / from Redis. The implementation honors potential cache lock flags that might be set.
+ * <p>
+ * The default {@link RedisCacheWriter} implementation can be customized with {@link BatchStrategy} to tune performance
+ * behavior.
  *
  * @author Christoph Strobl
  * @author Mark Paluch
@@ -40,10 +43,24 @@ public interface RedisCacheWriter extends CacheStatisticsProvider {
 	 * @return new instance of {@link DefaultRedisCacheWriter}.
 	 */
 	static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
+		return nonLockingRedisCacheWriter(connectionFactory, BatchStrategy.keys());
+	}
+
+	/**
+	 * Create new {@link RedisCacheWriter} without locking behavior.
+	 *
+	 * @param connectionFactory must not be {@literal null}.
+	 * @param batchStrategy must not be {@literal null}.
+	 * @return new instance of {@link DefaultRedisCacheWriter}.
+	 * @since 2.6
+	 */
+	static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connectionFactory,
+			BatchStrategy batchStrategy) {
 
 		Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
+		Assert.notNull(batchStrategy, "BatchStrategy must not be null!");
 
-		return new DefaultRedisCacheWriter(connectionFactory);
+		return new DefaultRedisCacheWriter(connectionFactory, batchStrategy);
 	}
 
 	/**
@@ -53,10 +70,23 @@ static RedisCacheWriter nonLockingRedisCacheWriter(RedisConnectionFactory connec
 	 * @return new instance of {@link DefaultRedisCacheWriter}.
 	 */
 	static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory) {
+		return lockingRedisCacheWriter(connectionFactory, BatchStrategy.keys());
+	}
+
+	/**
+	 * Create new {@link RedisCacheWriter} with locking behavior.
+	 *
+	 * @param connectionFactory must not be {@literal null}.
+	 * @param batchStrategy must not be {@literal null}.
+	 * @return new instance of {@link DefaultRedisCacheWriter}.
+	 * @since 2.6
+	 */
+	static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory,
+			BatchStrategy batchStrategy) {
 
 		Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
 
-		return new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50));
+		return new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50), batchStrategy);
 	}
 
 	/**

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

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

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

@@ -102,7 +102,8 @@ private RedisCache createCache() {
 			cacheConfiguration = cacheConfiguration.disableCachingNullValues();
 		}
 
-		return new RedisCache(CACHE_NAME, new DefaultRedisCacheWriter(connectionFactory), cacheConfiguration);
+		return new RedisCache(CACHE_NAME, RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory),
+				cacheConfiguration);
 	}
 
 	protected Object getValue() {