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]>

Author: EddieChoCho

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

@@ -16,11 +16,12 @@
 
 package org.springframework.integration.support.locks;
 
+import java.time.Duration;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 
 /**
- * A {@link Lock} implementing for spring distributed locks
+ * A distributed {@link Lock} extension.
  *
  * @author Eddie Cho
  *
@@ -30,24 +31,22 @@ 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
+	 * @param ttl the specific time-to-live for the lock status data
 	 */
-	void lock(long customTtl, TimeUnit customTtlTimeUnit);
+	void lock(Duration ttl);
 
 	/**
 	 * 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
+	 * @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(long time, TimeUnit unit, long customTtl, TimeUnit customTtlTimeUnit) throws InterruptedException;
+	boolean tryLock(long time, TimeUnit unit, Duration ttl) throws InterruptedException;
 }

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

@@ -16,7 +16,7 @@
 
 package org.springframework.integration.support.locks;
 
-import java.util.concurrent.TimeUnit;
+import java.time.Duration;
 import java.util.concurrent.locks.Lock;
 
 import org.springframework.scheduling.TaskScheduler;
@@ -45,11 +45,10 @@ public interface RenewableLockRegistry<L extends Lock> extends LockRegistry<L> {
 	 * 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
+	 * @param ttl the specific time-to-live for the lock status data
 	 *
 	 */
-	void renewLock(Object lockKey, long customTtl, TimeUnit customTtlTimeUnit);
+	void renewLock(Object lockKey, Duration ttl);
 
 	/**
 	 * Set the {@link TaskScheduler} to use for the renewal task.

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java

@@ -77,12 +77,19 @@ public class DefaultLockRepository
 	 */
 	public static final String DEFAULT_TABLE_PREFIX = "INT_";
 
+	/**
+	 * Default value for the time-to-live property.
+	 */
+	public static final Duration DEFAULT_TTL = Duration.ofSeconds(10);
+
 	private final String id;
 
 	private final JdbcTemplate template;
 
 	private final AtomicBoolean started = new AtomicBoolean();
 
+	private Duration ttl = DEFAULT_TTL;
+
 	private String prefix = DEFAULT_TABLE_PREFIX;
 
 	private String region = "DEFAULT";
@@ -109,8 +116,8 @@ public class DefaultLockRepository
 			""";
 
 	private String insertQuery = """
-			INSERT INTO %sLOCK (REGION, LOCK_KEY, CLIENT_ID, EXPIRED_AFTER)
-			VALUES (?, ?, ?, ?)
+			INSERT INTO %sLOCK (REGION, LOCK_KEY, CLIENT_ID, CREATED_DATE, EXPIRED_AFTER)
+			VALUES (?, ?, ?, ?, ?)
 			""";
 
 	private String countQuery = """
@@ -182,6 +189,14 @@ public void setPrefix(String prefix) {
 		this.prefix = prefix;
 	}
 
