INT-3616: Zookeeper LockRegistry

JIRA: https://jira.spring.io/browse/INT-3616

Zookeeper implementation of `LockRegistry`, for example to support
clustered aggregators (when using external message group stores).

INT-3616: Polishing; PR Comments

spring-integration-core/src/main/java/org/springframework/integration/metadata/ListenableMetadataStore.java

@@ -22,6 +22,7 @@ package org.springframework.integration.metadata;
  * invoked when changes occur in the metadata store.
  *
  * @author Marius Bogoevici
+ * @since 4.2
  */
 public interface ListenableMetadataStore extends ConcurrentMetadataStore {
 

spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListener.java

@@ -20,6 +20,7 @@ package org.springframework.integration.metadata;
  * A callback to be invoked whenever a value changes in the data store.
  *
  * @author Marius Bogoevici
+ * @since 4.2
  */
 public interface MetadataStoreListener {
 

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2015 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.support.locks;
+
+/**
+ * A {@link LockRegistry} implementing this interface supports the removal of aged locks
+ * that are not currently locked.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public interface ExpirableLockRegistry extends LockRegistry {
+
+	/**
+	 * Remove locks last acquired more than 'age' ago that are not currently locked.
+	 * @param age the time since the lock was last obtained.
+	 * @throws IllegalStateException if the registry configuration does not support this feature.
+	 */
+	void expireUnusedOlderThan(long age);
+
+}

spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EqualsResultMatcher.java

@@ -29,10 +29,11 @@ import org.springframework.util.ObjectUtils;
  * Mainly useful in conjunction with retrying matcherss such as {@link EventuallyMatcher}
  *
  * @author Marius Bogoevici
+ * @since 4.2
  */
 public class EqualsResultMatcher<U> extends DiagnosingMatcher<U> {
 
-	private Evaluator<U> evaluator;
+	private final Evaluator<U> evaluator;
 
 	public EqualsResultMatcher(Evaluator<U> evaluator) {
 		this.evaluator = evaluator;

spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EventuallyMatcher.java

@@ -29,6 +29,7 @@ import org.hamcrest.Matcher;
  * (Copied from {@code org.springframework.xd.test.fixtures.EventuallyMatcher})
  *
  * @author Eric Bottard
+ * @since 4.2
  */
 public class EventuallyMatcher<U> extends DiagnosingMatcher<U> {
 

spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java

@@ -0,0 +1,262 @@
+/*
+ * Copyright 2015 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.zookeeper.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 org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.locks.InterProcessMutex;
+
+import org.springframework.integration.support.locks.ExpirableLockRegistry;
+import org.springframework.messaging.MessagingException;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ExpirableLockRegistry} implementation using Zookeeper, or more specifically,
+ * Curator {@link InterProcessMutex}.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class ZookeeperLockRegistry implements ExpirableLockRegistry {
+
+	private static final String DEFAULT_ROOT = "/SpringIntegration-LockRegistry";
+
+	private final CuratorFramework client;
+
+	private final KeyToPathStrategy keyToPath;
+
+	private final Map<String, ZkLock> locks = new HashMap<String, ZkLock>();
+
+	private final boolean trackingTime;
+
+	/**
+	 * Construct a lock registry using the default {@link KeyToPathStrategy} which
+	 * simple appends the key to '/SpringIntegration-LockRegistry/'.
+	 * @param client the {@link CuratorFramework}.
+	 */
+	public ZookeeperLockRegistry(CuratorFramework client) {
+		this(client, DEFAULT_ROOT);
+	}
+
+	/**
+	 * Construct a lock registry using the default {@link KeyToPathStrategy} which
+	 * simple appends the key to {@code '<root>/'}.
+	 * @param client the {@link CuratorFramework}.
+	 * @param path the path root (no trailing /).
+	 */
+	public ZookeeperLockRegistry(CuratorFramework client, String root) {
+		this(client, new DefaultKeyToPathStrategy(root));
+	}
+
+	/**
+	 * Construct a lock registry using the supplied {@link KeyToPathStrategy}.
+	 * @param client the {@link CuratorFramework}.
+	 * @param keyToPath the implementation of {@link KeyToPathStrategy}.
+	 */
+	public ZookeeperLockRegistry(CuratorFramework client, KeyToPathStrategy keyToPath) {
+		Assert.notNull(client, "'client' cannot be null");
+		Assert.notNull(client, "'keyToPath' cannot be null");
+		this.client = client;
+		this.keyToPath = keyToPath;
+		this.trackingTime = !keyToPath.bounded();
+	}
+
+	@Override
+	public Lock obtain(Object lockKey) {
+		Assert.isInstanceOf(String.class, lockKey);
+		String path = this.keyToPath.pathFor((String) lockKey);
+		ZkLock lock = this.locks.get(path);
+		if (lock == null) {
+			synchronized (this.locks) {
+				lock = this.locks.get(path);
+				if (lock == null) {
+					lock = new ZkLock(this.client, path);
+					this.locks.put(path, lock);
+				}
+				if (this.trackingTime) {
+					lock.setLastUsed(System.currentTimeMillis());
+				}
+			}
+		}
+		return lock;
+	}
+
+	/**
+	 * Remove locks last acquired more than 'age' ago that are not currently locked.
+	 * Expiry is not supported if the {@link KeyToPathStrategy} is bounded (returns a finite
+	 * number of paths). With such a {@link KeyToPathStrategy}, the overhead of tracking when
+	 * a lock is obtained is avoided.
+	 * @param age the time since the lock was last obtained.
+	 */
+	@Override
+	public void expireUnusedOlderThan(long age) {
+		if (!this.trackingTime) {
+			throw new IllegalStateException("Ths KeyToPathStrategy is bounded; expiry is not supported");
+		}
+		synchronized(this.locks) {
+			Iterator<Entry<String, ZkLock>> iterator = this.locks.entrySet().iterator();
+			long now = System.currentTimeMillis();
+			while(iterator.hasNext()) {
+				Entry<String, ZkLock> entry = iterator.next();
+				ZkLock lock = entry.getValue();
+				if (now - lock.getLastUsed() > age
+						&& !lock.isAcquiredInThisProcess()) {
+					iterator.remove();
+				}
+			}
+		}
+	}
+
+	/**
+	 * Strategy to convert a lock key (e.g. aggregation correlation id) to a
+	 * Zookeeper path.
+	 *
+	 */
+	public interface KeyToPathStrategy {
+
+		/**
+		 * Return the path for the key.
+		 * @param key the key.
+		 * @return the path.
+		 */
+		String pathFor(String key);
+
+		/**
+		 * @return true if this strategy returns a bounded number of locks, removing
+		 * the need for removing LRU locks.
+		 */
+		boolean bounded();
+
+	}
+
+	private static class DefaultKeyToPathStrategy implements KeyToPathStrategy {
+
+		private final String root;
+
+		public DefaultKeyToPathStrategy(String rootPath) {
+			Assert.notNull(rootPath, "'rootPath' cannot be null");
+			if (!rootPath.endsWith("/")) {
+				this.root = rootPath + "/";
+			}
+			else {
+				this.root = rootPath;
+			}
+		}
+
+		@Override
+		public String pathFor(String key) {
+			return root + key;
+		}
+
+		@Override
+		public boolean bounded() {
+			return false;
+		}
+
+	}
+
+	private static class ZkLock implements Lock {
+
+		private final InterProcessMutex mutex;
+
+		private final String path;
+
+		private long lastUsed;
+
+		public ZkLock(CuratorFramework client, String path) {
+			this.mutex = new InterProcessMutex(client, path);
+			this.path = path;
+		}
+
+		public long getLastUsed() {
+			return this.lastUsed;
+		}
+
+		public void setLastUsed(long lastUsed) {
+			this.lastUsed = lastUsed;
+		}
+
+		@Override
+		public void lock() {
+			try {
+				this.mutex.acquire();
+			}
+			catch (Exception e) {
+				throw new RuntimeException("Failed to aquire mutex at " + this.path, e);
+			}
+		}
+
+		@Override
+		public void lockInterruptibly() throws InterruptedException {
+			boolean locked = false;
+			// this is a bit ugly, but...
+			while (!locked) {
+				locked = tryLock(1, TimeUnit.SECONDS);
+			}
+
+		}
+
+		@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 {
+			try {
+				return this.mutex.acquire(time, unit);
+			}
+			catch (Exception e) {
+				throw new MessagingException("Failed to aquire mutex at " + this.path, e);
+			}
+		}
+
+		@Override
+		public void unlock() {
+			try {
+				this.mutex.release();
+			}
+			catch (Exception e) {
+				throw new MessagingException("Failed to release mutex at " + this.path, e);
+			}
+		}
+
+		@Override
+		public Condition newCondition() {
+			throw new UnsupportedOperationException("Conditions are not supported");
+		}
+
+		public boolean isAcquiredInThisProcess() {
+			return this.mutex.isAcquiredInThisProcess();
+		}
+
+	}
+
+}

spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java

@@ -42,6 +42,7 @@ import org.springframework.util.Assert;
  * the names of which are stored as keys.
  *
  * @author Marius Bogoevici
+ * @since 4.2
  */
 public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLifecycle {
 
@@ -320,9 +321,9 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
 
 	private static class LocalChildData {
 
-		private String value;
+		private final String value;
 
-		private int version;
+		private final int version;
 
 		public LocalChildData(String value, int version) {
 			this.value = value;

spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreException.java

@@ -18,6 +18,7 @@ package org.springframework.integration.zookeeper.metadata;
 
 /**
  * @author Marius Bogoevici
+ * @since 4.2
  */
 @SuppressWarnings("serial")
 public class ZookeeperMetadataStoreException extends RuntimeException {

spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/ZookeeperTestSupport.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2015 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.zookeeper;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.BoundedExponentialBackoffRetry;
+import org.apache.curator.test.TestingServer;
+import org.apache.curator.utils.CloseableUtils;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+
+/**
+ * @author Marius Bogoevici
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class ZookeeperTestSupport {
+
+	private static final Log logger = LogFactory.getLog(ZookeeperTestSupport.class);
+
+	protected final Log log = LogFactory.getLog(this.getClass());
+
+	protected static TestingServer testingServer;
+
+	protected CuratorFramework client;
+
+	@BeforeClass
+	public static void setUpClass() throws Exception {
+		testingServer = new TestingServer(true);
+	}
+
+	@AfterClass
+	public static void tearDownClass() throws Exception {
+		try {
+			testingServer.stop();
+		}
+		catch (IOException e) {
+			logger.warn("Exception thrown while shutting down ZooKeeper: ", e);
+		}
+		testingServer.getTempDirectory().delete();
+	}
+
+	@Before
+	public void setUp() throws Exception{
+		client = createNewClient();
+	}
+
+	@After
+	public void tearDown() throws Exception {
+		CloseableUtils.closeQuietly(this.client);
+	}
+
+	protected CuratorFramework createNewClient() throws InterruptedException {
+		CuratorFramework client = CuratorFrameworkFactory.newClient(testingServer.getConnectString(),
+				new BoundedExponentialBackoffRetry(100, 1000, 3));
+		client.start();
+		client.blockUntilConnected(10000, TimeUnit.SECONDS);
+		return client;
+	}
+
+	protected void closeClient(CuratorFramework client) {
+		try {
+			CloseableUtils.closeQuietly(client);
+		}
+		catch (Exception e) {
+			log.warn("Exception thrown while closing client: ", e);
+		}
+	}
+
+}

spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java

@@ -0,0 +1,313 @@
+/*
+ * Copyright 2015 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.zookeeper.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 static org.junit.Assert.fail;
+
+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.Test;
+
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.integration.zookeeper.ZookeeperTestSupport;
+import org.springframework.integration.zookeeper.lock.ZookeeperLockRegistry.KeyToPathStrategy;
+import org.springframework.messaging.MessagingException;
+
+
+
+/**
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class ZkLockRegistryTests extends ZookeeperTestSupport {
+
+	@Test
+	public void testLock() throws Exception {
+		ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client);
+		for (int i = 0; i < 10; i++) {
+			Lock lock = registry.obtain("foo");
+			lock.lock();
+			try {
+				assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
+			}
+			finally {
+				lock.unlock();
+			}
+		}
+
+		Thread.sleep(10);
+		registry.expireUnusedOlderThan(0);
+		assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
+	}
+
+	@Test
+	public void testLockInterruptibly() throws Exception {
+		ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client);
+		for (int i = 0; i < 10; i++) {
+			Lock lock = registry.obtain("foo");
+			lock.lockInterruptibly();
+			try {
+				assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
+			}
+			finally {
+				lock.unlock();
+			}
+		}
+	}
+
+	@Test
+	public void testReentrantLock() throws Exception {
+		ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client);
+		for (int i = 0; i < 10; i++) {
+			Lock lock1 = registry.obtain("foo");
+			lock1.lock();
+			try {
+				Lock lock2 = registry.obtain("foo");
+				assertSame(lock1, lock2);
+				lock2.lock();
+				lock2.unlock();
+			}
+			finally {
+				lock1.unlock();
+			}
+		}
+	}
+
+	@Test
+	public void testReentrantLockInterruptibly() throws Exception {
+		ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client);
+		for (int i = 0; i < 10; i++) {
+			Lock lock1 = registry.obtain("foo");
+			lock1.lockInterruptibly();
+			try {
+				Lock lock2 = registry.obtain("foo");
+				assertSame(lock1, lock2);
+				lock2.lockInterruptibly();
+				lock2.unlock();
+			}
+			finally {
+				lock1.unlock();
+			}
+		}
+	}
+
+	@Test
+	public void testTwoLocks() throws Exception {
+		ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client);
+		for (int i = 0; i < 10; i++) {
+			Lock lock1 = registry.obtain("foo");
+			lock1.lockInterruptibly();
+			try {
+				Lock lock2 = registry.obtain("bar");
+				assertNotSame(lock1, lock2);
+				lock2.lockInterruptibly();
+				lock2.unlock();
+			}
+			finally {
+				lock1.unlock();
+			}
+		}
+	}
+
+	@Test
+	public void testTwoThreadsSecondFailsToGetLock() throws Exception {
+		final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client);
+		final Lock lock1 = 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 = registry.obtain("foo");
+				locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
+				latch.countDown();
+				try {
+					lock2.unlock();
+				}
+				catch (MessagingException e) {
+					return e.getCause();
+				}
+				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 ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client);
+		final Lock lock1 = 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 = 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 {
+		final ZookeeperLockRegistry registry1 = new ZookeeperLockRegistry(this.client);
+		final ZookeeperLockRegistry registry2 = new ZookeeperLockRegistry(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 ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client);
+		final Lock lock = 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.getCause();
+				}
+				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"));
+	}
+
+	@Test
+	public void testLockWithBoundedStrategy() throws Exception {
+		ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client, new KeyToPathStrategy() {
+
+			@Override
+			public String pathFor(String key) {
+				return "/SpringIntegration-LockRegistry/singleLock";
+			}
+
+			@Override
+			public boolean bounded() {
+				return true;
+			}
+		});
+		for (int i = 0; i < 10; i++) {
+			Lock lock = registry.obtain("foo");
+			lock.lock();
+			try {
+				assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
+			}
+			finally {
+				lock.unlock();
+			}
+		}
+
+		Thread.sleep(10);
+		try {
+			registry.expireUnusedOlderThan(0);
+			fail("expected exception");
+		}
+		catch (IllegalStateException e) {
+			assertThat(e.getMessage(), containsString("expiry is not supported"));
+		}
+		assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size());
+	}
+
+}

spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreTests.java

@@ -25,7 +25,6 @@ import static org.junit.Assert.fail;
 import static org.springframework.integration.test.matcher.EqualsResultMatcher.equalsResult;
 import static org.springframework.integration.test.matcher.EventuallyMatcher.eventually;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -34,71 +33,44 @@ import java.util.Map;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.BoundedExponentialBackoffRetry;
-import org.apache.curator.test.TestingServer;
 import org.apache.curator.utils.CloseableUtils;
 import org.hamcrest.collection.IsIterableContainingInOrder;
 import org.junit.After;
-import org.junit.AfterClass;
 import org.junit.Before;
-import org.junit.BeforeClass;
 import org.junit.Test;
 import org.mockito.Mockito;
 
 import org.springframework.beans.DirectFieldAccessor;
-import org.springframework.integration.metadata.MetadataStoreListenerAdapter;
 import org.springframework.integration.metadata.MetadataStoreListener;
+import org.springframework.integration.metadata.MetadataStoreListenerAdapter;
 import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.integration.test.matcher.EqualsResultMatcher.Evaluator;
+import org.springframework.integration.zookeeper.ZookeeperTestSupport;
 
 /**
  * @author Marius Bogoevici
+ * @since 4.2
  */
-public class ZookeeperMetadataStoreTests {
-
-	private static final Log log = LogFactory.getLog(ZookeeperMetadataStore.class);
-
-	private static TestingServer testingServer;
-
-	private CuratorFramework client;
+public class ZookeeperMetadataStoreTests extends ZookeeperTestSupport {
 
 	private ZookeeperMetadataStore metadataStore;
 
-	@BeforeClass
-	public static void setUpClass() throws Exception {
-		testingServer = new TestingServer(true);
-	}
-
-	@AfterClass
-	public static void tearDownClass() throws Exception {
-		try {
-			testingServer.stop();
-		}
-		catch (IOException e) {
-			log.warn("Exception thrown while shutting down ZooKeeper: ", e);
-		}
-		testingServer.getTempDirectory().delete();
-	}
-
+	@Override
 	@Before
-	public void setUp() throws Exception{
-		client = createNewClient();
-		metadataStore = new ZookeeperMetadataStore(client);
-		metadataStore.start();
+	public void setUp() throws Exception {
+		super.setUp();
+		this.metadataStore = new ZookeeperMetadataStore(client);
+		this.metadataStore.start();
 	}
 
+	@Override
 	@After
 	public void tearDown() throws Exception {
 		this.metadataStore.stop();
 		this.client.delete().deletingChildrenIfNeeded().forPath(this.metadataStore.getRoot());
-		CloseableUtils.closeQuietly(this.client);
 	}
 
-
 	@Test
 	public void testGetNonExistingKeyValue() {
 		String retrievedValue = metadataStore.get("does-not-exist");
@@ -413,12 +385,4 @@ public class ZookeeperMetadataStoreTests {
 		}
 	}
 
-	private CuratorFramework createNewClient() throws InterruptedException {
-		CuratorFramework client = CuratorFrameworkFactory.newClient(testingServer.getConnectString(),
-				new BoundedExponentialBackoffRetry(100, 1000, 3));
-		client.start();
-		client.blockUntilConnected(10000, TimeUnit.SECONDS);
-		return client;
-	}
-
 }