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}.
+ * <p>
+ * 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<?";
+
+	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 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>=?";
+
+
+	@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 <code>DEFAULT</code>.
+	 * @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<String, JdbcLock> locks = new HashMap<String, JdbcLock>();
+
+	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<Entry<String, JdbcLock>> iterator = this.locks.entrySet().iterator();
+			long now = System.currentTimeMillis();
+			while (iterator.hasNext()) {
+				Entry<String, JdbcLock> 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<String> locked = new ArrayList<String>();
+			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<String> locked = new ArrayList<String>();
+			final CountDownLatch latch = new CountDownLatch(20);
+			ExecutorService pool = Executors.newFixedThreadPool(6);
+			ArrayList<Callable<Boolean>> tasks = new ArrayList<Callable<Boolean>>();
+			for (int j = 0; j < 20; j++) {
+				final DefaultLockRepository client = new DefaultLockRepository(this.dataSource);
+				client.afterPropertiesSet();
+				this.context.getAutowireCapableBeanFactory().autowireBean(client);
+				Callable<Boolean> task = new Callable<Boolean>() {
+
+					@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<Integer> data = new LinkedBlockingQueue<Integer>();
+		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 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:tx="http://www.springframework.org/schema/tx"
+	xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
+		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">
+
+	<jdbc:embedded-database id="dataSource" type="DERBY"/>
+
+	<jdbc:initialize-database data-source="dataSource" ignore-failures="ALL">
+		<jdbc:script location="${int.drop.script}"/>
+		<jdbc:script location="${int.schema.script}"/>
+	</jdbc:initialize-database>
+
+	<context:property-placeholder location="int-${ENVIRONMENT:derby}.properties"
+								  system-properties-mode="OVERRIDE"
+								  ignore-unresolvable="true"
+								  order="1"/>
+
+	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
+		<property name="dataSource" ref="dataSource"/>
+	</bean>
+
+	<bean id="lockRegistry" class="org.springframework.integration.jdbc.lock.JdbcLockRegistry">
+		<constructor-arg name="client" ref="lockClient"/>
+	</bean>
+
+	<bean id="lockClient" class="org.springframework.integration.jdbc.lock.DefaultLockRepository">
+		<constructor-arg name="dataSource" ref="dataSource"/>
+	</bean>
+	
+	<tx:annotation-driven/>
+
+</beans>
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<Object> result = Executors.newSingleThreadExecutor().submit(new Callable<Object>() {
+
+			@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<Object> result = Executors.newSingleThreadExecutor().submit(new Callable<Object>() {
+
+			@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`
     <int-jdbc:parameter name="ID" expression="payload" />
 </int-jdbc: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 <<stream-transformer>> 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 <<integration-graph>> for more information.
 
+==== JDBC Lock Registry
+
+A new `JdbcLockRegistry` is provided for distributed locks shared through the data base table.
+See <<jdbc-lock-registry>> for more information.
+
 [[x4.3-general]]
 === General Changes