From 63e36cadb6a12929d4e4dc23076a77fc35f06f24 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 15 Aug 2016 22:15:41 -0400 Subject: [PATCH] INT-4087: Fix `tryLock(timeout)` Contract for ZK JIRA: https://jira.spring.io/browse/INT-4087 Fixes GH-1863 (https://github.com/spring-projects/spring-integration/issues/1863) If ZK Client loses the connection to ensemble, it tries to reconnect on the next operation and does that exactly for the `connectionTimeoutMs` and provided `RetryPolicy`. According to the `Lock.tryLock(long time, TimeUnit unit)` contract we can wait for the lock only during provided `timeout`. * Introduce `mutexTaskExecutor` and perform `ZK.forPath()` command in the separate `Thread` and wait for the `Future` during provided timeout. * Perform `mutex.acquire()` only after successful `Future` result and only for the remained timeout. **Cherry-pick to 4.2.x** Add `setMutexTaskExecutor()` Fix Checkstyle typo Javadoc --- .../zookeeper/lock/ZookeeperLockRegistry.java | 93 +++++++++++++++++-- .../zookeeper/ZookeeperTestSupport.java | 2 - .../zookeeper/lock/ZkLockRegistryTests.java | 48 +++++++++- 3 files changed, 132 insertions(+), 11 deletions(-) diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java index e4fe7af5d2..3edb6521b3 100644 --- a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java @@ -20,15 +20,23 @@ import java.util.HashMap; import java.util.Iterator; import java.util.Map; import java.util.Map.Entry; +import java.util.concurrent.Callable; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; 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.apache.zookeeper.CreateMode; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.integration.support.locks.ExpirableLockRegistry; import org.springframework.messaging.MessagingException; +import org.springframework.scheduling.concurrent.ExecutorConfigurationSupport; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.Assert; /** @@ -36,10 +44,11 @@ import org.springframework.util.Assert; * Curator {@link InterProcessMutex}. * * @author Gary Russell + * @author Artem Bilan * @since 4.2 * */ -public class ZookeeperLockRegistry implements ExpirableLockRegistry { +public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableBean { private static final String DEFAULT_ROOT = "/SpringIntegration-LockRegistry"; @@ -51,6 +60,17 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry { private final boolean trackingTime; + private AsyncTaskExecutor mutexTaskExecutor = new ThreadPoolTaskExecutor(); + + { + ThreadPoolTaskExecutor threadPoolTaskExecutor = (ThreadPoolTaskExecutor) this.mutexTaskExecutor; + threadPoolTaskExecutor.setAllowCoreThreadTimeOut(true); + threadPoolTaskExecutor.setBeanName("ZookeeperLockRegistryExecutor"); + threadPoolTaskExecutor.initialize(); + } + + private boolean mutexTaskExecutorExplicitlySet; + /** * Construct a lock registry using the default {@link KeyToPathStrategy} which * simple appends the key to '/SpringIntegration-LockRegistry/'. @@ -83,6 +103,23 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry { this.trackingTime = !keyToPath.bounded(); } + /** + * Set an {@link AsyncTaskExecutor} to use when establishing (and testing) the + * connection with Zookeeper. This must be performed asynchronously so the + * {@link Lock#tryLock(long, TimeUnit)} contract can be honored. While an executor is + * used internally, an external executor may be required in some environments, for + * example those that require the use of a {@code WorkManagerTaskExecutor}. + * @param mutexTaskExecutor the executor. + * + * @since 4.2.10 + */ + public void setMutexTaskExecutor(AsyncTaskExecutor mutexTaskExecutor) { + Assert.notNull(mutexTaskExecutor, "'mutexTaskExecutor' cannot be null"); + ((ExecutorConfigurationSupport) this.mutexTaskExecutor).shutdown(); + this.mutexTaskExecutor = mutexTaskExecutor; + this.mutexTaskExecutorExplicitlySet = true; + } + @Override public Lock obtain(Object lockKey) { Assert.isInstanceOf(String.class, lockKey); @@ -92,7 +129,7 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry { synchronized (this.locks) { lock = this.locks.get(path); if (lock == null) { - lock = new ZkLock(this.client, path); + lock = new ZkLock(this.client, this.mutexTaskExecutor, path); this.locks.put(path, lock); } if (this.trackingTime) { @@ -129,6 +166,13 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry { } } + @Override + public void destroy() throws Exception { + if (!this.mutexTaskExecutorExplicitlySet) { + ((ExecutorConfigurationSupport) this.mutexTaskExecutor).shutdown(); + } + } + /** * Strategy to convert a lock key (e.g. aggregation correlation id) to a * Zookeeper path. @@ -179,14 +223,20 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry { private static final class ZkLock implements Lock { + private final CuratorFramework client; + private final InterProcessMutex mutex; + private final AsyncTaskExecutor mutexTaskExecutor; + private final String path; private long lastUsed; - private ZkLock(CuratorFramework client, String path) { + private ZkLock(CuratorFramework client, AsyncTaskExecutor mutexTaskExecutor, String path) { + this.client = client; this.mutex = new InterProcessMutex(client, path); + this.mutexTaskExecutor = mutexTaskExecutor; this.path = path; } @@ -204,7 +254,7 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry { this.mutex.acquire(); } catch (Exception e) { - throw new RuntimeException("Failed to aquire mutex at " + this.path, e); + throw new RuntimeException("Failed to acquire mutex at " + this.path, e); } } @@ -231,11 +281,42 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry { @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { + Future future = null; try { - return this.mutex.acquire(time, unit); + long startTime = System.currentTimeMillis(); + + future = this.mutexTaskExecutor.submit(new Callable() { + + @Override + public String call() throws Exception { + return ZkLock.this.client.create() + .creatingParentContainersIfNeeded() + .withProtection() + .withMode(CreateMode.EPHEMERAL_SEQUENTIAL) + .forPath(ZkLock.this.path); + } + + }); + + long waitTime = unit.toMillis(time); + + String ourPath = future.get(waitTime, TimeUnit.MILLISECONDS); + + if (ourPath == null) { + future.cancel(true); + return false; + } + else { + waitTime = waitTime - (System.currentTimeMillis() - startTime); + return this.mutex.acquire(waitTime, TimeUnit.MILLISECONDS); + } + } + catch (TimeoutException e) { + future.cancel(true); + return false; } catch (Exception e) { - throw new MessagingException("Failed to aquire mutex at " + this.path, e); + throw new MessagingException("Failed to acquire mutex at " + this.path, e); } } diff --git a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/ZookeeperTestSupport.java b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/ZookeeperTestSupport.java index 6eb0cfb97d..64384724e2 100644 --- a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/ZookeeperTestSupport.java +++ b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/ZookeeperTestSupport.java @@ -17,7 +17,6 @@ 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; @@ -77,7 +76,6 @@ public class ZookeeperTestSupport { CuratorFramework client = CuratorFrameworkFactory.newClient(testingServer.getConnectString(), new BoundedExponentialBackoffRetry(100, 1000, 3)); client.start(); - client.blockUntilConnected(10000, TimeUnit.SECONDS); return client; } diff --git a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java index 13129c5b07..491bbfcd77 100644 --- a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java +++ b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-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. @@ -42,10 +42,9 @@ import org.springframework.integration.zookeeper.ZookeeperTestSupport; import org.springframework.integration.zookeeper.lock.ZookeeperLockRegistry.KeyToPathStrategy; import org.springframework.messaging.MessagingException; - - /** * @author Gary Russell + * @author Artem Bilan * @since 4.2 * */ @@ -68,6 +67,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { Thread.sleep(10); registry.expireUnusedOlderThan(0); assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size()); + registry.destroy(); } @Test @@ -83,6 +83,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { lock.unlock(); } } + registry.destroy(); } @Test @@ -101,6 +102,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { lock1.unlock(); } } + registry.destroy(); } @Test @@ -119,6 +121,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { lock1.unlock(); } } + registry.destroy(); } @Test @@ -137,6 +140,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { lock1.unlock(); } } + registry.destroy(); } @Test @@ -168,6 +172,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { Object ise = result.get(10, TimeUnit.SECONDS); assertThat(ise, instanceOf(IllegalMonitorStateException.class)); assertThat(((Exception) ise).getMessage(), containsString("You do not own")); + registry.destroy(); } @Test @@ -205,6 +210,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { latch2.countDown(); assertTrue(latch3.await(10, TimeUnit.SECONDS)); assertTrue(locked.get()); + registry.destroy(); } @Test @@ -243,6 +249,8 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { latch2.countDown(); assertTrue(latch3.await(10, TimeUnit.SECONDS)); assertTrue(locked.get()); + registry1.destroy(); + registry2.destroy(); } @Test @@ -272,6 +280,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { Object imse = result.get(10, TimeUnit.SECONDS); assertThat(imse, instanceOf(IllegalMonitorStateException.class)); assertThat(((Exception) imse).getMessage(), containsString("You do not own")); + registry.destroy(); } @Test @@ -308,6 +317,39 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { assertThat(e.getMessage(), containsString("expiry is not supported")); } assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size()); + registry.destroy(); + } + + @Test + public void voidLockFailsWhenServerDown() throws Exception { + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + + Lock lock1 = registry.obtain("foo"); + lock1.lock(); + + testingServer.stop(); + + Lock lock2 = registry.obtain("bar"); + + assertFalse("Should not have been able to lock with zookeeper server stopped!", + lock2.tryLock(1, TimeUnit.SECONDS)); + + testingServer.restart(); + + assertTrue("Should have been able to lock with zookeeper server restarted!", + lock2.tryLock(1, TimeUnit.SECONDS)); + + assertTrue("Should have still held lock1", lock1.tryLock(1, TimeUnit.SECONDS)); + + Lock lock3 = registry.obtain("foobar"); + + assertTrue("Should have been able to a obtain new lock!", lock3.tryLock(1, TimeUnit.SECONDS)); + + lock1.unlock(); + lock1.unlock(); + lock2.unlock(); + lock3.unlock(); + registry.destroy(); } }