spring-projectsGH-3444: Add Custom TTL support for RedisLock, and JdbcLock

Fixes: spring-projects#3444

* Add `DistributedLock` interface which is implemented by `RedisLock` and `JdbcLock`.
* Modify `LockRegistry`, `ExpirableLockRegistry`, `RenewableLockRegistry` interfaces.
* Modify implementation of `DefaultLockRepository, `JdbcLockRegistry`, `RedisLockRegistry`
* Modify ddl of `INT_LOCK` table.
* Maintain test cases and documents.

Signed-off-by: Eddie Cho <[email protected]>

Author: EddieChoCho

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -135,7 +135,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private boolean sequenceAware;
 
-	private LockRegistry lockRegistry = new DefaultLockRegistry();
+	private LockRegistry<?> lockRegistry = new DefaultLockRegistry();
 
 	private boolean lockRegistrySet = false;
 
@@ -198,7 +198,7 @@ public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor) {
 		this(processor, new SimpleMessageStore(0), null, null);
 	}
 
-	public void setLockRegistry(LockRegistry lockRegistry) {
+	public void setLockRegistry(LockRegistry<?> lockRegistry) {
 		Assert.isTrue(!this.lockRegistrySet, "'this.lockRegistry' can not be reset once its been set");
 		Assert.notNull(lockRegistry, "'lockRegistry' must not be null");
 		this.lockRegistry = lockRegistry;
@@ -516,7 +516,7 @@ protected boolean isSequenceAware() {
 		return this.sequenceAware;
 	}
 
-	protected LockRegistry getLockRegistry() {
+	protected LockRegistry<?> getLockRegistry() {
 		return this.lockRegistry;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java

@@ -66,7 +66,7 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 
 	private String outputChannelName;
 
-	private LockRegistry lockRegistry;
+	private LockRegistry<?> lockRegistry;
 
 	private MessageGroupStore messageStore;
 
@@ -125,7 +125,7 @@ public void setOutputChannelName(String outputChannelName) {
 		this.outputChannelName = outputChannelName;
 	}
 
-	public void setLockRegistry(LockRegistry lockRegistry) {
+	public void setLockRegistry(LockRegistry<?> lockRegistry) {
 		this.lockRegistry = lockRegistry;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java

@@ -312,7 +312,7 @@ public S forceReleaseAdvice(Advice... advice) {
 	 * @param lockRegistry the {@link LockRegistry} to use.
 	 * @return the endpoint spec.
 	 */
-	public S lockRegistry(LockRegistry lockRegistry) {
+	public S lockRegistry(LockRegistry<?> lockRegistry) {
 		Assert.notNull(lockRegistry, "'lockRegistry' must not be null.");
 		this.handler.setLockRegistry(lockRegistry);
 		return _this();

spring-integration-core/src/main/java/org/springframework/integration/handler/advice/LockRequestHandlerAdvice.java

@@ -48,7 +48,7 @@
  */
 public class LockRequestHandlerAdvice extends AbstractRequestHandlerAdvice {
 
-	private final LockRegistry lockRegistry;
+	private final LockRegistry<?> lockRegistry;
 
 	private final Expression lockKeyExpression;
 
@@ -65,7 +65,7 @@ public class LockRequestHandlerAdvice extends AbstractRequestHandlerAdvice {
 	 * @param lockRegistry the {@link LockRegistry} to use.
 	 * @param lockKey the static (shared) lock key for all the calls.
 	 */
-	public LockRequestHandlerAdvice(LockRegistry lockRegistry, Object lockKey) {
+	public LockRequestHandlerAdvice(LockRegistry<?> lockRegistry, Object lockKey) {
 		this(lockRegistry, new ValueExpression<>(lockKey));
 	}
 
@@ -75,7 +75,7 @@ public LockRequestHandlerAdvice(LockRegistry lockRegistry, Object lockKey) {
 	 * @param lockRegistry the {@link LockRegistry} to use.
 	 * @param lockKeyExpression the SpEL expression to evaluate a lock key against request message.
 	 */
-	public LockRequestHandlerAdvice(LockRegistry lockRegistry, Expression lockKeyExpression) {
+	public LockRequestHandlerAdvice(LockRegistry<?> lockRegistry, Expression lockKeyExpression) {
 		Assert.notNull(lockRegistry, "'lockRegistry' must not be null");
 		Assert.notNull(lockKeyExpression, "'lockKeyExpression' must not be null");
 		this.lockRegistry = lockRegistry;
@@ -88,7 +88,7 @@ public LockRequestHandlerAdvice(LockRegistry lockRegistry, Expression lockKeyExp
 	 * @param lockRegistry the {@link LockRegistry} to use.
 	 * @param lockKeyFunction the function to evaluate a lock key against request message.
 	 */
-	public LockRequestHandlerAdvice(LockRegistry lockRegistry, Function<Message<?>, Object> lockKeyFunction) {
+	public LockRequestHandlerAdvice(LockRegistry<?> lockRegistry, Function<Message<?>, Object> lockKeyFunction) {
 		Assert.notNull(lockRegistry, "'lockRegistry' must not be null");
 		Assert.notNull(lockKeyFunction, "'lockKeyFunction' must not be null");
 		this.lockRegistry = lockRegistry;

spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2025 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.
@@ -65,7 +65,7 @@ public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStor
 
 	private final DefaultPropertiesPersister persister = new DefaultPropertiesPersister();
 
-	private final LockRegistry lockRegistry = new DefaultLockRegistry();
+	private final LockRegistry<Lock> lockRegistry = new DefaultLockRegistry();
 
 	private String baseDirectory = System.getProperty("java.io.tmpdir") + "/spring-integration/";
 

spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java

@@ -67,7 +67,7 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG
 
 	private boolean timeoutOnIdle;
 
-	private LockRegistry lockRegistry = new DefaultLockRegistry();
+	private LockRegistry<?> lockRegistry = new DefaultLockRegistry();
 
 	protected AbstractMessageGroupStore() {
 	}
@@ -127,12 +127,12 @@ public void setLazyLoadMessageGroups(boolean lazyLoadMessageGroups) {
 	 * @param lockRegistry lockRegistryType
 	 * @since 6.5
 	 */
-	public final void setLockRegistry(LockRegistry lockRegistry) {
+	public final void setLockRegistry(LockRegistry<?> lockRegistry) {
 		Assert.notNull(lockRegistry, "The LockRegistry cannot be null");
 		this.lockRegistry = lockRegistry;
 	}
 
-	protected LockRegistry getLockRegistry() {
+	protected LockRegistry<?> getLockRegistry() {
 		return this.lockRegistry;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java

@@ -106,7 +106,7 @@ public SimpleMessageStore(int individualCapacity, int groupCapacity, long upperB
 	 * @param lockRegistry       The lock registry.
 	 * @see #SimpleMessageStore(int, int, long, LockRegistry)
 	 */
-	public SimpleMessageStore(int individualCapacity, int groupCapacity, LockRegistry lockRegistry) {
+	public SimpleMessageStore(int individualCapacity, int groupCapacity, LockRegistry<?> lockRegistry) {
 		this(individualCapacity, groupCapacity, 0, lockRegistry);
 	}
 
@@ -122,7 +122,7 @@ public SimpleMessageStore(int individualCapacity, int groupCapacity, LockRegistr
 	 */
 	@SuppressWarnings("this-escape")
 	public SimpleMessageStore(int individualCapacity, int groupCapacity, long upperBoundTimeout,
-			LockRegistry lockRegistry) {
+			LockRegistry<?> lockRegistry) {
 
 		super(false);
 		Assert.notNull(lockRegistry, "The LockRegistry cannot be null");

spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2024 the original author or authors.
+ * Copyright 2016-2025 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.
@@ -74,7 +74,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 	 * A lock registry. The locks it manages should be global (whatever that means for the
 	 * system) and expiring, in case the holder dies without notifying anyone.
 	 */
-	private final LockRegistry locks;
+	private final LockRegistry<?> locks;
 
 	/**
 	 * Candidate for leader election. User injects this to receive callbacks on leadership
@@ -162,7 +162,7 @@ public String getRole() {
 	 * candidate (which just logs the leadership events).
 	 * @param locks lock registry
 	 */
-	public LockRegistryLeaderInitiator(LockRegistry locks) {
+	public LockRegistryLeaderInitiator(LockRegistry<?> locks) {
 		this(locks, new DefaultCandidate());
 	}
 
@@ -172,7 +172,7 @@ public LockRegistryLeaderInitiator(LockRegistry locks) {
 	 * @param locks lock registry
 	 * @param candidate leadership election candidate
 	 */
-	public LockRegistryLeaderInitiator(LockRegistry locks, Candidate candidate) {
+	public LockRegistryLeaderInitiator(LockRegistry<?> locks, Candidate candidate) {
 		Assert.notNull(locks, "'locks' must not be null");
 		Assert.notNull(candidate, "'candidate' must not be null");
 		this.locks = locks;

spring-integration-core/src/main/java/org/springframework/integration/support/locks/DefaultLockRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2024 the original author or authors.
+ * Copyright 2002-2025 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.
@@ -34,7 +34,7 @@
  * @since 2.1.1
  *
  */
-public final class DefaultLockRegistry implements LockRegistry {
+public final class DefaultLockRegistry implements LockRegistry<Lock> {
 
 	private final Lock[] lockTable;
 

spring-integration-core/src/main/java/org/springframework/integration/support/locks/DistributedLock.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2025 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.integration.support.locks;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+
+/**
+ * A {@link Lock} implementing for spring distributed locks
+ *
+ * @author Eddie Cho
+ *
+ * @since 7.0
+ */
+public interface DistributedLock extends Lock {
+
+	/**
+	 * Attempt to acquire a lock with a specific time-to-live
+	 * @param customTtl the specific time-to-live for the lock status data
+	 * @param customTtlTimeUnit the time unit of the {@code customTtl} argument
+	 */
+	void lock(long customTtl, TimeUnit customTtlTimeUnit);
+
+	/**
+	 * Acquires the lock with a specific time-to-live if it is free within the
+	 * given waiting time and the current thread has not been {@linkplain Thread#interrupt interrupted}.
+	 * @param time the maximum time to wait for the lock
+	 * @param unit the time unit of the {@code time} argument
+	 * @param customTtl the specific time-to-live for the lock status data
+	 * @param customTtlTimeUnit the time unit of the customTtl argument
+	 * @return {@code true} if the lock was acquired and {@code false}
+	 *         if the waiting time elapsed before the lock was acquired
+	 *
+	 * @throws InterruptedException if the current thread is interrupted
+	 *         while acquiring the lock (and interruption of lock
+	 *         acquisition is supported)
+	 */
+	boolean tryLock(long time, TimeUnit unit, long customTtl, TimeUnit customTtlTimeUnit) throws InterruptedException;
+}

spring-integration-core/src/main/java/org/springframework/integration/support/locks/ExpirableLockRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2025 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.
@@ -16,15 +16,18 @@
 
 package org.springframework.integration.support.locks;
 
+import java.util.concurrent.locks.Lock;
+
 /**
  * A {@link LockRegistry} implementing this interface supports the removal of aged locks
  * that are not currently locked.
+ * @param <L> The expected class of the lock implementation
  *
  * @author Gary Russell
  * @since 4.2
  *
  */
-public interface ExpirableLockRegistry extends LockRegistry {
+public interface ExpirableLockRegistry<L extends Lock> extends LockRegistry<L> {
 
 	/**
 	 * Remove locks last acquired more than 'age' ago that are not currently locked.

spring-integration-core/src/main/java/org/springframework/integration/support/locks/LockRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2023 the original author or authors.
+ * Copyright 2002-2025 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.
@@ -26,6 +26,7 @@
 
 /**
  * Strategy for maintaining a registry of shared locks.
+ * @param <L> The expected class of the lock implementation
  *
  * @author Oleg Zhurakousky
  * @author Gary Russell
@@ -34,14 +35,14 @@
  * @since 2.1.1
  */
 @FunctionalInterface
-public interface LockRegistry {
+public interface LockRegistry<L extends Lock> {
 
 	/**
 	 * Obtain the lock associated with the parameter object.
 	 * @param lockKey The object with which the lock is associated.
 	 * @return The associated lock.
 	 */
-	Lock obtain(Object lockKey);
+	L obtain(Object lockKey);
 
 	/**
 	 * Perform the provided task when the lock for the key is locked.

spring-integration-core/src/main/java/org/springframework/integration/support/locks/PassThruLockRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2025 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.
@@ -33,7 +33,7 @@
  * @since 2.2
  *
  */
-public final class PassThruLockRegistry implements LockRegistry {
+public final class PassThruLockRegistry implements LockRegistry<Lock> {
 
 	@Override
 	public Lock obtain(Object lockKey) {

spring-integration-core/src/main/java/org/springframework/integration/support/locks/RenewableLockRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 the original author or authors.
+ * Copyright 2020-2025 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.
@@ -16,19 +16,23 @@
 
 package org.springframework.integration.support.locks;
 
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+
 import org.springframework.scheduling.TaskScheduler;
 
 /**
  * A {@link LockRegistry} implementing this interface supports the renewal
  * of the time to live of a lock.
+ * @param <L> The expected class of the lock implementation
  *
  * @author Alexandre Strubel
  * @author Artem Bilan
  * @author Youbin Wu
  *
  * @since 5.4
  */
-public interface RenewableLockRegistry extends LockRegistry {
+public interface RenewableLockRegistry<L extends Lock> extends LockRegistry<L> {
 
 	/**
 	 * Renew the time to live of the lock is associated with the parameter object.
@@ -37,6 +41,16 @@ public interface RenewableLockRegistry extends LockRegistry {
 	 */
 	void renewLock(Object lockKey);
 
+	/**
+	 * Renew the time to live of the lock is associated with the parameter object with a specific value.
+	 * The lock must be held by the current thread
+	 * @param lockKey The object with which the lock is associated.
+	 * @param customTtl the specific time-to-live for the lock status data
+	 * @param customTtlTimeUnit the time unit of the {@code customTtl} argument
+	 *
+	 */
+	void renewLock(Object lockKey, long customTtl, TimeUnit customTtlTimeUnit);
+
 	/**
 	 * Set the {@link TaskScheduler} to use for the renewal task.
 	 * When renewalTaskScheduler is set, it will be used to periodically renew the lock to ensure that