diff --git a/build.gradle b/build.gradle index c2c95167de..fd033f9d10 100644 --- a/build.gradle +++ b/build.gradle @@ -228,6 +228,14 @@ subprojects { subproject -> if (name ==~ /(testAll)/) { systemProperty 'RUN_LONG_INTEGRATION_TESTS', 'true' + systemProperty 'junit.jupiter.execution.timeout.default', '30 m' + } + else { + // Fail a stuck test instead of stalling the whole build with no output at all. + // The default `SAME_THREAD` mode interrupts the test thread and then still waits + // for the invocation to return, so a test wedged in a non-interruptible call + // is reported, but not aborted. + systemProperty 'junit.jupiter.execution.timeout.default', '10 m' } environment 'SI_FATAL_WHEN_NO_BEANFACTORY', 'true' 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 a8010a84a0..c054eb4e75 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 @@ -31,6 +31,7 @@ import org.apache.curator.framework.recipes.locks.InterProcessMutex; import org.springframework.beans.factory.DisposableBean; +import org.springframework.core.log.LogAccessor; import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.integration.support.locks.ExpirableLockRegistry; import org.springframework.messaging.MessagingException; @@ -54,6 +55,8 @@ */ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableBean { + private static final LogAccessor LOGGER = new LogAccessor(ZookeeperLockRegistry.class); + private static final String DEFAULT_ROOT = "/SpringIntegration-LockRegistry"; private final CuratorFramework client; @@ -118,7 +121,7 @@ public ZookeeperLockRegistry(CuratorFramework client, String root) { */ public ZookeeperLockRegistry(CuratorFramework client, KeyToPathStrategy keyToPath) { Assert.notNull(client, "'client' cannot be null"); - Assert.notNull(client, "'keyToPath' cannot be null"); + Assert.notNull(keyToPath, "'keyToPath' cannot be null"); this.client = client; this.keyToPath = keyToPath; this.trackingTime = !keyToPath.bounded(); @@ -286,12 +289,22 @@ public void lock() { @Override public void lockInterruptibly() throws InterruptedException { - boolean locked = false; + // The Lock contract: the interrupt status set on entry means no acquisition attempt at all. + checkInterruption(); // this is a bit ugly, but... - while (!locked) { - locked = tryLock(1, TimeUnit.SECONDS); + while (!tryLock(1, TimeUnit.SECONDS)) { + // In practice an interrupt is raised from the connection check the tryLock() blocks in. + // This is only a guard for the paths where the tryLock() returns 'false' instead, + // so this loop cannot silently spin on with the interrupt status set. + checkInterruption(); + LOGGER.debug(() -> "Mutex at " + this.path + " is not acquired yet; retrying..."); } + } + private void checkInterruption() throws InterruptedException { + if (Thread.interrupted()) { + throw new InterruptedException("Interrupted while acquiring mutex at " + this.path); + } } @Override @@ -326,18 +339,25 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { if (!connected) { future.cancel(true); + LOGGER.debug(() -> "No Zookeeper connection to acquire mutex at " + this.path); return false; } else { - waitTime = waitTime - (System.currentTimeMillis() - startTime); + // Defensive: never hand the mutex a negative deadline. + waitTime = Math.max(0, waitTime - (System.currentTimeMillis() - startTime)); return this.mutex.acquire(waitTime, TimeUnit.MILLISECONDS); } } catch (@SuppressWarnings("unused") TimeoutException e) { future.cancel(true); + LOGGER.debug(() -> + "Timed out while checking the Zookeeper connection to acquire mutex at " + this.path); return false; } catch (InterruptedException e) { + // Otherwise the abandoned connection check keeps the single executor thread busy + // for the whole `connectionTimeoutMs`, and every subsequent `tryLock()` queues behind it. + future.cancel(true); Thread.currentThread().interrupt(); throw e; } 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 e118642883..bb1e695f14 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 @@ -25,8 +25,12 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.Lock; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.BoundedExponentialBackoffRetry; +import org.apache.curator.test.TestingServer; import org.junit.jupiter.api.Test; import org.springframework.integration.test.util.TestUtils; @@ -51,7 +55,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { public void testLock() throws Exception { ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client, new TestKeyToPathStrategy()); for (int i = 0; i < 10; i++) { - Lock lock = registry.obtain("foo"); + Lock lock = registry.obtain("orders"); lock.lock(); try { assertThat(TestUtils.>getPropertyValue(registry, "locks")).hasSize(1); @@ -71,7 +75,7 @@ public void testLock() throws Exception { public void testLockInterruptibly() throws Exception { ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); for (int i = 0; i < 10; i++) { - Lock lock = registry.obtain("foo"); + Lock lock = registry.obtain("orders"); lock.lockInterruptibly(); try { assertThat(TestUtils.>getPropertyValue(registry, "locks")).hasSize(1); @@ -87,10 +91,10 @@ public void testLockInterruptibly() throws Exception { public void testReentrantLock() { ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); for (int i = 0; i < 10; i++) { - Lock lock1 = registry.obtain("foo"); + Lock lock1 = registry.obtain("orders"); lock1.lock(); try { - Lock lock2 = registry.obtain("foo"); + Lock lock2 = registry.obtain("orders"); assertThat(lock2).isSameAs(lock1); lock2.lock(); lock2.unlock(); @@ -106,10 +110,10 @@ public void testReentrantLock() { public void testReentrantLockInterruptibly() throws Exception { ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); for (int i = 0; i < 10; i++) { - Lock lock1 = registry.obtain("foo"); + Lock lock1 = registry.obtain("orders"); lock1.lockInterruptibly(); try { - Lock lock2 = registry.obtain("foo"); + Lock lock2 = registry.obtain("orders"); assertThat(lock2).isSameAs(lock1); lock2.lockInterruptibly(); lock2.unlock(); @@ -125,10 +129,10 @@ public void testReentrantLockInterruptibly() throws Exception { public void testTwoLocks() throws Exception { ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); for (int i = 0; i < 10; i++) { - Lock lock1 = registry.obtain("foo"); + Lock lock1 = registry.obtain("orders"); lock1.lockInterruptibly(); try { - Lock lock2 = registry.obtain("bar"); + Lock lock2 = registry.obtain("invoices"); assertThat(lock2).isNotSameAs(lock1); lock2.lockInterruptibly(); lock2.unlock(); @@ -143,13 +147,13 @@ public void testTwoLocks() throws Exception { @Test public void testTwoThreadsSecondFailsToGetLock() throws Exception { final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); - final Lock lock1 = registry.obtain("foo"); + final Lock lock1 = registry.obtain("orders"); lock1.lockInterruptibly(); final AtomicBoolean locked = new AtomicBoolean(); final CountDownLatch latch = new CountDownLatch(1); ExecutorService executorService = Executors.newSingleThreadExecutor(); Future result = executorService.submit(() -> { - Lock lock2 = registry.obtain("foo"); + Lock lock2 = registry.obtain("orders"); locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS)); latch.countDown(); try { @@ -173,7 +177,7 @@ public void testTwoThreadsSecondFailsToGetLock() throws Exception { @Test public void testTwoThreads() throws Exception { final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); - final Lock lock1 = registry.obtain("foo"); + final Lock lock1 = registry.obtain("orders"); final AtomicBoolean locked = new AtomicBoolean(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); @@ -181,7 +185,7 @@ public void testTwoThreads() throws Exception { lock1.lockInterruptibly(); ExecutorService executorService = Executors.newSingleThreadExecutor(); executorService.execute(() -> { - Lock lock2 = registry.obtain("foo"); + Lock lock2 = registry.obtain("orders"); try { latch1.countDown(); lock2.lockInterruptibly(); @@ -210,7 +214,7 @@ public void testTwoThreads() throws Exception { 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 Lock lock1 = registry1.obtain("orders"); final AtomicBoolean locked = new AtomicBoolean(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); @@ -218,7 +222,7 @@ public void testTwoThreadsDifferentRegistries() throws Exception { lock1.lockInterruptibly(); ExecutorService executorService = Executors.newSingleThreadExecutor(); executorService.execute(() -> { - Lock lock2 = registry2.obtain("foo"); + Lock lock2 = registry2.obtain("orders"); try { latch1.countDown(); lock2.lockInterruptibly(); @@ -247,7 +251,7 @@ public void testTwoThreadsDifferentRegistries() throws Exception { @Test public void testTwoThreadsWrongOneUnlocks() throws Exception { final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); - final Lock lock = registry.obtain("foo"); + final Lock lock = registry.obtain("orders"); lock.lockInterruptibly(); final AtomicBoolean locked = new AtomicBoolean(); final CountDownLatch latch = new CountDownLatch(1); @@ -277,7 +281,7 @@ public void testLockWithBoundedStrategy() throws Exception { ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client, key -> "/SpringIntegration-LockRegistry/singleLock"); for (int i = 0; i < 10; i++) { - Lock lock = registry.obtain("foo"); + Lock lock = registry.obtain("orders"); lock.lock(); try { assertThat(TestUtils.>getPropertyValue(registry, "locks")).hasSize(1); @@ -297,41 +301,89 @@ public void testLockWithBoundedStrategy() throws Exception { @Test public void voidLockFailsWhenServerDown() throws Exception { - ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + // This test brings the server down, so it must not use the one shared by the rest of the class: + // a failure to restart it would leave every subsequent test blocked forever in `InterProcessMutex.acquire()`. + try (var server = new TestingServer(); + var client = CuratorFrameworkFactory.newClient(server.getConnectString(), + new BoundedExponentialBackoffRetry(100, 1000, 3))) { - Lock lock1 = registry.obtain("foo"); - lock1.lock(); + client.start(); - testingServer.stop(); + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(client); - Lock lock2 = registry.obtain("bar"); + Lock ordersLock = registry.obtain("orders"); + ordersLock.lock(); - assertThat(lock2.tryLock(1, TimeUnit.SECONDS)) - .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); + server.stop(); - testingServer.restart(); + Lock invoicesLock = registry.obtain("invoices"); - assertThat(lock2.tryLock(10, TimeUnit.SECONDS)) - .as("Should have been able to lock with zookeeper server restarted!").isTrue(); + long startTime = System.currentTimeMillis(); - assertThat(lock1.tryLock(1, TimeUnit.SECONDS)).as("Should have still held lock1").isTrue(); + assertThat(invoicesLock.tryLock(1, TimeUnit.SECONDS)) + .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); - Lock lock3 = registry.obtain("foobar"); + assertThat(System.currentTimeMillis() - startTime) + .as("The tryLock() must not block far beyond the time requested!").isLessThan(3_000); - assertThat(lock3.tryLock(1, TimeUnit.SECONDS)).as("Should have been able to a obtain new lock!").isTrue(); + // The `lockInterruptibly()` retries until it succeeds, so with the server down + // an interrupt is the only way out of its loop. + Lock interruptibleLock = registry.obtain("interruptible"); + CountDownLatch lockAttemptLatch = new CountDownLatch(1); + AtomicReference lockException = new AtomicReference<>(); - lock1.unlock(); - lock1.unlock(); - lock2.unlock(); - lock3.unlock(); - registry.destroy(); + Thread lockThread = new Thread(() -> { + try { + interruptibleLock.lockInterruptibly(); + } + catch (Exception ex) { + lockException.set(ex); + } + finally { + lockAttemptLatch.countDown(); + } + }); + lockThread.start(); + + lockThread.interrupt(); + + assertThat(lockAttemptLatch.await(10, TimeUnit.SECONDS)) + .as("The lockInterruptibly() must return on an interrupt with zookeeper server stopped!") + .isTrue(); + assertThat(lockException.get()) + .as("The lockInterruptibly() must propagate the interrupt, not attempt an acquisition") + .isInstanceOf(InterruptedException.class); + + // Otherwise a failed assertion above would leave this thread racing for the lock below. + lockThread.join(10_000); + assertThat(lockThread.isAlive()).isFalse(); + + server.restart(); + + assertThat(invoicesLock.tryLock(10, TimeUnit.SECONDS)) + .as("Should have been able to lock with zookeeper server restarted!").isTrue(); + + assertThat(ordersLock.tryLock(1, TimeUnit.SECONDS)) + .as("Should have still held the lock obtained before the server went down!").isTrue(); + + Lock shipmentsLock = registry.obtain("shipments"); + + assertThat(shipmentsLock.tryLock(1, TimeUnit.SECONDS)) + .as("Should have been able to obtain a new lock!").isTrue(); + + ordersLock.unlock(); + ordersLock.unlock(); + invoicesLock.unlock(); + shipmentsLock.unlock(); + registry.destroy(); + } } @Test public void testTryLock() throws Exception { ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); for (int i = 0; i < 10; i++) { - Lock lock = registry.obtain("foo"); + Lock lock = registry.obtain("orders"); int n = 0; while (!lock.tryLock() && n++ < 100) { @@ -367,7 +419,7 @@ public void concurrentObtainCapacityTest() throws InterruptedException { catch (InterruptedException e) { Thread.currentThread().interrupt(); } - String keyId = "foo:" + finalI; + String keyId = "orders:" + finalI; Lock obtain = registry.obtain(keyId); maincountDownLatch.countDown(); obtain.lock(); @@ -375,7 +427,7 @@ public void concurrentObtainCapacityTest() throws InterruptedException { }); } executorService.shutdown(); - maincountDownLatch.await(); + assertThat(maincountDownLatch.await(30, TimeUnit.SECONDS)).isTrue(); executorService.awaitTermination(5, TimeUnit.SECONDS); //capacity limit test @@ -401,7 +453,7 @@ public void concurrentObtainRemoveOrderTest() throws InterruptedException { //Removed due to capcity limit for (int i = 0; i < DUMMY_LOCK_CNT; i++) { - Lock obtainLock0 = registry.obtain("foo:" + i); + Lock obtainLock0 = registry.obtain("orders:" + i); obtainLock0.lock(); obtainLock0.unlock(); } @@ -416,7 +468,7 @@ public void concurrentObtainRemoveOrderTest() throws InterruptedException { catch (InterruptedException e) { Thread.currentThread().interrupt(); } - String keyId = "foo:" + finalI; + String keyId = "orders:" + finalI; remainLockCheckQueue.offer(toKey(keyId)); Lock obtain = registry.obtain(keyId); obtain.lock(); @@ -437,7 +489,7 @@ public void concurrentObtainAccessRemoveOrderTest() throws InterruptedException final int DUMMY_LOCK_CNT = 3; final int CAPACITY_CNT = THREAD_CNT + 1; - final String REMAIN_DUMMY_LOCK_KEY = "foo:1"; + final String REMAIN_DUMMY_LOCK_KEY = "orders:1"; final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); @@ -447,7 +499,7 @@ public void concurrentObtainAccessRemoveOrderTest() throws InterruptedException //Removed due to capcity limit for (int i = 0; i < DUMMY_LOCK_CNT; i++) { - Lock obtainLock0 = registry.obtain("foo:" + i); + Lock obtainLock0 = registry.obtain("orders:" + i); obtainLock0.lock(); obtainLock0.unlock(); } @@ -467,7 +519,7 @@ public void concurrentObtainAccessRemoveOrderTest() throws InterruptedException catch (InterruptedException e) { Thread.currentThread().interrupt(); } - String keyId = "foo:" + finalI; + String keyId = "orders:" + finalI; remainLockCheckQueue.offer(toKey(keyId)); Lock obtain = registry.obtain(keyId); obtain.lock(); @@ -488,23 +540,23 @@ public void setCapacityTest() { final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); registry.setCacheCapacity(CAPACITY_CNT); - registry.obtain("foo:1"); - registry.obtain("foo:2"); - registry.obtain("foo:3"); + registry.obtain("orders:1"); + registry.obtain("orders:2"); + registry.obtain("orders:3"); //capacity 4->3 registry.setCacheCapacity(CAPACITY_CNT - 1); - registry.obtain("foo:4"); + registry.obtain("orders:4"); assertThat(getRegistryLocks(registry)).hasSize(3); - assertThat(getRegistryLocks(registry)).containsKeys(toKey("foo:2"), toKey("foo:3"), toKey("foo:4")); + assertThat(getRegistryLocks(registry)).containsKeys(toKey("orders:2"), toKey("orders:3"), toKey("orders:4")); //capacity 3->4 registry.setCacheCapacity(CAPACITY_CNT); - registry.obtain("foo:5"); + registry.obtain("orders:5"); assertThat(getRegistryLocks(registry)).hasSize(4); - assertThat(getRegistryLocks(registry)).containsKeys(toKey("foo:3"), toKey("foo:4"), toKey("foo:5")); + assertThat(getRegistryLocks(registry)).containsKeys(toKey("orders:3"), toKey("orders:4"), toKey("orders:5")); registry.destroy(); } @@ -558,6 +610,7 @@ public String pathFor(String key) { public boolean bounded() { return false; } + } }