Allow configuration of cache lock TTL.

RedisCacheWriter now can now issue locks that expire using TtlFunction to prevent eternal locks.

Closes #2300
Pull request: #2597

Author: mp911de

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

@@ -50,6 +50,8 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 
 	private final RedisConnectionFactory connectionFactory;
 	private final Duration sleepTime;
+
+	private final TtlFunction lockTtl;
 	private final CacheStatisticsCollector statistics;
 	private final BatchStrategy batchStrategy;
 
@@ -68,26 +70,29 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 	 * @param batchStrategy must not be {@literal null}.
 	 */
 	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime, BatchStrategy batchStrategy) {
-		this(connectionFactory, sleepTime, CacheStatisticsCollector.none(), batchStrategy);
+		this(connectionFactory, sleepTime, TtlFunction.persistent(), 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 lockTtl Lock TTL function must not be {@literal null}.
 	 * @param cacheStatisticsCollector must not be {@literal null}.
 	 * @param batchStrategy must not be {@literal null}.
 	 */
-	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime,
+	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime, TtlFunction lockTtl,
 			CacheStatisticsCollector cacheStatisticsCollector, BatchStrategy batchStrategy) {
 
 		Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
 		Assert.notNull(sleepTime, "SleepTime must not be null");
+		Assert.notNull(lockTtl, "Lock TTL Function 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.lockTtl = lockTtl;
 		this.statistics = cacheStatisticsCollector;
 		this.batchStrategy = batchStrategy;
 	}
@@ -142,7 +147,7 @@ public byte[] putIfAbsent(String name, byte[] key, byte[] value, @Nullable Durat
 		return execute(name, connection -> {
 
 			if (isLockingCacheWriter()) {
-				doLock(name, connection);
+				doLock(name, key, value, connection);
 			}
 
 			try {
@@ -193,7 +198,7 @@ public void clean(String name, byte[] pattern) {
 			try {
 
 				if (isLockingCacheWriter()) {
-					doLock(name, connection);
+					doLock(name, name, pattern, connection);
 					wasLocked = true;
 				}
 
@@ -227,7 +232,8 @@ public void clearStatistics(String name) {
 
 	@Override
 	public RedisCacheWriter withStatisticsCollector(CacheStatisticsCollector cacheStatisticsCollector) {
-		return new DefaultRedisCacheWriter(connectionFactory, sleepTime, cacheStatisticsCollector, this.batchStrategy);
+		return new DefaultRedisCacheWriter(connectionFactory, sleepTime, lockTtl, cacheStatisticsCollector,
+				this.batchStrategy);
 	}
 
 	/**
@@ -236,7 +242,7 @@ public RedisCacheWriter withStatisticsCollector(CacheStatisticsCollector cacheSt
 	 * @param name the name of the cache to lock.
 	 */
 	void lock(String name) {
-		execute(name, connection -> doLock(name, connection));
+		execute(name, connection -> doLock(name, name, null, connection));
 	}
 
 	/**
@@ -248,8 +254,12 @@ void unlock(String name) {
 		executeLockFree(connection -> doUnlock(name, connection));
 	}
 
-	private Boolean doLock(String name, RedisConnection connection) {
-		return connection.setNX(createCacheLockKey(name), new byte[0]);
+	private Boolean doLock(String name, Object contextualKey, Object contextualValue, RedisConnection connection) {
+
+		Expiration expiration = lockTtl == null ? Expiration.persistent()
+				: Expiration.from(lockTtl.getTimeToLive(contextualKey, contextualValue));
+
+		return connection.set(createCacheLockKey(name), new byte[0], expiration, SetOption.SET_IF_ABSENT);
 	}
 
 	private Long doUnlock(String name, RedisConnection connection) {

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

@@ -67,10 +67,10 @@ public class RedisCache extends AbstractValueAdaptingCache {
 	 * Create a new {@link RedisCache}.
 	 *
 	 * @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 appropriate 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 appropriate
+	 * 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}.
 	 */
@@ -87,7 +87,6 @@ protected RedisCache(String name, RedisCacheWriter cacheWriter, RedisCacheConfig
 		this.cacheConfiguration = cacheConfiguration;
 	}
 
-
 	/**
 	 * Get {@link RedisCacheConfiguration} used.
 	 *
@@ -132,8 +131,7 @@ public <T> T get(Object key, Callable<T> valueLoader) {
 
 		ValueWrapper result = get(key);
 
-		return result != null ? (T) result.get()
-			: getSynchronized(key, valueLoader);
+		return result != null ? (T) result.get() : getSynchronized(key, valueLoader);
 	}
 
 	@SuppressWarnings("unchecked")
@@ -142,8 +140,7 @@ private synchronized <T> T getSynchronized(Object key, Callable<T> valueLoader)
 
 		ValueWrapper result = get(key);
 
-		return result != null ? (T) result.get()
-			: loadCacheValue(key, valueLoader);
+		return result != null ? (T) result.get() : loadCacheValue(key, valueLoader);
 	}
 
 	protected <T> T loadCacheValue(Object key, Callable<T> valueLoader) {
@@ -152,7 +149,8 @@ protected <T> T loadCacheValue(Object key, Callable<T> valueLoader) {
 
 		try {
 			value = valueLoader.call();
-		} catch (Exception cause) {
+		}
+		catch (Exception cause) {
 			throw new ValueRetrievalException(key, valueLoader, cause);
 		}
 
@@ -177,9 +175,8 @@ public void put(Object key, @Nullable Object value) {
 		if (!isAllowNullValues() && cacheValue == null) {
 
 			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());
+				+ " via '@Cacheable(unless=\"#result == null\")' or configure RedisCache to allow 'null'"
+				+ " via RedisCacheConfiguration", getName());
 
 			throw new IllegalArgumentException(message);
 		}
@@ -244,9 +241,7 @@ public void evict(Object key) {
 	@Nullable
 	protected Object preProcessCacheValue(@Nullable Object value) {
 
-		return value != null ? value
-			: isAllowNullValues() ? NullValue.INSTANCE
-			: null;
+		return value != null ? value : isAllowNullValues() ? NullValue.INSTANCE : null;
 	}
 
 	/**
@@ -327,7 +322,8 @@ protected String convertKey(Object key) {
 		if (conversionService.canConvert(source, TypeDescriptor.valueOf(String.class))) {
 			try {
 				return conversionService.convert(key, String.class);
-			} catch (ConversionFailedException cause) {
+			}
+			catch (ConversionFailedException cause) {
 
 				// May fail if the given key is a collection
 				if (isCollectionLikeOrMap(source)) {
@@ -342,7 +338,8 @@ protected String convertKey(Object key) {
 			return key.toString();
 		}
 
-		String message = String.format("Cannot convert cache key %s to String; Please register a suitable Converter"
+		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());
 
@@ -380,12 +377,13 @@ private String convertCollectionLikeOrMapKey(Object key, TypeDescriptor source)
 			target.append("}");
 
 			return target.toString();
-		} else if (source.isCollection() || source.isArray()) {
+		}
+		else if (source.isCollection() || source.isArray()) {
 
 			StringJoiner stringJoiner = new StringJoiner(",");
 
 			Collection<?> collection = source.isCollection() ? (Collection<?>) key
-					: Arrays.asList(ObjectUtils.toObjectArray(key));
+				: Arrays.asList(ObjectUtils.toObjectArray(key));
 
 			for (Object collectedKey : collection) {
 				stringJoiner.add(convertKey(collectedKey));

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

@@ -24,6 +24,7 @@
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.convert.converter.ConverterRegistry;
+import org.springframework.data.redis.cache.RedisCacheWriter.TtlFunction;
 import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.format.support.DefaultFormattingConversionService;
@@ -119,16 +120,24 @@ public static RedisCacheConfiguration defaultCacheConfig(@Nullable ClassLoader c
 
 	private final ConversionService conversionService;
 
-	private final Duration ttl;
+	private final TtlFunction ttl;
 
 	private final SerializationPair<String> keySerializationPair;
 	private final SerializationPair<Object> valueSerializationPair;
 
-	@SuppressWarnings("unchecked")
 	private RedisCacheConfiguration(Duration ttl, Boolean cacheNullValues, Boolean usePrefix, CacheKeyPrefix keyPrefix,
 			SerializationPair<String> keySerializationPair, SerializationPair<?> valueSerializationPair,
 			ConversionService conversionService) {
 
+		this(TtlFunction.just(ttl), cacheNullValues, usePrefix, keyPrefix, keySerializationPair, valueSerializationPair,
+				conversionService);
+	}
+
+	@SuppressWarnings("unchecked")
+	private RedisCacheConfiguration(TtlFunction ttl, Boolean cacheNullValues, Boolean usePrefix, CacheKeyPrefix keyPrefix,
+			SerializationPair<String> keySerializationPair, SerializationPair<?> valueSerializationPair,
+			ConversionService conversionService) {
+
 		this.ttl = ttl;
 		this.cacheNullValues = cacheNullValues;
 		this.usePrefix = usePrefix;
@@ -205,8 +214,23 @@ public RedisCacheConfiguration entryTtl(Duration ttl) {
 
 		Assert.notNull(ttl, "TTL duration must not be null");
 
-		return new RedisCacheConfiguration(ttl, cacheNullValues, usePrefix, keyPrefix, keySerializationPair,
-			valueSerializationPair, conversionService);
+		return entryTtl(TtlFunction.just(ttl));
+	}
+
+	/**
+	 * Set the {@link TtlFunction TTL function} to compute the time to live for cache entries.
+	 *
+	 * @param ttlFunction the {@link TtlFunction} to compute the time to live for cache entries, must not be
+	 *          {@literal null}.
+	 * @return new {@link RedisCacheConfiguration}.
+	 * @since 3.2
+	 */
+	public RedisCacheConfiguration entryTtl(TtlFunction ttlFunction) {
+
+		Assert.notNull(ttlFunction, "TtlFunction must not be null");
+
+		return new RedisCacheConfiguration(ttlFunction, cacheNullValues, usePrefix, keyPrefix, keySerializationPair,
+				valueSerializationPair, conversionService);
 	}
 
 	/**
@@ -304,7 +328,11 @@ public SerializationPair<Object> getValueSerializationPair() {
 	 * @return The expiration time (ttl) for cache entries. Never {@literal null}.
 	 */
 	public Duration getTtl() {
-		return ttl;
+		return getTtlFunction().getTimeToLive(null, null);
+	}
+
+	public TtlFunction getTtlFunction() {
+		return this.ttl;
 	}
 
 	/**

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

@@ -83,14 +83,31 @@ static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectio
 	 */
 	static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory,
 			BatchStrategy batchStrategy) {
+		return lockingRedisCacheWriter(connectionFactory, Duration.ofMillis(50), TtlFunction.persistent(), batchStrategy);
+	}
+
+	/**
+	 * Create new {@link RedisCacheWriter} with locking behavior.
+	 *
+	 * @param connectionFactory must not be {@literal null}.
+	 * @param sleepTime sleep time between lock access attempts, must not be {@literal null}.
+	 * @param lockTtlFunction TTL function to compute the Lock TTL. The function is called with contextual keys and values
+	 *          (such as the cache name on cleanup or the actual key/value on put requests). Must not be {@literal null}.
+	 * @param batchStrategy must not be {@literal null}.
+	 * @return new instance of {@link DefaultRedisCacheWriter}.
+	 * @since 3.2
+	 */
+	static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime,
+			TtlFunction lockTtlFunction, BatchStrategy batchStrategy) {
 
 		Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
 
-		return new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50), batchStrategy);
+		return new DefaultRedisCacheWriter(connectionFactory, sleepTime, lockTtlFunction, CacheStatisticsCollector.none(),
+				batchStrategy);
 	}
 
 	/**
-	 * Write the given key/value pair to Redis an set the expiration time if defined.
+	 * Write the given key/value pair to Redis and set the expiration time if defined.
 	 *
 	 * @param name The cache name must not be {@literal null}.
 	 * @param key The key for the cache entry. Must not be {@literal null}.
@@ -152,4 +169,48 @@ static RedisCacheWriter lockingRedisCacheWriter(RedisConnectionFactory connectio
 	 */
 	RedisCacheWriter withStatisticsCollector(CacheStatisticsCollector cacheStatisticsCollector);
 
+	/**
+	 * Function to compute the time to live from the cache {@code key} and {@code value}.
+	 *
+	 * @author Mark Paluch
+	 * @since 3.2
+	 */
+	@FunctionalInterface
+	interface TtlFunction {
+
+		/**
+		 * Creates a singleton {@link TtlFunction} using the given {@link Duration}.
+		 *
+		 * @param duration the time to live. Can be {@link Duration#ZERO} for persistent values (i.e. cache entry does not
+		 *          expire).
+		 * @return a singleton {@link TtlFunction} using {@link Duration}.
+		 */
+		static TtlFunction just(Duration duration) {
+
+			Assert.notNull(duration, "TTL Duration must not be null");
+
+			return new SingletonTtlFunction(duration);
+		}
+
+		/**
+		 * Returns a {@link TtlFunction} to create persistent entires that do not expire.
+		 *
+		 * @return a {@link TtlFunction} to create persistent entires that do not expire.
+		 */
+		static TtlFunction persistent() {
+			return just(Duration.ZERO);
+		}
+
+		/**
+		 * Compute a {@link Duration time to live duration} using the cache {@code key} and {@code value}. The time to live
+		 * is computed on each write operation. Redis uses milliseconds granularity for timeouts. Any more granular values
+		 * (e.g. micros or nanos) are not considered and are 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 time to live. Can be {@link Duration#ZERO} for persistent values (i.e. cache entry does not expire).
+		 */
+		Duration getTimeToLive(Object key, @Nullable Object value);
+	}
 }

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

@@ -0,0 +1,35 @@
+/*
+ * 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 java.time.Duration;
+
+import org.springframework.data.redis.cache.RedisCacheWriter.TtlFunction;
+import org.springframework.lang.Nullable;
+
+/**
+ * Singleton implementation of {@link TtlFunction}.
+ *
+ * @author Mark Paluch
+ * @since 3.2
+ */
+public record SingletonTtlFunction(Duration duration) implements TtlFunction {
+
+	@Override
+	public Duration getTimeToLive(Object key, @Nullable Object value) {
+		return this.duration;
+	}
+}