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

Fixes: #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]>

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

Fixes: #3444

* Modify `LockRegistry`, `DistributedLock` 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]>

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

Fixes: #3444

* Modify DistributedLock` interfaces.
* Modify implementation of `JdbcLockRegistry`, `RedisLockRegistry`.
* Modify JavaDoc.

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

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

Fixes: #3444

* Maintain document

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

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

Fixes: #3444

* Add @deprecated

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

@@ -49,7 +49,7 @@
  */
 public class LockRequestHandlerAdvice extends AbstractRequestHandlerAdvice {
 
-	private final LockRegistry lockRegistry;
+	private final LockRegistry<?> lockRegistry;
 
 	private final Expression lockKeyExpression;
 
@@ -66,7 +66,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));
 	}
 
@@ -76,7 +76,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;
@@ -89,7 +89,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

@@ -107,7 +107,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);
 	}
 
@@ -123,7 +123,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,48 @@
+/*
+ * 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.time.Duration;
+import java.util.concurrent.locks.Lock;
+
+/**
+ * A distributed {@link Lock} extension.
+ *
+ * @author Eddie Cho
+ *
+ * @since 7.0
+ */
+public interface DistributedLock extends Lock {
+
+	/**
+	 * Attempt to acquire a lock with a specific time-to-live
+	 * @param ttl the specific time-to-live for the lock status data
+	 */
+	void lock(Duration ttl);
+
+	/**
+	 * Attempt to acquire a lock with a specific time-to-live
+	 * @param waitTime the maximum time to wait for the lock
+	 * @param ttl the specific time-to-live for the lock status data
+	 * @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(Duration waitTime, Duration ttl) throws InterruptedException;
+}