+	/**
+	 * Specify the time (in milliseconds) to expire deadlocks.
+	 * @param timeToLive the time to expire deadlocks.
+	 */
+	public void setTimeToLive(int timeToLive) {
+		this.ttl = Duration.ofMillis(timeToLive);
+	}
+
 	/**
 	 * Set a {@link PlatformTransactionManager} for operations.
 	 * Otherwise, a primary {@link PlatformTransactionManager} bean is obtained
@@ -233,7 +248,7 @@ public String getUpdateQuery() {
 	 * Set a custom {@code INSERT} query for a lock record.
 	 * The {@link #getInsertQuery()} can be used as a template for customization.
 	 * The default query is
-	 * {@code INSERT INTO %sLOCK (REGION, LOCK_KEY, CLIENT_ID, EXPIRED_AFTER) VALUES (?, ?, ?, ?)}.
+	 * {@code INSERT INTO %sLOCK (REGION, LOCK_KEY, CLIENT_ID, CREATED_DATE, EXPIRED_AFTER) VALUES (?, ?, ?, ?, ?)}.
 	 * For example a PostgreSQL {@code ON CONFLICT DO NOTHING} hint can be provided like this:
 	 * <pre class="code">
 	 * {@code
@@ -380,6 +395,11 @@ public boolean delete(String lock) {
 				transactionStatus -> this.template.update(this.deleteQuery, this.region, lock, this.id)) == 1;
 	}
 
+	@Override
+	public boolean acquire(String lock) {
+		return this.acquire(lock, this.ttl);
+	}
+
 	@Override
 	public boolean acquire(String lock, Duration ttlDuration) {
 		Boolean result =
@@ -391,7 +411,7 @@ this.region, lock, this.id, epochMillis()) > 0) {
 							}
 							try {
 								return this.template.update(this.insertQuery, this.region, lock, this.id,
-										ttlEpochMillis(ttlDuration)) > 0;
+										epochMillis(), ttlEpochMillis(ttlDuration)) > 0;
 							}
 							catch (DataIntegrityViolationException ex) {
 								return false;
@@ -417,6 +437,11 @@ public void deleteExpired() {
 						this.template.update(this.deleteExpiredQuery, this.region, epochMillis()));
 	}
 
+	@Override
+	public boolean renew(String lock) {
+		return this.renew(lock, this.ttl);
+	}
+
 	@Override
 	public boolean renew(String lock, Duration ttlDuration) {
 		final Boolean result = this.defaultTransactionTemplate.execute(

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java

@@ -102,13 +102,13 @@ public JdbcLockRegistry(LockRepository client) {
 		this.ttl = DEFAULT_TTL;
 	}
 
-	public JdbcLockRegistry(LockRepository client, long expireAfter) {
+	public JdbcLockRegistry(LockRepository client, Duration expireAfter) {
 		this.client = client;
-		this.ttl = convertToDuration(expireAfter, TimeUnit.MILLISECONDS);
+		this.ttl = expireAfter;
 	}
 
 	/**
-	 * Specify a @link Duration} to sleep between lock record insert/update attempts.
+	 * Specify a {@link Duration} to sleep between lock record insert/update attempts.
 	 * Defaults to 100 milliseconds.
 	 * @param idleBetweenTries the {@link Duration} to sleep between insert/update attempts.
 	 * @since 5.1.8
@@ -166,12 +166,7 @@ public void renewLock(Object lockKey) {
 	}
 
 	@Override
-	public void renewLock(Object lockKey, long customTtl, TimeUnit customTtlTimeUnit) {
-		Duration customTtlDuration = convertToDuration(customTtl, customTtlTimeUnit);
-		this.renewLock(lockKey, customTtlDuration);
-	}
-
-	private void renewLock(Object lockKey, Duration customTtlDuration) {
+	public void renewLock(Object lockKey, Duration customTtl) {
 		Assert.isInstanceOf(String.class, lockKey);
 		String path = pathFor((String) lockKey);
 		JdbcLock jdbcLock;
@@ -186,7 +181,7 @@ private void renewLock(Object lockKey, Duration customTtlDuration) {
 		if (jdbcLock == null) {
 			throw new IllegalStateException("Could not found mutex at " + path);
 		}
-		if (!jdbcLock.renew(customTtlDuration)) {
+		if (!jdbcLock.renew(customTtl)) {
 			throw new IllegalStateException("Could not renew mutex at " + path);
 		}
 	}
@@ -224,12 +219,7 @@ public void lock() {
 		}
 
 		@Override
-		public void lock(long customTtl, TimeUnit customTtlUnit) {
-			Duration customTtlDuration = convertToDuration(customTtl, customTtlUnit);
-			lock(customTtlDuration);
-		}
-
-		private void lock(Duration ttl) {
+		public void lock(Duration ttl) {
 			this.delegate.lock();
 			while (true) {
 				try {
@@ -304,12 +294,7 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
 		}
 
 		@Override
-		public boolean tryLock(long time, TimeUnit unit, long customTtl, TimeUnit customTtlUnit) throws InterruptedException {
-			Duration customTtlDuration = convertToDuration(customTtl, customTtlUnit);
-			return tryLock(time, unit, customTtlDuration);
-		}
-
-		private boolean tryLock(long time, TimeUnit unit, Duration ttl) throws InterruptedException {
+		public boolean tryLock(long time, TimeUnit unit, Duration ttl) throws InterruptedException {
 			long now = System.currentTimeMillis();
 			if (!this.delegate.tryLock(time, unit)) {
 				return false;
@@ -389,10 +374,21 @@ public boolean isAcquiredInThisProcess() {
 			return this.delegate.isLocked();
 		}
 
+		/**
+		 * Renew the time-to-live of the distributed lock
+		 * @return {@code true} if the lock's time-to-live was successfully renewed;
+		 *         {@code false} if the time-to-live could not be renewed
+		 */
 		public boolean renew() {
 			return renew(JdbcLockRegistry.this.ttl);
 		}
 
