From 7c2731dbb9eb13935a99e9eb6c4d90330ae81098 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Tue, 7 Jun 2016 14:53:01 +0100 Subject: [PATCH] INT-4048: Add `JdbcLockRegistry` support JIRA: https://jira.spring.io/browse/INT-4048 The semantics of the locks are the same as for the default lock registry (they have to be unlocked by the same thread that locked them). Updates from feedback Add interface for JdbcClient so transactions can bind to proxy Fix typo in exception message Fix drop script JdbcClient -> LockRepository Make lock() contract more like native Lock Make lockInterruptibly throw InterruptedException more Delete all expired locks, not just the ones that we own INT-4048: Polishing * `DefaultLockRepository`: replace `prefix` only once in the `afterPropertiesSet()` * Add JavaDocs to the `DefaultLockRepository` * `JdbcLockRegistry`: add `Thread.sleep(100)` to lock loops do not request DB so often like in case of clean `while(true)` * `JdbcLockRegistry`: implement more robust `tryLock(long time, TimeUnit unit)`. The logic mostly copied from `RedisLockRegistry` * Add `testExclusiveAccess()` to be sure that JDBC locks have exclusive access to the data sequence Document `JdbcLockRegistry` Upgrade to SF-4.3 GA --- build.gradle | 2 +- .../jdbc/lock/DefaultLockRepository.java | 157 +++++++++ .../jdbc/lock/JdbcLockRegistry.java | 249 ++++++++++++++ .../integration/jdbc/lock/LockRepository.java | 40 +++ .../integration/jdbc/schema-db2.sql | 8 + .../integration/jdbc/schema-derby.sql | 8 + .../integration/jdbc/schema-drop-db2.sql | 1 + .../integration/jdbc/schema-drop-derby.sql | 1 + .../integration/jdbc/schema-drop-h2.sql | 1 + .../integration/jdbc/schema-drop-hsqldb.sql | 1 + .../jdbc/schema-drop-mysql-5_6_4.sql | 1 + .../integration/jdbc/schema-drop-mysql.sql | 1 + .../jdbc/schema-drop-oracle10g.sql | 1 + .../jdbc/schema-drop-postgresql.sql | 1 + .../jdbc/schema-drop-sqlserver.sql | 1 + .../integration/jdbc/schema-drop-sybase.sql | 1 + .../integration/jdbc/schema-h2.sql | 8 + .../integration/jdbc/schema-hsqldb.sql | 8 + .../integration/jdbc/schema-mysql-5_6_4.sql | 8 + .../integration/jdbc/schema-mysql.sql | 8 + .../integration/jdbc/schema-oracle10g.sql | 8 + .../integration/jdbc/schema-postgresql.sql | 8 + .../integration/jdbc/schema-sqlserver.sql | 8 + .../integration/jdbc/schema-sybase.sql | 8 + .../src/main/sql/destroy.sql.vpp | 1 + .../src/main/sql/schema.sql.vpp | 8 + .../JdbcLockRegistryDifferentClientTests.java | 308 ++++++++++++++++++ .../lock/JdbcLockRegistryTests-context.xml | 38 +++ .../jdbc/lock/JdbcLockRegistryTests.java | 287 ++++++++++++++++ src/reference/asciidoc/jdbc.adoc | 33 ++ src/reference/asciidoc/whats-new.adoc | 5 + 31 files changed, 1217 insertions(+), 1 deletion(-) create mode 100644 spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java create mode 100644 spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java create mode 100644 spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests-context.xml create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java diff --git a/build.gradle b/build.gradle index 2be08d6b58..222f766728 100644 --- a/build.gradle +++ b/build.gradle @@ -141,7 +141,7 @@ subprojects { subproject -> springSecurityVersion = '4.1.0.RELEASE' springSocialTwitterVersion = '1.1.2.RELEASE' springRetryVersion = '1.1.2.RELEASE' - springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.0.BUILD-SNAPSHOT' + springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.0.RELEASE' springWsVersion = '2.3.0.RELEASE' xmlUnitVersion = '1.6' xstreamVersion = '1.4.7' diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java new file mode 100644 index 0000000000..6d6e75fed1 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/DefaultLockRepository.java @@ -0,0 +1,157 @@ +/* + * Copyright 2016 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 + * + * http://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.jdbc.lock; + +import java.util.Date; +import java.util.UUID; + +import javax.sql.DataSource; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Repository; +import org.springframework.transaction.annotation.Isolation; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.util.Assert; + +/** + * The default implementation of the {@link LockRepository} based on the + * table from the script presented in the {@code org/springframework/integration/jdbc/schema-*.sql}. + *

