JDBC Lock acquire optimization

Thanks to the pair of `CLIENT_ID`/`CREATED_DATE`,
it is possible to know if the lock is expired, already held or can be reacquire.
I think that the pre delete row is unnecessary and redundant with these two columns.
The rows doesn't need to be purged because the number of row in the table is finite and equal to the number of locks.
At worst, the table will be able to be purged by `deleteExpired()`.

I propose to overwrite these two columns at acquire lock time instead of pre delete the row.
So `delete` then `update` or `insert` become `update` or `insert`.

One client held a lock.
- Another client will be able to hold the same lock only if:
1. the row doesn't exist. Done when the first client will call `unlock()`.
2. the lock expire in case of hardware shutdown. Done by "`OR CREATED_DATE<?`" in `updateQuery`.
- The same client will be able to reacquire it only if:
1. the row doesn't exist.
2. the lock is already held by him. Done by "`CLIENT_ID=? OR`" in `updateQuery`.

**Cost with PostgreSQL:**
```
EXPLAIN DELETE FROM INT_LOCK WHERE REGION='DEFAULT' AND LOCK_KEY='FOO' AND CREATED_DATE<'2020-07-14';
Delete on int_lock  (cost=0.14..8.17 rows=1 width=6)
  ->  Index Scan using int_lock_pk on int_lock  (cost=0.14..8.17 rows=1 width=6)
        Index Cond: (((region)::text = 'DEFAULT'::text) AND (lock_key = 'FOO'::bpchar))
        Filter: (created_date < '2020-07-14 00:00:00'::timestamp without time zone)

EXPLAIN UPDATE INT_LOCK SET CREATED_DATE='2020-07-15' WHERE REGION='DEFAULT' AND LOCK_KEY='FOO' AND CLIENT_ID=NULL;
Update on int_lock  (cost=0.00..11.40 rows=1 width=520)
  ->  Result  (cost=0.00..11.40 rows=1 width=520)
        One-Time Filter: false
        ->  Seq Scan on int_lock  (cost=0.00..11.40 rows=1 width=520)
```
```
EXPLAIN UPDATE INT_LOCK SET CLIENT_ID=NULL, CREATED_DATE='2020-07-15' WHERE REGION='DEFAULT' AND LOCK_KEY='FOO' AND (CLIENT_ID=NULL OR CREATED_DATE<'2020-07-14')
Update on int_lock  (cost=0.14..8.17 rows=1 width=372)
  ->  Index Scan using int_lock_pk on int_lock  (cost=0.14..8.17 rows=1 width=372)
        Index Cond: (((region)::text = 'DEFAULT'::text) AND (lock_key = 'FOO'::bpchar))
        Filter: (created_date < '2020-07-14 00:00:00'::timestamp without time zone)
```

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

@@ -73,17 +73,21 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
 
 	private String deleteQuery = "DELETE FROM %sLOCK WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=?";
 
-	private String deleteExpiredQuery = "DELETE FROM %sLOCK WHERE REGION=? AND LOCK_KEY=? AND CREATED_DATE<?";
+	private String deleteExpiredQuery = "DELETE FROM %sLOCK WHERE REGION=? AND CREATED_DATE<?";
 
 	private String deleteAllQuery = "DELETE FROM %sLOCK WHERE REGION=? AND CLIENT_ID=?";
 
-	private String updateQuery = "UPDATE %sLOCK SET CREATED_DATE=? WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=?";
+	private String updateQuery =
+			"UPDATE %sLOCK SET CLIENT_ID=?, CREATED_DATE=? WHERE REGION=? AND LOCK_KEY=? " +
+					"AND (CLIENT_ID=? OR CREATED_DATE<?)";
 
 	private String insertQuery = "INSERT INTO %sLOCK (REGION, LOCK_KEY, CLIENT_ID, CREATED_DATE) VALUES (?, ?, ?, ?)";
 
 	private String countQuery =
 			"SELECT COUNT(REGION) FROM %sLOCK WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=? AND CREATED_DATE>=?";
 