+		/**
+		 * Renew the time-to-live of the distributed lock
+		 * @param ttl the new time-to-live value for the lock status data
+		 * @return {@code true} if the lock's time-to-live was successfully renewed;
+		 *         {@code false} if the time-to-live could not be renewed
+		 */
 		public boolean renew(Duration ttl) {
 			if (!this.delegate.isHeldByCurrentThread()) {
 				throw new IllegalMonitorStateException("The current thread doesn't own mutex at " + this.path);

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java

@@ -52,6 +52,13 @@ public interface LockRepository extends Closeable {
 	 */
 	void deleteExpired();
 
+	/**
+	 * Acquire a lock for a key.
+	 * @param lock the key for lock to acquire.
+	 * @return acquired or not.
+	 */
+	boolean acquire(String lock);
+
 	/**
 	 * Acquire a lock for a key with specific time-to-live value
 	 * @param lock the key for lock to acquire.
@@ -60,6 +67,13 @@ public interface LockRepository extends Closeable {
 	 */
 	boolean acquire(String lock, Duration ttl);
 
+	/**
+	 * Renew the lease for a lock.
+	 * @param lock the lock to renew.
+	 * @return renewed or not.
+	 */
+	boolean renew(String lock);
+
 	/**
 	 * Renew the lease for a lock with specific time-to-live value
 	 * @param lock the key for lock to acquire.

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql

@@ -30,7 +30,8 @@ CREATE TABLE INT_LOCK  (
 	LOCK_KEY CHAR(36) NOT NULL,
 	REGION VARCHAR(100) NOT NULL,
 	CLIENT_ID CHAR(36),
-    EXPIRED_AFTER TIMESTAMP NOT NULL,
+        CREATED_DATE TIMESTAMP NOT NULL,
+        EXPIRED_AFTER TIMESTAMP NOT NULL,
 	constraint INT_LOCK_PK primary key (LOCK_KEY, REGION)
 );
 

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql

@@ -30,7 +30,8 @@ CREATE TABLE INT_LOCK  (
 	LOCK_KEY CHAR(36) NOT NULL,
 	REGION VARCHAR(100) NOT NULL,
 	CLIENT_ID CHAR(36),
-    EXPIRED_AFTER TIMESTAMP NOT NULL,
+        CREATED_DATE TIMESTAMP NOT NULL,
+        EXPIRED_AFTER TIMESTAMP NOT NULL,
 	constraint INT_LOCK_PK primary key (LOCK_KEY, REGION)
 );
 

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql

@@ -30,7 +30,8 @@ CREATE TABLE INT_LOCK  (
 	LOCK_KEY CHAR(36) NOT NULL,
 	REGION VARCHAR(100) NOT NULL,
 	CLIENT_ID CHAR(36),
-    EXPIRED_AFTER TIMESTAMP NOT NULL,
+        CREATED_DATE TIMESTAMP NOT NULL,
+        EXPIRED_AFTER TIMESTAMP NOT NULL,
 	constraint INT_LOCK_PK primary key (LOCK_KEY, REGION)
 );
 

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql

@@ -30,7 +30,8 @@ CREATE TABLE INT_LOCK  (
 	LOCK_KEY CHAR(36) NOT NULL,
 	REGION VARCHAR(100) NOT NULL,
 	CLIENT_ID CHAR(36),
-    EXPIRED_AFTER TIMESTAMP NOT NULL,
+        CREATED_DATE TIMESTAMP NOT NULL,
+        EXPIRED_AFTER TIMESTAMP NOT NULL,
 	constraint INT_LOCK_PK primary key (LOCK_KEY, REGION)
 );
 

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql

@@ -30,7 +30,8 @@ CREATE TABLE INT_LOCK  (
 	LOCK_KEY CHAR(36) NOT NULL,
 	REGION VARCHAR(100) NOT NULL,
 	CLIENT_ID CHAR(36),
-    EXPIRED_AFTER DATETIME(6) NOT NULL,
+        CREATED_DATE DATETIME(6) NOT NULL,
+        EXPIRED_AFTER DATETIME(6) NOT NULL,
 	constraint INT_LOCK_PK primary key (LOCK_KEY, REGION)
 ) ENGINE=InnoDB;