+ * This repository can't be shared between different {@link JdbcLockRegistry} instances. + * Otherwise it opens a possibility to break {@link java.util.concurrent.locks.Lock} contract, + * where {@link JdbcLockRegistry} uses non-shared {@link java.util.concurrent.locks.ReentrantLock}s + * for local synchronizations. + * + * @author Dave Syer + * @author Artem Bilan + * @since 4.3 + */ +@Repository +@Transactional +public class DefaultLockRepository implements LockRepository, InitializingBean { + + /** + * Default value for the table prefix property. + */ + public static final String DEFAULT_TABLE_PREFIX = "INT_"; + + private final String id = UUID.randomUUID().toString(); + + private final JdbcTemplate template; + + private int ttl = 10000; + + private String prefix = DEFAULT_TABLE_PREFIX; + + private String region = "DEFAULT"; + + 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=?"; + + + @Autowired + DefaultLockRepository(DataSource dataSource) { + this.template = new JdbcTemplate(dataSource); + } + + /** + * A unique grouping identifier for all locks persisted with this store. Using + * multiple regions allows the store to be partitioned (if necessary) for different + * purposes. Defaults to DEFAULT. + * @param region the region name to set + */ + public void setRegion(String region) { + Assert.hasText(region, "Region must not be null or empty."); + this.region = region; + } + + /** + * Specify the prefix for target data base table used from queries. + * @param prefix the prefix to set (default INT_). + */ + public void setPrefix(String prefix) { + this.prefix = prefix; + } + + /** + * Specify the time (in milliseconds) to expire dead locks. + * @param timeToLive the time to expire dead locks. + */ + public void setTimeToLive(int timeToLive) { + this.ttl = timeToLive; + } + + @Override + public void afterPropertiesSet() throws Exception { + this.deleteQuery = String.format(this.deleteQuery, this.prefix); + this.deleteExpiredQuery = String.format(this.deleteExpiredQuery, this.prefix); + this.deleteAllQuery = String.format(this.deleteAllQuery, this.prefix); + this.updateQuery = String.format(this.updateQuery, this.prefix); + this.insertQuery = String.format(this.insertQuery, this.prefix); + this.countQuery = String.format(this.countQuery, this.prefix); + } + + @Override + public void close() { + this.template.update(this.deleteAllQuery, this.region, this.id); + } + + @Override + public void delete(String lock) { + this.template.update(this.deleteQuery, this.region, lock, this.id); + } + + @Transactional(isolation = Isolation.SERIALIZABLE, timeout = 1) + @Override + public boolean acquire(String lock) { + deleteExpired(lock); + if (this.template.update(this.updateQuery, new Date(), this.region, lock, this.id) > 0) { + return true; + } + try { + return this.template.update(this.insertQuery, this.region, lock, this.id, new Date()) > 0; + } + catch (DuplicateKeyException e) { + return false; + } + } + + @Override + public boolean isAcquired(String lock) { + deleteExpired(lock); + return this.template.queryForObject(this.countQuery, Integer.class, this.region, lock, this.id, + new Date(System.currentTimeMillis() - this.ttl)) == 1; + } + + private int deleteExpired(String lock) { + return this.template.update(this.deleteExpiredQuery, this.region, lock, + new Date(System.currentTimeMillis() - this.ttl)); + } + +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java new file mode 100644 index 0000000000..bceae2fab5 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java @@ -0,0 +1,249 @@ +/* + * Copyright 2016 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 + * + * http://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.jdbc.lock; + +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +import org.springframework.integration.support.locks.DefaultLockRegistry; +import org.springframework.integration.support.locks.ExpirableLockRegistry; +import org.springframework.integration.support.locks.LockRegistry; +import org.springframework.integration.util.UUIDConverter; +import org.springframework.transaction.TransactionTimedOutException; +import org.springframework.util.Assert; + +/** + * + * A {@link LockRegistry} using a shared database to co-ordinate the locks. Provides the + * same semantics as the {@link DefaultLockRegistry}, but the locks taken will be global, + * as long as the underlying database supports the "serializable" isolation level in its + * transactions. + * + * @author Dave Syer + * @author Artem Bilan + * @since 4.3 + */ +public class JdbcLockRegistry implements ExpirableLockRegistry { + + private final Map locks = new HashMap(); + + private LockRepository client; + + public JdbcLockRegistry(LockRepository client) { + this.client = client; + } + + @Override + public Lock obtain(Object lockKey) { + Assert.isInstanceOf(String.class, lockKey); + String path = pathFor((String) lockKey); + JdbcLock lock = this.locks.get(path); + if (lock == null) { + synchronized (this.locks) { + lock = this.locks.get(path); + if (lock == null) { + lock = new JdbcLock(this.client, path); + this.locks.put(path, lock); + } + } + } + return lock; + } + + private String pathFor(String input) { + return input == null ? null : UUIDConverter.getUUID(input).toString(); + } + + @Override + public void expireUnusedOlderThan(long age) { + synchronized (this.locks) { + Iterator> iterator = this.locks.entrySet().iterator(); + long now = System.currentTimeMillis(); + while (iterator.hasNext()) { + Entry entry = iterator.next(); + JdbcLock lock = entry.getValue(); + if (now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess()) { + iterator.remove(); + } + } + } + } + + private static final class JdbcLock implements Lock { + + private final LockRepository mutex; + + private final String path; + + private volatile long lastUsed = System.currentTimeMillis(); + + private ReentrantLock delegate = new ReentrantLock(); + + private JdbcLock(LockRepository client, String path) { + this.mutex = client; + this.path = path; + } + + public long getLastUsed() { + return this.lastUsed; + } + + @Override + public void lock() { + this.delegate.lock(); + try { + while (true) { + try { + while (!doLock()) { + Thread.sleep(100); + } + break; + } + catch (TransactionTimedOutException e) { + // try again + } + catch (InterruptedException e) { + /* + * This method must be uninterruptible so catch and ignore + * interrupts and only break out of the while loop when + * we get the lock. + */ + } + } + } + catch (Exception e) { + this.delegate.unlock(); + throw new RuntimeException("Failed to lock mutex at " + this.path, e); + } + } + + @Override + public void lockInterruptibly() throws InterruptedException { + this.delegate.lockInterruptibly(); + try { + while (true) { + try { + while (!this.doLock()) { + Thread.sleep(100); + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException(); + } + } + break; + } + catch (TransactionTimedOutException e) { + // try again + } + } + } + catch (InterruptedException ie) { + this.delegate.unlock(); + throw ie; + } + catch (Exception e) { + this.delegate.unlock(); + throw new RuntimeException("Failed to lock mutex at " + this.path, e); + } + } + + @Override + public boolean tryLock() { + try { + return tryLock(0, TimeUnit.MICROSECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + + @Override + public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { + long now = System.currentTimeMillis(); + if (!this.delegate.tryLock(time, unit)) { + return false; + } + try { + long expire = now + TimeUnit.MILLISECONDS.convert(time, unit); + boolean acquired; + while (true) { + try { + while (!(acquired = doLock()) && System.currentTimeMillis() < expire) { + Thread.sleep(100); + } + if (!acquired) { + this.delegate.unlock(); + } + return acquired; + } + catch (TransactionTimedOutException e) { + // try again + } + } + } + catch (Exception e) { + this.delegate.unlock(); + throw new RuntimeException("Failed to lock mutex at " + this.path, e); + } + } + + private boolean doLock() { + boolean acquired = this.mutex.acquire(this.path); + if (acquired) { + this.lastUsed = System.currentTimeMillis(); + } + return acquired; + } + + @Override + public void unlock() { + if (!this.delegate.isHeldByCurrentThread()) { + throw new IllegalMonitorStateException("You do not own mutex at " + this.path); + } + if (this.delegate.getHoldCount() > 1) { + this.delegate.unlock(); + return; + } + try { + this.mutex.delete(this.path); + } + catch (Exception e) { + throw new RuntimeException("Failed to release mutex at " + this.path, e); + } + finally { + this.delegate.unlock(); + } + } + + @Override + public Condition newCondition() { + throw new UnsupportedOperationException("Conditions are not supported"); + } + + public boolean isAcquiredInThisProcess() { + return this.mutex.isAcquired(this.path); + } + + } + +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java new file mode 100644 index 0000000000..6bb508f3f6 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/LockRepository.java @@ -0,0 +1,40 @@ +/* + * Copyright 2016 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 + * + * http://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.jdbc.lock; + +import java.io.Closeable; + +/** + * Encapsulation of the SQL shunting that is needed for locks. A {@link JdbcLockRegistry} + * needs a reference to a spring-managed (transactional) client service, so this component + * has to be declared as a bean. + * + * @author Dave Syer + * @since 4.3 + */ +public interface LockRepository extends Closeable { + + boolean isAcquired(String lock); + + void delete(String lock); + + boolean acquire(String lock); + + @Override + void close(); + +} diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql index 2aa333d6a3..7385db5be8 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE TIMESTAMP DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ); + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql index 2aa333d6a3..7385db5be8 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE TIMESTAMP DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ); + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql index 77cc319e83..cfb68b5bc4 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-db2.sql @@ -3,4 +3,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP TABLE INT_LOCK ; DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql index 77cc319e83..cfb68b5bc4 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-derby.sql @@ -3,4 +3,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP TABLE INT_LOCK ; DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql index dd4c973773..c51aaafa05 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-h2.sql @@ -3,4 +3,5 @@ DROP TABLE INT_MESSAGE IF EXISTS; DROP TABLE INT_MESSAGE_GROUP IF EXISTS; DROP TABLE INT_GROUP_TO_MESSAGE IF EXISTS; +DROP TABLE INT_LOCK IF EXISTS; DROP INDEX INT_MESSAGE_IX1 IF EXISTS; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql index dd4c973773..c51aaafa05 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-hsqldb.sql @@ -3,4 +3,5 @@ DROP TABLE INT_MESSAGE IF EXISTS; DROP TABLE INT_MESSAGE_GROUP IF EXISTS; DROP TABLE INT_GROUP_TO_MESSAGE IF EXISTS; +DROP TABLE INT_LOCK IF EXISTS; DROP INDEX INT_MESSAGE_IX1 IF EXISTS; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql-5_6_4.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql-5_6_4.sql index fce4d8c950..badab9b282 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql-5_6_4.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql-5_6_4.sql @@ -3,4 +3,5 @@ DROP TABLE IF EXISTS INT_MESSAGE ; DROP TABLE IF EXISTS INT_MESSAGE_GROUP ; DROP TABLE IF EXISTS INT_GROUP_TO_MESSAGE ; +DROP TABLE IF EXISTS INT_LOCK ; DROP INDEX IF EXISTS INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql index fce4d8c950..badab9b282 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-mysql.sql @@ -3,4 +3,5 @@ DROP TABLE IF EXISTS INT_MESSAGE ; DROP TABLE IF EXISTS INT_MESSAGE_GROUP ; DROP TABLE IF EXISTS INT_GROUP_TO_MESSAGE ; +DROP TABLE IF EXISTS INT_LOCK ; DROP INDEX IF EXISTS INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql index 77cc319e83..cfb68b5bc4 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-oracle10g.sql @@ -3,4 +3,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP TABLE INT_LOCK ; DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql index 77cc319e83..cfb68b5bc4 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-postgresql.sql @@ -3,4 +3,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP TABLE INT_LOCK ; DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql index 77cc319e83..cfb68b5bc4 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sqlserver.sql @@ -3,4 +3,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP TABLE INT_LOCK ; DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql index 77cc319e83..cfb68b5bc4 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-drop-sybase.sql @@ -3,4 +3,5 @@ DROP TABLE INT_MESSAGE ; DROP TABLE INT_MESSAGE_GROUP ; DROP TABLE INT_GROUP_TO_MESSAGE ; +DROP TABLE INT_LOCK ; DROP INDEX INT_MESSAGE_IX1 ; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql index ec471f1f8c..9e65493c8b 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE TIMESTAMP DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ); + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql index ec471f1f8c..9e65493c8b 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE TIMESTAMP DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ); + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql-5_6_4.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql-5_6_4.sql index c11b398679..a55cdd1f38 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql-5_6_4.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql-5_6_4.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE DATETIME(6) DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ) ENGINE=InnoDB; + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE DATETIME(6) NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +) ENGINE=InnoDB; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql index 1546c7e623..336768179c 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE DATETIME DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ) ENGINE=InnoDB; + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE DATETIME NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +) ENGINE=InnoDB; diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql index 163c149499..1641269a7c 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE TIMESTAMP DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ); + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR2(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql index 2ae20d27d5..036c497b3e 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE TIMESTAMP DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ); + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql index 9dc762c882..f51739ae08 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE DATETIME DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ); + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE DATETIME NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql index a57d16196d..327770804a 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE DATETIME DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) ) LOCK DATAROWS; + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE DATETIME NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +) LOCK DATAROWS; diff --git a/spring-integration-jdbc/src/main/sql/destroy.sql.vpp b/spring-integration-jdbc/src/main/sql/destroy.sql.vpp index ed223b3683..ab382a93a9 100644 --- a/spring-integration-jdbc/src/main/sql/destroy.sql.vpp +++ b/spring-integration-jdbc/src/main/sql/destroy.sql.vpp @@ -1,4 +1,5 @@ DROP TABLE $!{IFEXISTSBEFORE} INT_MESSAGE $!{IFEXISTS}; DROP TABLE $!{IFEXISTSBEFORE} INT_MESSAGE_GROUP $!{IFEXISTS}; DROP TABLE $!{IFEXISTSBEFORE} INT_GROUP_TO_MESSAGE $!{IFEXISTS}; +DROP TABLE $!{IFEXISTSBEFORE} INT_LOCK $!{IFEXISTS}; DROP INDEX $!{IFEXISTSBEFORE} INT_MESSAGE_IX1 $!{IFEXISTS}; diff --git a/spring-integration-jdbc/src/main/sql/schema.sql.vpp b/spring-integration-jdbc/src/main/sql/schema.sql.vpp index 6ac97074b2..6b23674326 100644 --- a/spring-integration-jdbc/src/main/sql/schema.sql.vpp +++ b/spring-integration-jdbc/src/main/sql/schema.sql.vpp @@ -27,3 +27,11 @@ CREATE TABLE INT_MESSAGE_GROUP ( UPDATED_DATE ${TIMESTAMP} DEFAULT NULL, constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION) )#if(${VOODOO}) ${VOODOO}#end; + +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION ${VARCHAR}(100), + CLIENT_ID CHAR(36), + CREATED_DATE ${TIMESTAMP} NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +)#if(${VOODOO}) ${VOODOO}#end; diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java new file mode 100644 index 0000000000..02c2bc5657 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryDifferentClientTests.java @@ -0,0 +1,308 @@ +/* + * Copyright 2016 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 + * + * http://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.jdbc.lock; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; + +import javax.sql.DataSource; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.StopWatch; + +/** + * @author Dave Syer + * @author Artem Bilan + * @since 4.3 + */ +@ContextConfiguration("JdbcLockRegistryTests-context.xml") +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext // close at the end after class +public class JdbcLockRegistryDifferentClientTests { + + private static Log logger = LogFactory.getLog(JdbcLockRegistryDifferentClientTests.class); + + @Autowired + private JdbcLockRegistry registry; + + @Autowired + private LockRepository client; + + @Autowired + private ConfigurableApplicationContext context; + + private AnnotationConfigApplicationContext child; + + @Autowired + private DataSource dataSource; + + @Before + public void clear() { + this.registry.expireUnusedOlderThan(0); + this.client.close(); + this.child = new AnnotationConfigApplicationContext(); + this.child.register(DefaultLockRepository.class); + this.child.setParent(this.context); + this.child.refresh(); + } + + @After + public void close() { + if (this.child != null) { + this.child.close(); + } + } + + @Test + public void testSecondThreadLoses() throws Exception { + + for (int i = 0; i < 100; i++) { + + final JdbcLockRegistry registry1 = this.registry; + final JdbcLockRegistry registry2 = this.child.getBean(JdbcLockRegistry.class); + final Lock lock1 = registry1.obtain("foo"); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + lock1.lockInterruptibly(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + Lock lock2 = registry2.obtain("foo"); + try { + latch1.countDown(); + lock2.lockInterruptibly(); + latch2.await(10, TimeUnit.SECONDS); + locked.set(true); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + lock2.unlock(); + latch3.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + latch2.countDown(); + assertTrue(latch3.await(10, TimeUnit.SECONDS)); + assertTrue(locked.get()); + + } + + } + + @Test + public void testBothLock() throws Exception { + + for (int i = 0; i < 100; i++) { + + final JdbcLockRegistry registry1 = this.registry; + final JdbcLockRegistry registry2 = this.child.getBean(JdbcLockRegistry.class); + final List locked = new ArrayList(); + final CountDownLatch latch = new CountDownLatch(2); + ExecutorService pool = Executors.newFixedThreadPool(2); + pool.execute(new Runnable() { + + @Override + public void run() { + Lock lock = registry1.obtain("foo"); + try { + lock.lockInterruptibly(); + locked.add("1"); + latch.countDown(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + try { + lock.unlock(); + } + catch (Exception e) { + // ignore + } + } + } + }); + + pool.execute(new Runnable() { + + @Override + public void run() { + Lock lock = registry2.obtain("foo"); + try { + lock.lockInterruptibly(); + locked.add("2"); + latch.countDown(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + try { + lock.unlock(); + } + catch (Exception e) { + // ignore + } + } + } + }); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + // eventually they both get the lock and release it + assertTrue(locked.contains("1")); + assertTrue(locked.contains("2")); + + } + + } + + @Test + public void testOnlyOneLock() throws Exception { + + for (int i = 0; i < 100; i++) { + + final List locked = new ArrayList(); + final CountDownLatch latch = new CountDownLatch(20); + ExecutorService pool = Executors.newFixedThreadPool(6); + ArrayList> tasks = new ArrayList>(); + for (int j = 0; j < 20; j++) { + final DefaultLockRepository client = new DefaultLockRepository(this.dataSource); + client.afterPropertiesSet(); + this.context.getAutowireCapableBeanFactory().autowireBean(client); + Callable task = new Callable() { + + @Override + public Boolean call() { + Lock lock = new JdbcLockRegistry(client).obtain("foo"); + try { + if (locked.isEmpty() && lock.tryLock()) { + if (locked.isEmpty()) { + locked.add("done"); + return true; + } + } + } + finally { + try { + lock.unlock(); + } + catch (Exception e) { + // ignore + } + latch.countDown(); + } + return false; + } + }; + tasks.add(task); + } + logger.info("Starting: " + i); + pool.invokeAll(tasks); + + assertTrue(latch.await(10, TimeUnit.SECONDS)); + // eventually they both get the lock and release it + assertEquals(1, locked.size()); + assertTrue(locked.contains("done")); + + } + + } + + @Test + public void testExclusiveAccess() throws Exception { + DefaultLockRepository client1 = new DefaultLockRepository(dataSource); + client1.afterPropertiesSet(); + final DefaultLockRepository client2 = new DefaultLockRepository(dataSource); + client2.afterPropertiesSet(); + Lock lock1 = new JdbcLockRegistry(client1).obtain("foo"); + final BlockingQueue data = new LinkedBlockingQueue(); + final CountDownLatch latch1 = new CountDownLatch(1); + lock1.lockInterruptibly(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + Lock lock2 = new JdbcLockRegistry(client2).obtain("foo"); + try { + latch1.countDown(); + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + lock2.lockInterruptibly(); + stopWatch.stop(); + data.add(4); + Thread.sleep(10); + data.add(5); + Thread.sleep(10); + data.add(6); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + lock2.unlock(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + data.add(1); + Thread.sleep(1000); + data.add(2); + Thread.sleep(1000); + data.add(3); + lock1.unlock(); + for (int i = 0; i < 6; i++) { + Integer integer = data.poll(10, TimeUnit.SECONDS); + assertNotNull(integer); + assertEquals(i + 1, integer.intValue()); + } + } + +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests-context.xml new file mode 100644 index 0000000000..75fbd12174 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests-context.xml @@ -0,0 +1,38 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java new file mode 100644 index 0000000000..ae731edc7c --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java @@ -0,0 +1,287 @@ +/* + * Copyright 2016 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 + * + * http://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.jdbc.lock; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * @since 4.3 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext // close at the end after class +public class JdbcLockRegistryTests { + + @Autowired + private JdbcLockRegistry registry; + + @Autowired + private LockRepository client; + + @Before + public void clear() { + this.registry.expireUnusedOlderThan(0); + this.client.close(); + } + + @Test + public void testLock() throws Exception { + for (int i = 0; i < 10; i++) { + Lock lock = this.registry.obtain("foo"); + lock.lock(); + try { + assertEquals(1, TestUtils.getPropertyValue(this.registry, "locks", Map.class).size()); + } + finally { + lock.unlock(); + } + } + + Thread.sleep(10); + this.registry.expireUnusedOlderThan(0); + assertEquals(0, TestUtils.getPropertyValue(this.registry, "locks", Map.class).size()); + } + + @Test + public void testLockInterruptibly() throws Exception { + for (int i = 0; i < 10; i++) { + Lock lock = this.registry.obtain("foo"); + lock.lockInterruptibly(); + try { + assertEquals(1, TestUtils.getPropertyValue(this.registry, "locks", Map.class).size()); + } + finally { + lock.unlock(); + } + } + } + + @Test + public void testReentrantLock() throws Exception { + for (int i = 0; i < 10; i++) { + Lock lock1 = this.registry.obtain("foo"); + lock1.lock(); + try { + Lock lock2 = this.registry.obtain("foo"); + assertSame(lock1, lock2); + lock2.lock(); + lock2.unlock(); + } + finally { + lock1.unlock(); + } + } + } + + @Test + public void testReentrantLockInterruptibly() throws Exception { + for (int i = 0; i < 10; i++) { + Lock lock1 = this.registry.obtain("foo"); + lock1.lockInterruptibly(); + try { + Lock lock2 = this.registry.obtain("foo"); + assertSame(lock1, lock2); + lock2.lockInterruptibly(); + lock2.unlock(); + } + finally { + lock1.unlock(); + } + } + } + + @Test + public void testTwoLocks() throws Exception { + for (int i = 0; i < 10; i++) { + Lock lock1 = this.registry.obtain("foo"); + lock1.lockInterruptibly(); + try { + Lock lock2 = this.registry.obtain("bar"); + assertNotSame(lock1, lock2); + lock2.lockInterruptibly(); + lock2.unlock(); + } + finally { + lock1.unlock(); + } + } + } + + @Test + public void testTwoThreadsSecondFailsToGetLock() throws Exception { + final Lock lock1 = this.registry.obtain("foo"); + lock1.lockInterruptibly(); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch = new CountDownLatch(1); + Future result = Executors.newSingleThreadExecutor().submit(new Callable() { + + @Override + public Object call() throws Exception { + Lock lock2 = JdbcLockRegistryTests.this.registry.obtain("foo"); + locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS)); + latch.countDown(); + try { + lock2.unlock(); + } + catch (Exception e) { + return e; + } + return null; + } + }); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + Object ise = result.get(10, TimeUnit.SECONDS); + assertThat(ise, instanceOf(IllegalMonitorStateException.class)); + assertThat(((Exception) ise).getMessage(), containsString("You do not own")); + } + + @Test + public void testTwoThreads() throws Exception { + final Lock lock1 = this.registry.obtain("foo"); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + lock1.lockInterruptibly(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + Lock lock2 = JdbcLockRegistryTests.this.registry.obtain("foo"); + try { + latch1.countDown(); + lock2.lockInterruptibly(); + latch2.await(10, TimeUnit.SECONDS); + locked.set(true); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + lock2.unlock(); + latch3.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + latch2.countDown(); + assertTrue(latch3.await(10, TimeUnit.SECONDS)); + assertTrue(locked.get()); + } + + @Test + public void testTwoThreadsDifferentRegistries() throws Exception { + for (int i = 0; i < 100; i++) { + + final JdbcLockRegistry registry1 = new JdbcLockRegistry(this.client); + final JdbcLockRegistry registry2 = new JdbcLockRegistry(this.client); + final Lock lock1 = registry1.obtain("foo"); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + lock1.lockInterruptibly(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + Lock lock2 = registry2.obtain("foo"); + try { + latch1.countDown(); + lock2.lockInterruptibly(); + latch2.await(10, TimeUnit.SECONDS); + locked.set(true); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + lock2.unlock(); + latch3.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + latch2.countDown(); + assertTrue(latch3.await(10, TimeUnit.SECONDS)); + assertTrue(locked.get()); + + } + } + + @Test + public void testTwoThreadsWrongOneUnlocks() throws Exception { + final Lock lock = this.registry.obtain("foo"); + lock.lockInterruptibly(); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch = new CountDownLatch(1); + Future result = Executors.newSingleThreadExecutor().submit(new Callable() { + + @Override + public Object call() throws Exception { + try { + lock.unlock(); + } + catch (Exception e) { + latch.countDown(); + return e; + } + return null; + } + }); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock.unlock(); + Object imse = result.get(10, TimeUnit.SECONDS); + assertThat(imse, instanceOf(IllegalMonitorStateException.class)); + assertThat(((Exception) imse).getMessage(), containsString("You do not own")); + } + +} diff --git a/src/reference/asciidoc/jdbc.adoc b/src/reference/asciidoc/jdbc.adoc index 6519cdb638..61305a65d0 100644 --- a/src/reference/asciidoc/jdbc.adoc +++ b/src/reference/asciidoc/jdbc.adoc @@ -974,3 +974,36 @@ a `stored-proc-outbound-gateway` ---- + +[[jdbc-lock-registry]] +=== JDBC Lock Registry + +Starting with _version 4.3_, the `JdbcLockRegistry` is available. +Certain components (for example aggregator and resequencer) use a lock obtained from a `LockRegistry` instance to ensure that only one thread is manipulating a group at a time. +The `DefaultLockRegistry` performs this function within a single component; you can now configure an external lock registry on these components. +When used with a shared `MessageGroupStore`, the `JdbcLockRegistry` can be use to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time. + +When a lock is released by a local thread, another local thread will generally be able to acquire the lock immediately. +If a lock is released by a thread using a different registry instance, it can take up to 100ms to acquire the lock. + +The `JdbcLockRegistry` is based on the `LockRepository` abstraction, where a `DefaultLockRepository` implementation is present. +The data base schema scripts are located in the `org.springframework.integration.jdbc` package divided to the particular RDBMS vendors. +For example the H2 DDL for lock table looks like: + +[source,sql] +---- +CREATE TABLE INT_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); +---- + +The `INT_` can be changed according to the target data base design requirements. +Therefore `prefix` property must be used on the `DefaultLockRepository` bean definition. + +Sometimes it happens that one application has moved to the state when it can't release distributed lock - remove the particular record in the data base. +For this purpose such dead locks can be expired by the other application on the next locking invocation. +The `timeToLive` (TTL) option on the `DefaultLockRepository` is provided for this purpose. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e8d900bcb3..a4fac4efb8 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -43,6 +43,11 @@ See <> for more information. A new `IntegrationGraphServer` together with the `IntegrationGraphController` REST service are provided to expose the runtime model of a Spring Integration application as a graph. See <> for more information. +==== JDBC Lock Registry + +A new `JdbcLockRegistry` is provided for distributed locks shared through the data base table. +See <> for more information. + [[x4.3-general]] === General Changes