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

Fixes: #3444

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

Author: EddieChoCho

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.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.
@@ -130,7 +130,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private boolean sequenceAware;
 
-	private LockRegistry lockRegistry = new DefaultLockRegistry();
+	private LockRegistry<?> lockRegistry = new DefaultLockRegistry();
 
 	private boolean lockRegistrySet = false;
 
@@ -193,7 +193,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;
@@ -499,7 +499,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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 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.
@@ -65,7 +65,7 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe
 
 	private String outputChannelName;
 
-	private LockRegistry lockRegistry;
+	private LockRegistry<?> lockRegistry;
 
 	private MessageGroupStore messageStore;
 
@@ -122,7 +122,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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 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.
@@ -299,7 +299,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/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/SimpleMessageStore.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.
@@ -71,7 +71,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
 
 	private final long upperBoundTimeout;
 
-	private LockRegistry lockRegistry;
+	private LockRegistry<?> lockRegistry;
 
 	private boolean copyOnGet = false;
 
@@ -111,7 +111,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);
 	}
 
@@ -126,7 +126,7 @@ public SimpleMessageStore(int individualCapacity, int groupCapacity, LockRegistr
 	 * @since 4.3
 	 */
 	public SimpleMessageStore(int individualCapacity, int groupCapacity, long upperBoundTimeout,
-			LockRegistry lockRegistry) {
+			LockRegistry<?> lockRegistry) {
 
 		super(false);
 		Assert.notNull(lockRegistry, "The LockRegistry cannot be null");
@@ -162,7 +162,7 @@ public void setCopyOnGet(boolean copyOnGet) {
 		this.copyOnGet = copyOnGet;
 	}
 
-	public void setLockRegistry(LockRegistry lockRegistry) {
+	public void setLockRegistry(LockRegistry<?> lockRegistry) {
 		Assert.notNull(lockRegistry, "The LockRegistry cannot be null");
 		Assert.isTrue(!(this.isUsed), "Cannot change the lock registry after the store has been used");
 		this.lockRegistry = lockRegistry;

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,49 @@
+/*
+ * 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 6.3
+ */
+public interface DistributedLock extends Lock {
+
+	/**
+	 * Attempt to acquire a lock with a specific time-to-live
+	 * @param time the maximum time to wait for the lock unit
+	 * @param unit the time unit of the time argument
+	 * @param customTtl the specific time-to-live for the lock status data
+	 * @param customTtlUnit the time unit of the customTtl argument
+	 * @return true if the lock was acquired and 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 customTtlUnit) throws InterruptedException;
+
+	/**
+	 * Attempt to acquire a lock with a specific time-to-live
+	 * @param customTtl the specific time-to-live for the lock status data
+	 * @param customTtlUnit the time unit of the customTtl argument
+	 */
+	void lock(long customTtl, TimeUnit customTtlUnit);
+}

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.