+	private String renewQuery = "UPDATE %sLOCK SET CREATED_DATE=? WHERE REGION=? AND LOCK_KEY=? AND CLIENT_ID=?";
+
 	/**
 	 * Constructor that initializes the client id that will be associated for
 	 * all the locks persisted by the store instance to a random {@link UUID}.
@@ -142,6 +146,7 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
 		this.updateQuery = String.format(this.updateQuery, this.prefix);
 		this.insertQuery = String.format(this.insertQuery, this.prefix);
 		this.countQuery = String.format(this.countQuery, this.prefix);
+		this.renewQuery = String.format(this.renewQuery, this.prefix);
 	}
 
 	@Override
@@ -154,11 +159,11 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
 		this.template.update(this.deleteQuery, this.region, lock, this.id);
 	}
 
-	@Transactional(isolation = Isolation.SERIALIZABLE, timeout = 1)
+	@Transactional(isolation = Isolation.SERIALIZABLE)
 	@Override
 	public boolean acquire(String lock) {
-		deleteExpired(lock);
-		if (this.template.update(this.updateQuery, new Date(), this.region, lock, this.id) > 0) {
+		if (this.template.update(this.updateQuery, this.id, new Date(), this.region, lock, this.id,
+				new Date(System.currentTimeMillis() - this.ttl)) > 0) {
 			return true;
 		}
 		try {
@@ -171,19 +176,18 @@ public class DefaultLockRepository implements LockRepository, InitializingBean {
 
 	@Override
 	public boolean isAcquired(String lock) {
-		deleteExpired(lock);
 		return this.template.queryForObject(this.countQuery, Integer.class, // NOSONAR query never returns null
 				this.region, lock, this.id, new Date(System.currentTimeMillis() - this.ttl)) == 1;
 	}
 
-	private void deleteExpired(String lock) {
-		this.template.update(this.deleteExpiredQuery, this.region, lock,
-				new Date(System.currentTimeMillis() - this.ttl));
+	@Override
+	public void deleteExpired() {
+		this.template.update(this.deleteExpiredQuery, this.region, new Date(System.currentTimeMillis() - this.ttl));
 	}
 
 	@Override
 	public boolean renew(String lock) {
-		return this.template.update(this.updateQuery, new Date(), this.region, lock, this.id) > 0;
+		return this.template.update(this.renewQuery, new Date(), this.region, lock, this.id) > 0;
 	}
 
 }

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

@@ -34,6 +34,8 @@ public interface LockRepository extends Closeable {
 
 	void delete(String lock);
 
+	void deleteExpired();
+
 	boolean acquire(String lock);
 
 	boolean renew(String lock);

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java

@@ -36,10 +36,13 @@ import org.springframework.integration.test.util.TestUtils;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
+import javax.sql.DataSource;
+
 /**
  * @author Dave Syer
  * @author Artem Bilan
  * @author Stefan Vassilev
+ * @author Alexandre Strubel
  *
  * @since 4.3
  */
@@ -55,6 +58,9 @@ public class JdbcLockRegistryTests {
 	@Autowired
 	private LockRepository client;
 
+	@Autowired
+	private DataSource dataSource;
+
 	@BeforeEach
 	public void clear() {
 		this.registry.expireUnusedOlderThan(0);
@@ -127,6 +133,26 @@ public class JdbcLockRegistryTests {
 		}
 	}
 
+	@Test
+	public void testReentrantLockAfterExpiration() throws Exception {
+		DefaultLockRepository client = new DefaultLockRepository(dataSource);
+		client.setTimeToLive(1);
+		client.afterPropertiesSet();
+		JdbcLockRegistry registry = new JdbcLockRegistry(client);
+		Lock lock1 = registry.obtain("foo");
+		assertThat(lock1.tryLock()).isTrue();
+		Thread.sleep(100);
+		try {
+			Lock lock2 = registry.obtain("foo");
+			assertThat(lock2).isSameAs(lock1);
+			assertThat(lock2.tryLock()).isTrue();
+			lock2.unlock();
+		}
+		finally {
+			lock1.unlock();
+		}
+	}
+
 	@Test
 	public void testTwoLocks() throws Exception {
 		for (int i = 0; i < 10; i++) {