From c1c6fe753e25703f8840e0324b1a061a49a42ee3 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Aug 2026 16:22:37 -0400 Subject: [PATCH 01/14] Fix busy-spin hang in ZookeeperLockRegistry `ZkLock.lockInterruptibly()` looped on `tryLock(1, SECONDS)` until it succeeded. Since `tryLock()` returns `false` (rather than throwing) when the Zookeeper connection probe times out, an unreachable server turned that loop into an endless spin with no deadline, no interrupt check and no logging. The build had no test timeout configured at all, so such a hang stalled Gradle silently until the CI job limit. * check `Thread.interrupted()` per iteration in `lockInterruptibly()` and throw `InterruptedException`; log a `DEBUG` message per retry * log a `WARN` in `tryLock()` when the connection probe times out or reports no connection, instead of silently returning `false` * clamp the recomputed `waitTime` with `Math.max(0, ...)` so a slow probe cannot pass a negative deadline to `InterProcessMutex.acquire()` * set `queueCapacity` to `0` on the default `mutexTaskExecutor`: with Spring's defaults (core `1`, unbounded queue) the pool never grew past a single thread, so concurrent `tryLock()` calls serialized behind one connection probe * give `ZkLockRegistryTests.voidLockFailsWhenServerDown()` its own `TestingServer`; it used to stop and restart the one shared by the class, and a failed restart left the six tests declared after it blocked forever in `InterProcessMutex.acquire()` * bound the `maincountDownLatch.await()` in `concurrentObtainCapacityTest()` * add a `junit.jupiter.execution.timeout.default` for all `Test` tasks (`10 m`, `30 m` for `testAll`) so a stuck test fails with a stack trace of its thread **Auto-cherry-pick to `7.1.x` & `7.0.x`** --- build.gradle | 6 +++ .../zookeeper/lock/ZookeeperLockRegistry.java | 26 ++++++++-- .../zookeeper/lock/ZkLockRegistryTests.java | 52 ++++++++++++------- 3 files changed, 60 insertions(+), 24 deletions(-) diff --git a/build.gradle b/build.gradle index c2c95167de..0fd2de3d2a 100644 --- a/build.gradle +++ b/build.gradle @@ -228,6 +228,12 @@ subprojects { subproject -> if (name ==~ /(testAll)/) { systemProperty 'RUN_LONG_INTEGRATION_TESTS', 'true' + systemProperty 'junit.jupiter.execution.timeout.default', '30 m' + } + else { + // Fail a stuck test with a stack trace of its thread + // instead of stalling the whole build with no output at all. + 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..899c048a77 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 @@ -27,6 +27,8 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.recipes.locks.InterProcessMutex; @@ -54,6 +56,8 @@ */ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableBean { + private static final Log LOGGER = LogFactory.getLog(ZookeeperLockRegistry.class); + private static final String DEFAULT_ROOT = "/SpringIntegration-LockRegistry"; private final CuratorFramework client; @@ -84,6 +88,9 @@ protected boolean removeEldestEntry(Entry eldest) { { ThreadPoolTaskExecutor threadPoolTaskExecutor = (ThreadPoolTaskExecutor) this.mutexTaskExecutor; threadPoolTaskExecutor.setAllowCoreThreadTimeOut(true); + // A queue would make this executor effectively single-threaded, so concurrent `tryLock()` calls + // would serialize behind a single connection check - and time out on a slow or lost connection. + threadPoolTaskExecutor.setQueueCapacity(0); threadPoolTaskExecutor.setBeanName("ZookeeperLockRegistryExecutor"); threadPoolTaskExecutor.initialize(); } @@ -286,12 +293,19 @@ public void lock() { @Override public void lockInterruptibly() throws InterruptedException { - boolean locked = false; // this is a bit ugly, but... - while (!locked) { - locked = tryLock(1, TimeUnit.SECONDS); + while (!tryLock(1, TimeUnit.SECONDS)) { + // The tryLock() above may return 'false' without blocking at all, + // e.g. when Zookeeper is not reachable. + // Therefore, the interrupt status has to be checked explicitly + // to avoid an endless, silent spin in this loop. + if (Thread.interrupted()) { + throw new InterruptedException("Interrupted while acquiring mutex at " + this.path); + } + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Mutex at " + this.path + " is not acquired yet; retrying..."); + } } - } @Override @@ -326,15 +340,17 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { if (!connected) { future.cancel(true); + LOGGER.warn("No Zookeeper connection to acquire mutex at " + this.path); return false; } else { - waitTime = waitTime - (System.currentTimeMillis() - startTime); + waitTime = Math.max(0, waitTime - (System.currentTimeMillis() - startTime)); return this.mutex.acquire(waitTime, TimeUnit.MILLISECONDS); } } catch (@SuppressWarnings("unused") TimeoutException e) { future.cancel(true); + LOGGER.warn("Timed out while checking the Zookeeper connection to acquire mutex at " + this.path); return false; } catch (InterruptedException 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..89dfcc1218 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 @@ -27,6 +27,10 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; +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.junit.jupiter.api.Test; import org.springframework.integration.test.util.TestUtils; @@ -297,34 +301,44 @@ 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 (TestingServer ownTestingServer = new TestingServer(); + CuratorFramework ownClient = + CuratorFrameworkFactory.newClient(ownTestingServer.getConnectString(), + new BoundedExponentialBackoffRetry(100, 1000, 3))) { - Lock lock1 = registry.obtain("foo"); - lock1.lock(); + ownClient.start(); - testingServer.stop(); + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(ownClient); - Lock lock2 = registry.obtain("bar"); + Lock lock1 = registry.obtain("foo"); + lock1.lock(); - assertThat(lock2.tryLock(1, TimeUnit.SECONDS)) - .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); + ownTestingServer.stop(); - testingServer.restart(); + Lock lock2 = registry.obtain("bar"); - assertThat(lock2.tryLock(10, TimeUnit.SECONDS)) - .as("Should have been able to lock with zookeeper server restarted!").isTrue(); + assertThat(lock2.tryLock(1, TimeUnit.SECONDS)) + .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); - assertThat(lock1.tryLock(1, TimeUnit.SECONDS)).as("Should have still held lock1").isTrue(); + ownTestingServer.restart(); - Lock lock3 = registry.obtain("foobar"); + assertThat(lock2.tryLock(10, TimeUnit.SECONDS)) + .as("Should have been able to lock with zookeeper server restarted!").isTrue(); - assertThat(lock3.tryLock(1, TimeUnit.SECONDS)).as("Should have been able to a obtain new lock!").isTrue(); + assertThat(lock1.tryLock(1, TimeUnit.SECONDS)).as("Should have still held lock1").isTrue(); - lock1.unlock(); - lock1.unlock(); - lock2.unlock(); - lock3.unlock(); - registry.destroy(); + Lock lock3 = registry.obtain("foobar"); + + assertThat(lock3.tryLock(1, TimeUnit.SECONDS)).as("Should have been able to a obtain new lock!").isTrue(); + + lock1.unlock(); + lock1.unlock(); + lock2.unlock(); + lock3.unlock(); + registry.destroy(); + } } @Test @@ -375,7 +389,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 From 43d64a74cb5898ab18350eb1fac85fc8afb2c86f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Aug 2026 16:28:26 -0400 Subject: [PATCH 02/14] Polish ZookeeperLockRegistry test and comments * use `var` in the try-with-resources of `voidLockFailsWhenServerDown()` * rewrap the retry comment in `lockInterruptibly()` * code style in `ZkLockRegistryTests` --- .../zookeeper/lock/ZookeeperLockRegistry.java | 3 +- .../zookeeper/lock/ZkLockRegistryTests.java | 48 +++++++++---------- 2 files changed, 25 insertions(+), 26 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 899c048a77..08bc46029c 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 @@ -295,8 +295,7 @@ public void lock() { public void lockInterruptibly() throws InterruptedException { // this is a bit ugly, but... while (!tryLock(1, TimeUnit.SECONDS)) { - // The tryLock() above may return 'false' without blocking at all, - // e.g. when Zookeeper is not reachable. + // The tryLock() above may return 'false' without blocking at all, e.g. when Zookeeper is not reachable. // Therefore, the interrupt status has to be checked explicitly // to avoid an endless, silent spin in this loop. if (Thread.interrupted()) { 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 89dfcc1218..cf845a82ca 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 @@ -27,7 +27,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; -import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.retry.BoundedExponentialBackoffRetry; import org.apache.curator.test.TestingServer; @@ -303,41 +302,41 @@ public void testLockWithBoundedStrategy() throws Exception { public void voidLockFailsWhenServerDown() throws Exception { // 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 (TestingServer ownTestingServer = new TestingServer(); - CuratorFramework ownClient = - CuratorFrameworkFactory.newClient(ownTestingServer.getConnectString(), - new BoundedExponentialBackoffRetry(100, 1000, 3))) { + try (var server = new TestingServer()) { + try (var client = CuratorFrameworkFactory.newClient(server.getConnectString(), + new BoundedExponentialBackoffRetry(100, 1000, 3))) { - ownClient.start(); + client.start(); - ZookeeperLockRegistry registry = new ZookeeperLockRegistry(ownClient); + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(client); - Lock lock1 = registry.obtain("foo"); - lock1.lock(); + Lock lock1 = registry.obtain("foo"); + lock1.lock(); - ownTestingServer.stop(); + server.stop(); - Lock lock2 = registry.obtain("bar"); + Lock lock2 = registry.obtain("bar"); - assertThat(lock2.tryLock(1, TimeUnit.SECONDS)) - .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); + assertThat(lock2.tryLock(1, TimeUnit.SECONDS)) + .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); - ownTestingServer.restart(); + server.restart(); - assertThat(lock2.tryLock(10, TimeUnit.SECONDS)) - .as("Should have been able to lock with zookeeper server restarted!").isTrue(); + assertThat(lock2.tryLock(10, TimeUnit.SECONDS)) + .as("Should have been able to lock with zookeeper server restarted!").isTrue(); - assertThat(lock1.tryLock(1, TimeUnit.SECONDS)).as("Should have still held lock1").isTrue(); + assertThat(lock1.tryLock(1, TimeUnit.SECONDS)).as("Should have still held lock1").isTrue(); - Lock lock3 = registry.obtain("foobar"); + Lock lock3 = registry.obtain("foobar"); - assertThat(lock3.tryLock(1, TimeUnit.SECONDS)).as("Should have been able to a obtain new lock!").isTrue(); + assertThat(lock3.tryLock(1, TimeUnit.SECONDS)).as("Should have been able to a obtain new lock!").isTrue(); - lock1.unlock(); - lock1.unlock(); - lock2.unlock(); - lock3.unlock(); - registry.destroy(); + lock1.unlock(); + lock1.unlock(); + lock2.unlock(); + lock3.unlock(); + registry.destroy(); + } } } @@ -572,6 +571,7 @@ public String pathFor(String key) { public boolean bounded() { return false; } + } } From 770bced6448cf6eb561c1d07425ae1565f1dde3a Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Aug 2026 16:34:55 -0400 Subject: [PATCH 03/14] Add the PR reference to this branch history Fixes: https://github.com/spring-projects/spring-integration/pull/11244 Force-push is disabled on this branch, so the reference cannot be added to the message of the first commit; squash it at merge time. From 917aa1da1d0fdbaa98f6eb05d19ea39388d73e1a Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Aug 2026 17:23:53 -0400 Subject: [PATCH 04/14] Demote connection probe failure logs to DEBUG `tryLock()` logged a `WARN` on every failed connection probe. Since `lockInterruptibly()` retries roughly once per second, a prolonged Zookeeper outage produced a continuous stream of per-second `WARN` lines per lock. A `false` from `tryLock()` is a contract outcome, not an anomaly. --- .../integration/zookeeper/lock/ZookeeperLockRegistry.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 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 08bc46029c..53819921e4 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 @@ -339,7 +339,9 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { if (!connected) { future.cancel(true); - LOGGER.warn("No Zookeeper connection to acquire mutex at " + this.path); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("No Zookeeper connection to acquire mutex at " + this.path); + } return false; } else { @@ -349,7 +351,9 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { } catch (@SuppressWarnings("unused") TimeoutException e) { future.cancel(true); - LOGGER.warn("Timed out while checking the Zookeeper connection to acquire mutex at " + this.path); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Timed out while checking the Zookeeper connection to acquire mutex at " + this.path); + } return false; } catch (InterruptedException e) { From 21a1ddf104ef23c0540ba558c4cf2de6b810e8f4 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Aug 2026 18:12:40 -0400 Subject: [PATCH 05/14] Remove the mutexTaskExecutor from ZookeeperLockRegistry The executor existed only to run a `checkExists().forPath("/")` round-trip against the server under a timeout, so a `tryLock(time, unit)` would not block indefinitely on a lost connection. That is a thread (and a ZK round-trip) per lock attempt, and with `queueCapacity` of `0` and an unbounded `maxPoolSize` a sustained outage could grow the pool without a limit. Curator already exposes the connection state without blocking. * check `CuratorZookeeperClient.isConnected()` instead of submitting a probe task; wait for a re-connect via `blockUntilConnected()` (interruptibly, and only within the time requested for `tryLock()`) when not connected * treat a connection loss, an expired session or an operation timeout from `InterProcessMutex.acquire()` as a `false`, matching the `Lock.tryLock()` contract - previously such a failure was hidden behind the probe timeout * deprecate `setMutexTaskExecutor()` with no replacement * `destroy()` is a no-op now: the registry does not own any resource anymore Deprecated for removal since `7.0.6`. --- .../zookeeper/lock/ZookeeperLockRegistry.java | 85 +++++++------------ .../antora/modules/ROOT/pages/zookeeper.adoc | 4 + 2 files changed, 34 insertions(+), 55 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 53819921e4..f2fdc2cdca 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,9 +20,7 @@ import java.util.LinkedHashMap; import java.util.Map; import java.util.Map.Entry; -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 java.util.concurrent.locks.ReentrantLock; @@ -31,13 +29,12 @@ import org.apache.commons.logging.LogFactory; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.recipes.locks.InterProcessMutex; +import org.apache.zookeeper.KeeperException; 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; /** @@ -83,20 +80,6 @@ protected boolean removeEldestEntry(Entry eldest) { private final boolean trackingTime; - private AsyncTaskExecutor mutexTaskExecutor = new ThreadPoolTaskExecutor(); - - { - ThreadPoolTaskExecutor threadPoolTaskExecutor = (ThreadPoolTaskExecutor) this.mutexTaskExecutor; - threadPoolTaskExecutor.setAllowCoreThreadTimeOut(true); - // A queue would make this executor effectively single-threaded, so concurrent `tryLock()` calls - // would serialize behind a single connection check - and time out on a slow or lost connection. - threadPoolTaskExecutor.setQueueCapacity(0); - threadPoolTaskExecutor.setBeanName("ZookeeperLockRegistryExecutor"); - threadPoolTaskExecutor.initialize(); - } - - private boolean mutexTaskExecutorExplicitlySet; - private int cacheCapacity = DEFAULT_CAPACITY; /** @@ -133,18 +116,18 @@ public ZookeeperLockRegistry(CuratorFramework client, KeyToPathStrategy keyToPat /** * 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}. + * connection with Zookeeper. * @param mutexTaskExecutor the executor. * @since 4.2.10 + * @deprecated since 7.0.6 with no replacement. + * The connection state is now determined via a non-blocking + * {@link org.apache.curator.CuratorZookeeperClient#isConnected()}, so no executor + * is involved into locking anymore and this option has no effect. */ + @Deprecated(since = "7.0.6", forRemoval = true) + @SuppressWarnings("unused") public void setMutexTaskExecutor(AsyncTaskExecutor mutexTaskExecutor) { - Assert.notNull(mutexTaskExecutor, "'mutexTaskExecutor' cannot be null"); - ((ExecutorConfigurationSupport) this.mutexTaskExecutor).shutdown(); - this.mutexTaskExecutor = mutexTaskExecutor; - this.mutexTaskExecutorExplicitlySet = true; + LOGGER.warn("The 'mutexTaskExecutor' is not used anymore and will be removed in a future release."); } /** @@ -163,7 +146,7 @@ public Lock obtain(Object lockKey) { ZkLock lock; this.locksLock.lock(); try { - lock = this.locks.computeIfAbsent(path, p -> new ZkLock(this.client, this.mutexTaskExecutor, p)); + lock = this.locks.computeIfAbsent(path, p -> new ZkLock(this.client, p)); } finally { this.locksLock.unlock(); @@ -201,11 +184,13 @@ public void expireUnusedOlderThan(long age) { } + /** + * No-op since version 7.0.6: this registry does not manage any resource of its own anymore. + * The {@link CuratorFramework} client is provided externally, therefore it has to be closed + * by the calling side as well. + */ @Override public void destroy() { - if (!this.mutexTaskExecutorExplicitlySet) { - ((ExecutorConfigurationSupport) this.mutexTaskExecutor).shutdown(); - } } /** @@ -260,16 +245,13 @@ private static final class ZkLock implements Lock { private final InterProcessMutex mutex; - private final AsyncTaskExecutor mutexTaskExecutor; - private final String path; private long lastUsed; - ZkLock(CuratorFramework client, AsyncTaskExecutor mutexTaskExecutor, String path) { + ZkLock(CuratorFramework client, String path) { this.client = client; this.mutex = new InterProcessMutex(client, path); - this.mutexTaskExecutor = mutexTaskExecutor; this.path = path; } @@ -320,39 +302,32 @@ public boolean tryLock() { @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { - Future future = null; try { long startTime = System.currentTimeMillis(); - - future = this.mutexTaskExecutor.submit(() -> { - try { - return ZkLock.this.client.checkExists().forPath("/") != null; - } - catch (Exception e) { - throw new IllegalStateException(e); - } - }); - long waitTime = unit.toMillis(time); - boolean connected = future.get(waitTime, TimeUnit.MILLISECONDS); + // The non-blocking state check first; only an unconnected client is waited for, + // interruptibly and within the requested time, to let a re-connect happen. + if (!this.client.getZookeeperClient().isConnected() && + !this.client.blockUntilConnected((int) Math.min(waitTime, Integer.MAX_VALUE), + TimeUnit.MILLISECONDS)) { - if (!connected) { - future.cancel(true); if (LOGGER.isDebugEnabled()) { LOGGER.debug("No Zookeeper connection to acquire mutex at " + this.path); } return false; } - else { - waitTime = Math.max(0, waitTime - (System.currentTimeMillis() - startTime)); - return this.mutex.acquire(waitTime, TimeUnit.MILLISECONDS); - } + + waitTime = Math.max(0, waitTime - (System.currentTimeMillis() - startTime)); + return this.mutex.acquire(waitTime, TimeUnit.MILLISECONDS); } - catch (@SuppressWarnings("unused") TimeoutException e) { - future.cancel(true); + catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException | + KeeperException.OperationTimeoutException e) { + + // The connection may be lost between the state check and the acquisition: + // this is a `false` for the `Lock.tryLock()` contract, not an error. if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Timed out while checking the Zookeeper connection to acquire mutex at " + this.path); + LOGGER.debug("Lost the Zookeeper connection to acquire mutex at " + this.path, e); } return false; } diff --git a/src/reference/antora/modules/ROOT/pages/zookeeper.adoc b/src/reference/antora/modules/ROOT/pages/zookeeper.adoc index 3c63e49232..bd6d080069 100644 --- a/src/reference/antora/modules/ROOT/pages/zookeeper.adoc +++ b/src/reference/antora/modules/ROOT/pages/zookeeper.adoc @@ -84,6 +84,10 @@ For unbounded strategies (such as the default), you need to periodically invoke Starting with version 5.5.6, the `ZookeeperLockRegistry` is support automatically clean up cache for ZkLock in `ZookeeperLockRegistry.locks` via `ZookeeperLockRegistry.setCacheCapacity()`. See its JavaDocs for more information. +Starting with version 7.0.6, the `ZookeeperLockRegistry` checks the Zookeeper connection state via a non-blocking `CuratorZookeeperClient.isConnected()` instead of an asynchronous `checkExists()` call against the server. +Therefore, no internal `TaskExecutor` is involved into locking anymore, and the `setMutexTaskExecutor()` is deprecated with no replacement. +An unconnected client is still waited for - interruptibly and within the time requested for a `Lock.tryLock(long, TimeUnit)` - to let a re-connect happen. + [[zk-leadership]] == Zookeeper Leadership Event Handling From a627d72b553d1ffe5da804820f87060333f324f9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Aug 2026 10:34:37 -0400 Subject: [PATCH 06/14] Polish the ZookeeperLockRegistry connection state check * re-check `CuratorZookeeperClient.isConnected()` after a successful `blockUntilConnected()`: the latter is served from the `ConnectionStateManager` alone, so it may return `true` immediately, without consuming any of the requested time, while the socket is already gone. An acquisition against such a stale state blocks in the Curator retry loop far beyond the `tryLock()` time * check the interrupt status on entry into `lockInterruptibly()`: per the `Lock` contract an already interrupted thread must not attempt an acquisition at all; extract the check into a `checkInterruption()` * validate the argument in the deprecated `setMutexTaskExecutor()` instead of suppressing the unused warning * document in the JavaDocs and the reference manual that the `tryLock()` time is not a hard bound: the Curator retry loop waits for a connection on its own * test that `tryLock()` does not exceed the requested time by orders of magnitude, and that `lockInterruptibly()` reacts to an interrupt while the Zookeeper server is stopped --- build.gradle | 4 +- .../zookeeper/lock/ZookeeperLockRegistry.java | 50 ++++++++++++++----- .../zookeeper/lock/ZkLockRegistryTests.java | 29 +++++++++++ .../antora/modules/ROOT/pages/zookeeper.adoc | 12 +++-- 4 files changed, 79 insertions(+), 16 deletions(-) diff --git a/build.gradle b/build.gradle index 0fd2de3d2a..c876f5d104 100644 --- a/build.gradle +++ b/build.gradle @@ -231,8 +231,10 @@ subprojects { subproject -> systemProperty 'junit.jupiter.execution.timeout.default', '30 m' } else { - // Fail a stuck test with a stack trace of its thread + // Interrupt a stuck test and report a stack trace of its thread // instead of stalling the whole build with no output at all. + // The default `SAME_THREAD` mode still waits for the invocation to return, + // so a test wedged in a non-interruptible call is not aborted. systemProperty 'junit.jupiter.execution.timeout.default', '10 m' } 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 f2fdc2cdca..5d882393a4 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 @@ -27,6 +27,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.curator.CuratorZookeeperClient; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.recipes.locks.InterProcessMutex; import org.apache.zookeeper.KeeperException; @@ -40,6 +41,17 @@ /** * {@link ExpirableLockRegistry} implementation using Zookeeper, or more specifically, * Curator {@link InterProcessMutex}. + *

+ * The {@link Lock#tryLock(long, TimeUnit)} of the locks from this registry bounds only the wait + * for a Zookeeper connection and for the mutex itself. + * The acquisition is delegated to an {@link InterProcessMutex} which internally goes through + * a Curator retry loop, and that loop waits for a connection on its own - up to the + * {@code connectionTimeoutMs} of the {@link CuratorFramework} plus its {@code RetryPolicy} budget. + * Therefore, when the connection is lost silently, e.g. a network partition where the socket is not + * closed, the requested time may be exceeded: the Zookeeper client reports itself as connected until + * its own read timeout expires (two-thirds of the session timeout). + * The {@code connectionTimeoutMs} and {@code RetryPolicy} have to be configured below the expected + * lock timeout when a tight bound is essential. * * @author Gary Russell * @author Artem Bilan @@ -120,13 +132,12 @@ public ZookeeperLockRegistry(CuratorFramework client, KeyToPathStrategy keyToPat * @param mutexTaskExecutor the executor. * @since 4.2.10 * @deprecated since 7.0.6 with no replacement. - * The connection state is now determined via a non-blocking - * {@link org.apache.curator.CuratorZookeeperClient#isConnected()}, so no executor - * is involved into locking anymore and this option has no effect. + * The connection is now awaited via a {@link CuratorFramework#blockUntilConnected(int, TimeUnit)}, + * so no executor is involved in locking anymore and this option has no effect. */ @Deprecated(since = "7.0.6", forRemoval = true) - @SuppressWarnings("unused") public void setMutexTaskExecutor(AsyncTaskExecutor mutexTaskExecutor) { + Assert.notNull(mutexTaskExecutor, "'mutexTaskExecutor' cannot be null"); LOGGER.warn("The 'mutexTaskExecutor' is not used anymore and will be removed in a future release."); } @@ -275,20 +286,26 @@ public void lock() { @Override public void lockInterruptibly() throws InterruptedException { + // The Lock contract: the interrupt status set on entry means no acquisition attempt at all. + checkInterruption(); // this is a bit ugly, but... while (!tryLock(1, TimeUnit.SECONDS)) { // The tryLock() above may return 'false' without blocking at all, e.g. when Zookeeper is not reachable. // Therefore, the interrupt status has to be checked explicitly // to avoid an endless, silent spin in this loop. - if (Thread.interrupted()) { - throw new InterruptedException("Interrupted while acquiring mutex at " + this.path); - } + checkInterruption(); if (LOGGER.isDebugEnabled()) { 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 public boolean tryLock() { try { @@ -306,11 +323,20 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { long startTime = System.currentTimeMillis(); long waitTime = unit.toMillis(time); - // The non-blocking state check first; only an unconnected client is waited for, - // interruptibly and within the requested time, to let a re-connect happen. - if (!this.client.getZookeeperClient().isConnected() && - !this.client.blockUntilConnected((int) Math.min(waitTime, Integer.MAX_VALUE), - TimeUnit.MILLISECONDS)) { + // Both Curator's state machines have to agree before the acquisition is attempted: + // the `CuratorZookeeperClient` reacts to a closed socket immediately, while its + // `ConnectionStateManager` may still report a connection for a while. + // An acquisition against such a stale state blocks in the Curator retry loop + // far beyond the requested time. + // Hence the second `isConnected()` after a successful `blockUntilConnected()`: + // the latter is served from the `ConnectionStateManager` alone, so it may return `true` + // immediately, without consuming any of the `waitTime`, while the socket is already gone. + // The same re-check also covers a connection which has flapped back down + // right after `blockUntilConnected()` has unblocked. + CuratorZookeeperClient zookeeperClient = this.client.getZookeeperClient(); + if (!zookeeperClient.isConnected() && + (!this.client.blockUntilConnected((int) Math.min(waitTime, Integer.MAX_VALUE), + TimeUnit.MILLISECONDS) || !zookeeperClient.isConnected())) { if (LOGGER.isDebugEnabled()) { LOGGER.debug("No Zookeeper connection to acquire mutex at " + this.path); 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 cf845a82ca..04cae4adb3 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,6 +25,7 @@ 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; @@ -317,9 +318,37 @@ public void voidLockFailsWhenServerDown() throws Exception { Lock lock2 = registry.obtain("bar"); + long startTime = System.currentTimeMillis(); + assertThat(lock2.tryLock(1, TimeUnit.SECONDS)) .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); + assertThat(System.currentTimeMillis() - startTime) + .as("The tryLock() must not block far beyond the time requested!").isLessThan(10_000); + + Lock lock4 = registry.obtain("interruptible"); + CountDownLatch lockAttemptLatch = new CountDownLatch(1); + AtomicReference lockException = new AtomicReference<>(); + + Thread lockThread = new Thread(() -> { + try { + lock4.lockInterruptibly(); + } + catch (Exception ex) { + lockException.set(ex); + } + finally { + lockAttemptLatch.countDown(); + } + }); + lockThread.start(); + + lockThread.interrupt(); + + assertThat(lockAttemptLatch.await(10, TimeUnit.SECONDS)) + .as("The lockInterruptibly() must not spin forever with zookeeper server stopped!").isTrue(); + assertThat(lockException.get()).isInstanceOf(InterruptedException.class); + server.restart(); assertThat(lock2.tryLock(10, TimeUnit.SECONDS)) diff --git a/src/reference/antora/modules/ROOT/pages/zookeeper.adoc b/src/reference/antora/modules/ROOT/pages/zookeeper.adoc index bd6d080069..182a184acc 100644 --- a/src/reference/antora/modules/ROOT/pages/zookeeper.adoc +++ b/src/reference/antora/modules/ROOT/pages/zookeeper.adoc @@ -84,9 +84,15 @@ For unbounded strategies (such as the default), you need to periodically invoke Starting with version 5.5.6, the `ZookeeperLockRegistry` is support automatically clean up cache for ZkLock in `ZookeeperLockRegistry.locks` via `ZookeeperLockRegistry.setCacheCapacity()`. See its JavaDocs for more information. -Starting with version 7.0.6, the `ZookeeperLockRegistry` checks the Zookeeper connection state via a non-blocking `CuratorZookeeperClient.isConnected()` instead of an asynchronous `checkExists()` call against the server. -Therefore, no internal `TaskExecutor` is involved into locking anymore, and the `setMutexTaskExecutor()` is deprecated with no replacement. -An unconnected client is still waited for - interruptibly and within the time requested for a `Lock.tryLock(long, TimeUnit)` - to let a re-connect happen. +Starting with version 7.0.6, the `ZookeeperLockRegistry` awaits the Zookeeper connection via a `CuratorFramework.blockUntilConnected()` instead of an asynchronous `checkExists()` call against the server. +The wait happens interruptibly and only within the time requested for a `Lock.tryLock(long, TimeUnit)`. +Therefore, no internal `TaskExecutor` is involved in locking anymore, and the `setMutexTaskExecutor()` is deprecated with no replacement. +In addition, a connection loss, an expired session or an operation timeout from the mutex acquisition is now reported as a `false` from the `Lock.tryLock()` instead of an exception. + +IMPORTANT: The `Lock.tryLock(long, TimeUnit)` bounds only the wait for a connection and for the mutex itself. +The acquisition is delegated to a Curator `InterProcessMutex` which goes through a retry loop, and that loop waits for a connection on its own - up to the `connectionTimeoutMs` of the `CuratorFramework` plus its `RetryPolicy` budget. +Therefore, when the connection is lost silently, for example, a network partition where the socket is not closed, the requested time may be exceeded: the Zookeeper client reports itself as connected until its own read timeout expires (two-thirds of the session timeout). +Configure the `connectionTimeoutMs` and `RetryPolicy` of the `CuratorFramework` below the expected lock timeout when a tight bound is essential. [[zk-leadership]] == Zookeeper Leadership Event Handling From 9f85bc44098603ef1812d203a486e2eea00f09bc Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Aug 2026 11:32:58 -0400 Subject: [PATCH 07/14] Roll back the mutexTaskExecutor deprecation The `setMutexTaskExecutor()` deprecation, the removal of the executor and the `blockUntilConnected()` rewrite of the connection check are an API and behavior change, not a bug fix, and this branch is meant to be cherry-picked to `7.1.x` and `7.0.x`. Every existing `@Deprecated(since = ...)` in the tree is a minor version; there is no precedent for deprecating in a patch. * restore the `mutexTaskExecutor`, its setter, `destroy()` and the asynchronous `checkExists()` connection probe exactly as they are on `main`, including the default executor configuration - the `queueCapacity = 0` from the first commit is dropped as well, so the probe serialization stays pre-existing and untouched * drop the `KeeperException` to `false` mapping in `tryLock()`: also a behavior change rather than a fix * drop the `blockUntilConnected()` state check, the class JavaDoc caveat and the `7.0.6` paragraph in the reference manual along with it What remains is the fix itself: the interrupt handling in `lockInterruptibly()`, the `Math.max(0, ...)` clamp of the recomputed `waitTime`, the `DEBUG` logging, the test isolation and the build test timeout. Note that the interrupt checks are a guard, not the mechanism: the restored `tryLock()` spends its wait inside `Future.get()`, which throws `InterruptedException` on its own, so `lockInterruptibly()` already terminated on an interrupt. Verified by removing the in-loop check - the test still passes. The "busy-spin hang" of the first commit was therefore the test isolation issue alone, and the branch title overclaims. --- build.gradle | 8 +- .../recipes/locks/LockInternals.java | 354 ++++++++++++++++++ .../zookeeper/lock/ZookeeperLockRegistry.java | 107 +++--- .../zookeeper/lock/ZkLockRegistryTests.java | 11 +- .../antora/modules/ROOT/pages/zookeeper.adoc | 10 - 5 files changed, 423 insertions(+), 67 deletions(-) create mode 100644 cur/org/apache/curator/framework/recipes/locks/LockInternals.java diff --git a/build.gradle b/build.gradle index c876f5d104..fd033f9d10 100644 --- a/build.gradle +++ b/build.gradle @@ -231,10 +231,10 @@ subprojects { subproject -> systemProperty 'junit.jupiter.execution.timeout.default', '30 m' } else { - // Interrupt a stuck test and report a stack trace of its thread - // instead of stalling the whole build with no output at all. - // The default `SAME_THREAD` mode still waits for the invocation to return, - // so a test wedged in a non-interruptible call is not aborted. + // 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' } diff --git a/cur/org/apache/curator/framework/recipes/locks/LockInternals.java b/cur/org/apache/curator/framework/recipes/locks/LockInternals.java new file mode 100644 index 0000000000..a22bfb1063 --- /dev/null +++ b/cur/org/apache/curator/framework/recipes/locks/LockInternals.java @@ -0,0 +1,354 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.curator.framework.recipes.locks; + +import com.google.common.base.Function; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Iterables; +import com.google.common.collect.Lists; +import org.apache.curator.RetryLoop; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.WatcherRemoveCuratorFramework; +import org.apache.curator.framework.api.CuratorWatcher; +import org.apache.curator.framework.imps.CuratorFrameworkState; +import org.apache.curator.utils.PathUtils; +import org.apache.curator.utils.ThreadUtils; +import org.apache.curator.utils.ZKPaths; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +public class LockInternals +{ + private final WatcherRemoveCuratorFramework client; + private final String path; + private final String basePath; + private final LockInternalsDriver driver; + private final String lockName; + private final AtomicReference revocable = new AtomicReference(null); + private final CuratorWatcher revocableWatcher = new CuratorWatcher() + { + @Override + public void process(WatchedEvent event) throws Exception + { + if ( event.getType() == Watcher.Event.EventType.NodeDataChanged ) + { + checkRevocableWatcher(event.getPath()); + } + } + }; + + private final Watcher watcher = new Watcher() + { + @Override + public void process(WatchedEvent event) + { + client.postSafeNotify(LockInternals.this); + } + }; + + private volatile int maxLeases; + + static final byte[] REVOKE_MESSAGE = "__REVOKE__".getBytes(); + + /** + * Attempt to delete the lock node so that sequence numbers get reset + * + * @throws Exception errors + */ + public void clean() throws Exception + { + try + { + client.delete().forPath(basePath); + } + catch ( KeeperException.BadVersionException ignore ) + { + // ignore - another thread/process got the lock + } + catch ( KeeperException.NotEmptyException ignore ) + { + // ignore - other threads/processes are waiting + } + } + + LockInternals(CuratorFramework client, LockInternalsDriver driver, String path, String lockName, int maxLeases) + { + this.driver = driver; + this.lockName = lockName; + this.maxLeases = maxLeases; + + this.client = client.newWatcherRemoveCuratorFramework(); + this.basePath = PathUtils.validatePath(path); + this.path = ZKPaths.makePath(path, lockName); + } + + synchronized void setMaxLeases(int maxLeases) + { + this.maxLeases = maxLeases; + notifyAll(); + } + + void makeRevocable(RevocationSpec entry) + { + revocable.set(entry); + } + + final void releaseLock(String lockPath) throws Exception + { + client.removeWatchers(); + revocable.set(null); + deleteOurPath(lockPath); + } + + CuratorFramework getClient() + { + return client; + } + + public static Collection getParticipantNodes(CuratorFramework client, final String basePath, String lockName, LockInternalsSorter sorter) throws Exception + { + List names = getSortedChildren(client, basePath, lockName, sorter); + Iterable transformed = Iterables.transform + ( + names, + new Function() + { + @Override + public String apply(String name) + { + return ZKPaths.makePath(basePath, name); + } + } + ); + return ImmutableList.copyOf(transformed); + } + + public static List getSortedChildren(CuratorFramework client, String basePath, final String lockName, final LockInternalsSorter sorter) throws Exception + { + try + { + List children = client.getChildren().forPath(basePath); + List sortedList = Lists.newArrayList(children); + Collections.sort + ( + sortedList, + new Comparator() + { + @Override + public int compare(String lhs, String rhs) + { + return sorter.fixForSorting(lhs, lockName).compareTo(sorter.fixForSorting(rhs, lockName)); + } + } + ); + return sortedList; + } + catch ( KeeperException.NoNodeException ignore ) + { + return Collections.emptyList(); + } + } + + public static List getSortedChildren(final String lockName, final LockInternalsSorter sorter, List children) + { + List sortedList = Lists.newArrayList(children); + Collections.sort + ( + sortedList, + new Comparator() + { + @Override + public int compare(String lhs, String rhs) + { + return sorter.fixForSorting(lhs, lockName).compareTo(sorter.fixForSorting(rhs, lockName)); + } + } + ); + return sortedList; + } + + List getSortedChildren() throws Exception + { + return getSortedChildren(client, basePath, lockName, driver); + } + + String getLockName() + { + return lockName; + } + + LockInternalsDriver getDriver() + { + return driver; + } + + String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception + { + final long startMillis = System.currentTimeMillis(); + final Long millisToWait = (unit != null) ? unit.toMillis(time) : null; + final byte[] localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes; + int retryCount = 0; + + String ourPath = null; + boolean hasTheLock = false; + boolean isDone = false; + while ( !isDone ) + { + isDone = true; + + try + { + ourPath = driver.createsTheLock(client, path, localLockNodeBytes); + hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath); + } + catch ( KeeperException.NoNodeException e ) + { + // gets thrown by StandardLockInternalsDriver when it can't find the lock node + // this can happen when the session expires, etc. So, if the retry allows, just try it all again + if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) ) + { + isDone = false; + } + else + { + throw e; + } + } + } + + if ( hasTheLock ) + { + return ourPath; + } + + return null; + } + + private void checkRevocableWatcher(String path) throws Exception + { + RevocationSpec entry = revocable.get(); + if ( entry != null ) + { + try + { + byte[] bytes = client.getData().usingWatcher(revocableWatcher).forPath(path); + if ( Arrays.equals(bytes, REVOKE_MESSAGE) ) + { + entry.getExecutor().execute(entry.getRunnable()); + } + } + catch ( KeeperException.NoNodeException ignore ) + { + // ignore + } + } + } + + private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception + { + boolean haveTheLock = false; + boolean doDelete = false; + try + { + if ( revocable.get() != null ) + { + client.getData().usingWatcher(revocableWatcher).forPath(ourPath); + } + + while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock ) + { + List children = getSortedChildren(); + String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash + + PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases); + if ( predicateResults.getsTheLock() ) + { + haveTheLock = true; + } + else + { + String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch(); + + synchronized(this) + { + try + { + // use getData() instead of exists() to avoid leaving unneeded watchers which is a type of resource leak + client.getData().usingWatcher(watcher).forPath(previousSequencePath); + if ( millisToWait != null ) + { + millisToWait -= (System.currentTimeMillis() - startMillis); + startMillis = System.currentTimeMillis(); + if ( millisToWait <= 0 ) + { + doDelete = true; // timed out - delete our node + break; + } + + wait(millisToWait); + } + else + { + wait(); + } + } + catch ( KeeperException.NoNodeException e ) + { + // it has been deleted (i.e. lock released). Try to acquire again + } + } + } + } + } + catch ( Exception e ) + { + ThreadUtils.checkInterrupted(e); + doDelete = true; + throw e; + } + finally + { + if ( doDelete ) + { + deleteOurPath(ourPath); + } + } + return haveTheLock; + } + + private void deleteOurPath(String ourPath) throws Exception + { + try + { + client.delete().guaranteed().forPath(ourPath); + } + catch ( KeeperException.NoNodeException e ) + { + // ignore - already deleted (possibly expired session, etc.) + } + } +} 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 5d882393a4..e3a41cd4e2 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,38 +20,29 @@ import java.util.LinkedHashMap; import java.util.Map; import java.util.Map.Entry; +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 java.util.concurrent.locks.ReentrantLock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.apache.curator.CuratorZookeeperClient; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.recipes.locks.InterProcessMutex; -import org.apache.zookeeper.KeeperException; 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; /** * {@link ExpirableLockRegistry} implementation using Zookeeper, or more specifically, * Curator {@link InterProcessMutex}. - *

- * The {@link Lock#tryLock(long, TimeUnit)} of the locks from this registry bounds only the wait - * for a Zookeeper connection and for the mutex itself. - * The acquisition is delegated to an {@link InterProcessMutex} which internally goes through - * a Curator retry loop, and that loop waits for a connection on its own - up to the - * {@code connectionTimeoutMs} of the {@link CuratorFramework} plus its {@code RetryPolicy} budget. - * Therefore, when the connection is lost silently, e.g. a network partition where the socket is not - * closed, the requested time may be exceeded: the Zookeeper client reports itself as connected until - * its own read timeout expires (two-thirds of the session timeout). - * The {@code connectionTimeoutMs} and {@code RetryPolicy} have to be configured below the expected - * lock timeout when a tight bound is essential. * * @author Gary Russell * @author Artem Bilan @@ -92,6 +83,17 @@ protected boolean removeEldestEntry(Entry eldest) { 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; + private int cacheCapacity = DEFAULT_CAPACITY; /** @@ -128,17 +130,18 @@ public ZookeeperLockRegistry(CuratorFramework client, KeyToPathStrategy keyToPat /** * Set an {@link AsyncTaskExecutor} to use when establishing (and testing) the - * connection with Zookeeper. + * 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 - * @deprecated since 7.0.6 with no replacement. - * The connection is now awaited via a {@link CuratorFramework#blockUntilConnected(int, TimeUnit)}, - * so no executor is involved in locking anymore and this option has no effect. */ - @Deprecated(since = "7.0.6", forRemoval = true) public void setMutexTaskExecutor(AsyncTaskExecutor mutexTaskExecutor) { Assert.notNull(mutexTaskExecutor, "'mutexTaskExecutor' cannot be null"); - LOGGER.warn("The 'mutexTaskExecutor' is not used anymore and will be removed in a future release."); + ((ExecutorConfigurationSupport) this.mutexTaskExecutor).shutdown(); + this.mutexTaskExecutor = mutexTaskExecutor; + this.mutexTaskExecutorExplicitlySet = true; } /** @@ -157,7 +160,7 @@ public Lock obtain(Object lockKey) { ZkLock lock; this.locksLock.lock(); try { - lock = this.locks.computeIfAbsent(path, p -> new ZkLock(this.client, p)); + lock = this.locks.computeIfAbsent(path, p -> new ZkLock(this.client, this.mutexTaskExecutor, p)); } finally { this.locksLock.unlock(); @@ -195,13 +198,11 @@ public void expireUnusedOlderThan(long age) { } - /** - * No-op since version 7.0.6: this registry does not manage any resource of its own anymore. - * The {@link CuratorFramework} client is provided externally, therefore it has to be closed - * by the calling side as well. - */ @Override public void destroy() { + if (!this.mutexTaskExecutorExplicitlySet) { + ((ExecutorConfigurationSupport) this.mutexTaskExecutor).shutdown(); + } } /** @@ -256,13 +257,16 @@ private static final class ZkLock implements Lock { private final InterProcessMutex mutex; + private final AsyncTaskExecutor mutexTaskExecutor; + private final String path; private long lastUsed; - ZkLock(CuratorFramework client, String path) { + ZkLock(CuratorFramework client, AsyncTaskExecutor mutexTaskExecutor, String path) { this.client = client; this.mutex = new InterProcessMutex(client, path); + this.mutexTaskExecutor = mutexTaskExecutor; this.path = path; } @@ -290,9 +294,9 @@ public void lockInterruptibly() throws InterruptedException { checkInterruption(); // this is a bit ugly, but... while (!tryLock(1, TimeUnit.SECONDS)) { - // The tryLock() above may return 'false' without blocking at all, e.g. when Zookeeper is not reachable. - // Therefore, the interrupt status has to be checked explicitly - // to avoid an endless, silent spin in this loop. + // 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(); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Mutex at " + this.path + " is not acquired yet; retrying..."); @@ -319,41 +323,40 @@ public boolean tryLock() { @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { + Future future = null; try { long startTime = System.currentTimeMillis(); + + future = this.mutexTaskExecutor.submit(() -> { + try { + return ZkLock.this.client.checkExists().forPath("/") != null; + } + catch (Exception e) { + throw new IllegalStateException(e); + } + }); + long waitTime = unit.toMillis(time); - // Both Curator's state machines have to agree before the acquisition is attempted: - // the `CuratorZookeeperClient` reacts to a closed socket immediately, while its - // `ConnectionStateManager` may still report a connection for a while. - // An acquisition against such a stale state blocks in the Curator retry loop - // far beyond the requested time. - // Hence the second `isConnected()` after a successful `blockUntilConnected()`: - // the latter is served from the `ConnectionStateManager` alone, so it may return `true` - // immediately, without consuming any of the `waitTime`, while the socket is already gone. - // The same re-check also covers a connection which has flapped back down - // right after `blockUntilConnected()` has unblocked. - CuratorZookeeperClient zookeeperClient = this.client.getZookeeperClient(); - if (!zookeeperClient.isConnected() && - (!this.client.blockUntilConnected((int) Math.min(waitTime, Integer.MAX_VALUE), - TimeUnit.MILLISECONDS) || !zookeeperClient.isConnected())) { + boolean connected = future.get(waitTime, TimeUnit.MILLISECONDS); + if (!connected) { + future.cancel(true); if (LOGGER.isDebugEnabled()) { LOGGER.debug("No Zookeeper connection to acquire mutex at " + this.path); } return false; } - - waitTime = Math.max(0, waitTime - (System.currentTimeMillis() - startTime)); - return this.mutex.acquire(waitTime, TimeUnit.MILLISECONDS); + else { + // A slow connection check must not pass a negative deadline to the mutex. + waitTime = Math.max(0, waitTime - (System.currentTimeMillis() - startTime)); + return this.mutex.acquire(waitTime, TimeUnit.MILLISECONDS); + } } - catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException | - KeeperException.OperationTimeoutException e) { - - // The connection may be lost between the state check and the acquisition: - // this is a `false` for the `Lock.tryLock()` contract, not an error. + catch (@SuppressWarnings("unused") TimeoutException e) { + future.cancel(true); if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Lost the Zookeeper connection to acquire mutex at " + this.path, e); + LOGGER.debug("Timed out while checking the Zookeeper connection to acquire mutex at " + this.path); } return false; } 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 04cae4adb3..cc1df23cbb 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 @@ -324,14 +324,18 @@ public void voidLockFailsWhenServerDown() throws Exception { .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); assertThat(System.currentTimeMillis() - startTime) - .as("The tryLock() must not block far beyond the time requested!").isLessThan(10_000); + .as("The tryLock() must not block far beyond the time requested!").isLessThan(3_000); + // The `lockInterruptibly()` retries until it succeeds, so with the server down + // an interrupt is the only way out: it must not be swallowed by the retry loop. Lock lock4 = registry.obtain("interruptible"); + CountDownLatch lockAttemptStartedLatch = new CountDownLatch(1); CountDownLatch lockAttemptLatch = new CountDownLatch(1); AtomicReference lockException = new AtomicReference<>(); Thread lockThread = new Thread(() -> { try { + lockAttemptStartedLatch.countDown(); lock4.lockInterruptibly(); } catch (Exception ex) { @@ -343,12 +347,17 @@ public void voidLockFailsWhenServerDown() throws Exception { }); lockThread.start(); + assertThat(lockAttemptStartedLatch.await(10, TimeUnit.SECONDS)).isTrue(); lockThread.interrupt(); assertThat(lockAttemptLatch.await(10, TimeUnit.SECONDS)) .as("The lockInterruptibly() must not spin forever with zookeeper server stopped!").isTrue(); assertThat(lockException.get()).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(lock2.tryLock(10, TimeUnit.SECONDS)) diff --git a/src/reference/antora/modules/ROOT/pages/zookeeper.adoc b/src/reference/antora/modules/ROOT/pages/zookeeper.adoc index 182a184acc..3c63e49232 100644 --- a/src/reference/antora/modules/ROOT/pages/zookeeper.adoc +++ b/src/reference/antora/modules/ROOT/pages/zookeeper.adoc @@ -84,16 +84,6 @@ For unbounded strategies (such as the default), you need to periodically invoke Starting with version 5.5.6, the `ZookeeperLockRegistry` is support automatically clean up cache for ZkLock in `ZookeeperLockRegistry.locks` via `ZookeeperLockRegistry.setCacheCapacity()`. See its JavaDocs for more information. -Starting with version 7.0.6, the `ZookeeperLockRegistry` awaits the Zookeeper connection via a `CuratorFramework.blockUntilConnected()` instead of an asynchronous `checkExists()` call against the server. -The wait happens interruptibly and only within the time requested for a `Lock.tryLock(long, TimeUnit)`. -Therefore, no internal `TaskExecutor` is involved in locking anymore, and the `setMutexTaskExecutor()` is deprecated with no replacement. -In addition, a connection loss, an expired session or an operation timeout from the mutex acquisition is now reported as a `false` from the `Lock.tryLock()` instead of an exception. - -IMPORTANT: The `Lock.tryLock(long, TimeUnit)` bounds only the wait for a connection and for the mutex itself. -The acquisition is delegated to a Curator `InterProcessMutex` which goes through a retry loop, and that loop waits for a connection on its own - up to the `connectionTimeoutMs` of the `CuratorFramework` plus its `RetryPolicy` budget. -Therefore, when the connection is lost silently, for example, a network partition where the socket is not closed, the requested time may be exceeded: the Zookeeper client reports itself as connected until its own read timeout expires (two-thirds of the session timeout). -Configure the `connectionTimeoutMs` and `RetryPolicy` of the `CuratorFramework` below the expected lock timeout when a tight bound is essential. - [[zk-leadership]] == Zookeeper Leadership Event Handling From d6d801ae7813f8a06e733f90409fd7972520e31a Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Aug 2026 11:34:46 -0400 Subject: [PATCH 08/14] Remove accidentally committed Curator sources An extracted `curator-recipes` source file landed in the repository root while verifying the `InterProcessMutex` timeout behavior. --- .../recipes/locks/LockInternals.java | 354 ------------------ 1 file changed, 354 deletions(-) delete mode 100644 cur/org/apache/curator/framework/recipes/locks/LockInternals.java diff --git a/cur/org/apache/curator/framework/recipes/locks/LockInternals.java b/cur/org/apache/curator/framework/recipes/locks/LockInternals.java deleted file mode 100644 index a22bfb1063..0000000000 --- a/cur/org/apache/curator/framework/recipes/locks/LockInternals.java +++ /dev/null @@ -1,354 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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.apache.curator.framework.recipes.locks; - -import com.google.common.base.Function; -import com.google.common.collect.ImmutableList; -import com.google.common.collect.Iterables; -import com.google.common.collect.Lists; -import org.apache.curator.RetryLoop; -import org.apache.curator.framework.CuratorFramework; -import org.apache.curator.framework.WatcherRemoveCuratorFramework; -import org.apache.curator.framework.api.CuratorWatcher; -import org.apache.curator.framework.imps.CuratorFrameworkState; -import org.apache.curator.utils.PathUtils; -import org.apache.curator.utils.ThreadUtils; -import org.apache.curator.utils.ZKPaths; -import org.apache.zookeeper.KeeperException; -import org.apache.zookeeper.WatchedEvent; -import org.apache.zookeeper.Watcher; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.Comparator; -import java.util.List; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; - -public class LockInternals -{ - private final WatcherRemoveCuratorFramework client; - private final String path; - private final String basePath; - private final LockInternalsDriver driver; - private final String lockName; - private final AtomicReference revocable = new AtomicReference(null); - private final CuratorWatcher revocableWatcher = new CuratorWatcher() - { - @Override - public void process(WatchedEvent event) throws Exception - { - if ( event.getType() == Watcher.Event.EventType.NodeDataChanged ) - { - checkRevocableWatcher(event.getPath()); - } - } - }; - - private final Watcher watcher = new Watcher() - { - @Override - public void process(WatchedEvent event) - { - client.postSafeNotify(LockInternals.this); - } - }; - - private volatile int maxLeases; - - static final byte[] REVOKE_MESSAGE = "__REVOKE__".getBytes(); - - /** - * Attempt to delete the lock node so that sequence numbers get reset - * - * @throws Exception errors - */ - public void clean() throws Exception - { - try - { - client.delete().forPath(basePath); - } - catch ( KeeperException.BadVersionException ignore ) - { - // ignore - another thread/process got the lock - } - catch ( KeeperException.NotEmptyException ignore ) - { - // ignore - other threads/processes are waiting - } - } - - LockInternals(CuratorFramework client, LockInternalsDriver driver, String path, String lockName, int maxLeases) - { - this.driver = driver; - this.lockName = lockName; - this.maxLeases = maxLeases; - - this.client = client.newWatcherRemoveCuratorFramework(); - this.basePath = PathUtils.validatePath(path); - this.path = ZKPaths.makePath(path, lockName); - } - - synchronized void setMaxLeases(int maxLeases) - { - this.maxLeases = maxLeases; - notifyAll(); - } - - void makeRevocable(RevocationSpec entry) - { - revocable.set(entry); - } - - final void releaseLock(String lockPath) throws Exception - { - client.removeWatchers(); - revocable.set(null); - deleteOurPath(lockPath); - } - - CuratorFramework getClient() - { - return client; - } - - public static Collection getParticipantNodes(CuratorFramework client, final String basePath, String lockName, LockInternalsSorter sorter) throws Exception - { - List names = getSortedChildren(client, basePath, lockName, sorter); - Iterable transformed = Iterables.transform - ( - names, - new Function() - { - @Override - public String apply(String name) - { - return ZKPaths.makePath(basePath, name); - } - } - ); - return ImmutableList.copyOf(transformed); - } - - public static List getSortedChildren(CuratorFramework client, String basePath, final String lockName, final LockInternalsSorter sorter) throws Exception - { - try - { - List children = client.getChildren().forPath(basePath); - List sortedList = Lists.newArrayList(children); - Collections.sort - ( - sortedList, - new Comparator() - { - @Override - public int compare(String lhs, String rhs) - { - return sorter.fixForSorting(lhs, lockName).compareTo(sorter.fixForSorting(rhs, lockName)); - } - } - ); - return sortedList; - } - catch ( KeeperException.NoNodeException ignore ) - { - return Collections.emptyList(); - } - } - - public static List getSortedChildren(final String lockName, final LockInternalsSorter sorter, List children) - { - List sortedList = Lists.newArrayList(children); - Collections.sort - ( - sortedList, - new Comparator() - { - @Override - public int compare(String lhs, String rhs) - { - return sorter.fixForSorting(lhs, lockName).compareTo(sorter.fixForSorting(rhs, lockName)); - } - } - ); - return sortedList; - } - - List getSortedChildren() throws Exception - { - return getSortedChildren(client, basePath, lockName, driver); - } - - String getLockName() - { - return lockName; - } - - LockInternalsDriver getDriver() - { - return driver; - } - - String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception - { - final long startMillis = System.currentTimeMillis(); - final Long millisToWait = (unit != null) ? unit.toMillis(time) : null; - final byte[] localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes; - int retryCount = 0; - - String ourPath = null; - boolean hasTheLock = false; - boolean isDone = false; - while ( !isDone ) - { - isDone = true; - - try - { - ourPath = driver.createsTheLock(client, path, localLockNodeBytes); - hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath); - } - catch ( KeeperException.NoNodeException e ) - { - // gets thrown by StandardLockInternalsDriver when it can't find the lock node - // this can happen when the session expires, etc. So, if the retry allows, just try it all again - if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) ) - { - isDone = false; - } - else - { - throw e; - } - } - } - - if ( hasTheLock ) - { - return ourPath; - } - - return null; - } - - private void checkRevocableWatcher(String path) throws Exception - { - RevocationSpec entry = revocable.get(); - if ( entry != null ) - { - try - { - byte[] bytes = client.getData().usingWatcher(revocableWatcher).forPath(path); - if ( Arrays.equals(bytes, REVOKE_MESSAGE) ) - { - entry.getExecutor().execute(entry.getRunnable()); - } - } - catch ( KeeperException.NoNodeException ignore ) - { - // ignore - } - } - } - - private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception - { - boolean haveTheLock = false; - boolean doDelete = false; - try - { - if ( revocable.get() != null ) - { - client.getData().usingWatcher(revocableWatcher).forPath(ourPath); - } - - while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock ) - { - List children = getSortedChildren(); - String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash - - PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases); - if ( predicateResults.getsTheLock() ) - { - haveTheLock = true; - } - else - { - String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch(); - - synchronized(this) - { - try - { - // use getData() instead of exists() to avoid leaving unneeded watchers which is a type of resource leak - client.getData().usingWatcher(watcher).forPath(previousSequencePath); - if ( millisToWait != null ) - { - millisToWait -= (System.currentTimeMillis() - startMillis); - startMillis = System.currentTimeMillis(); - if ( millisToWait <= 0 ) - { - doDelete = true; // timed out - delete our node - break; - } - - wait(millisToWait); - } - else - { - wait(); - } - } - catch ( KeeperException.NoNodeException e ) - { - // it has been deleted (i.e. lock released). Try to acquire again - } - } - } - } - } - catch ( Exception e ) - { - ThreadUtils.checkInterrupted(e); - doDelete = true; - throw e; - } - finally - { - if ( doDelete ) - { - deleteOurPath(ourPath); - } - } - return haveTheLock; - } - - private void deleteOurPath(String ourPath) throws Exception - { - try - { - client.delete().guaranteed().forPath(ourPath); - } - catch ( KeeperException.NoNodeException e ) - { - // ignore - already deleted (possibly expired session, etc.) - } - } -} From 96c153e06879f269d3693d9374f4a68f97096cf7 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Aug 2026 11:37:54 -0400 Subject: [PATCH 09/14] Assert what the interrupt test actually covers The assertion claimed the `lockInterruptibly()` would otherwise "spin forever", but the loop was already interruptible: its `tryLock()` waits inside a `Future.get()`, which raises the `InterruptedException` on its own. Removing the in-loop `checkInterruption()` leaves this test green. * assert the contract the test does cover: the `lockInterruptibly()` returns on an interrupt and propagates it instead of swallowing it in the retry loop * drop the started latch: it is counted down inside the thread just before the `lockInterruptibly()` call, so it is not a barrier and buys nothing over interrupting right after the `start()` --- .../zookeeper/lock/ZkLockRegistryTests.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) 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 cc1df23cbb..704feb60dc 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 @@ -327,15 +327,13 @@ public void voidLockFailsWhenServerDown() throws Exception { .as("The tryLock() must not block far beyond the time requested!").isLessThan(3_000); // The `lockInterruptibly()` retries until it succeeds, so with the server down - // an interrupt is the only way out: it must not be swallowed by the retry loop. + // an interrupt is the only way out of its loop. Lock lock4 = registry.obtain("interruptible"); - CountDownLatch lockAttemptStartedLatch = new CountDownLatch(1); CountDownLatch lockAttemptLatch = new CountDownLatch(1); AtomicReference lockException = new AtomicReference<>(); Thread lockThread = new Thread(() -> { try { - lockAttemptStartedLatch.countDown(); lock4.lockInterruptibly(); } catch (Exception ex) { @@ -347,12 +345,14 @@ public void voidLockFailsWhenServerDown() throws Exception { }); lockThread.start(); - assertThat(lockAttemptStartedLatch.await(10, TimeUnit.SECONDS)).isTrue(); lockThread.interrupt(); assertThat(lockAttemptLatch.await(10, TimeUnit.SECONDS)) - .as("The lockInterruptibly() must not spin forever with zookeeper server stopped!").isTrue(); - assertThat(lockException.get()).isInstanceOf(InterruptedException.class); + .as("The lockInterruptibly() must return on an interrupt with zookeeper server stopped!") + .isTrue(); + assertThat(lockException.get()) + .as("The lockInterruptibly() must propagate the interrupt, not swallow it in its loop") + .isInstanceOf(InterruptedException.class); // Otherwise a failed assertion above would leave this thread racing for the lock below. lockThread.join(10_000); From 985affeb734640f72a4359d46a594250724ceda1 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Aug 2026 11:51:16 -0400 Subject: [PATCH 10/14] Assert the keyToPath argument, not the client twice The `ZookeeperLockRegistry(CuratorFramework, KeyToPathStrategy)` constructor asserted `client` in both checks, so a `null` `KeyToPathStrategy` passed the validation and failed later with a plain `NullPointerException` from the `keyToPath.bounded()` call below. --- .../integration/zookeeper/lock/ZookeeperLockRegistry.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 e3a41cd4e2..547c2a1fce 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 @@ -122,7 +122,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(); From 7dcf7adc618308ab9517e542594100ec95270d93 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Aug 2026 11:51:32 -0400 Subject: [PATCH 11/14] Use LogAccessor in the ZookeeperLockRegistry * replace the `LogFactory.getLog()` with a `LogAccessor` * drop the `isDebugEnabled()` guards: the `Supplier` overloads defer the message concatenation on their own --- .../zookeeper/lock/ZookeeperLockRegistry.java | 17 +++++------------ 1 file changed, 5 insertions(+), 12 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 547c2a1fce..929d97eb31 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 @@ -27,12 +27,11 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.apache.curator.framework.CuratorFramework; 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; @@ -56,7 +55,7 @@ */ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableBean { - private static final Log LOGGER = LogFactory.getLog(ZookeeperLockRegistry.class); + private static final LogAccessor LOGGER = new LogAccessor(ZookeeperLockRegistry.class); private static final String DEFAULT_ROOT = "/SpringIntegration-LockRegistry"; @@ -298,9 +297,7 @@ public void lockInterruptibly() throws InterruptedException { // 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(); - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Mutex at " + this.path + " is not acquired yet; retrying..."); - } + LOGGER.debug(() -> "Mutex at " + this.path + " is not acquired yet; retrying..."); } } @@ -342,9 +339,7 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { if (!connected) { future.cancel(true); - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("No Zookeeper connection to acquire mutex at " + this.path); - } + LOGGER.debug(() -> "No Zookeeper connection to acquire mutex at " + this.path); return false; } else { @@ -355,9 +350,7 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { } catch (@SuppressWarnings("unused") TimeoutException e) { future.cancel(true); - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Timed out while checking the Zookeeper connection to acquire mutex at " + this.path); - } + LOGGER.debug(() -> "Timed out while checking the Zookeeper connection to acquire mutex at " + this.path); return false; } catch (InterruptedException e) { From 9209ab77802bc4a0721655d8a98edebd8c2dd903 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Aug 2026 16:51:07 -0400 Subject: [PATCH 12/14] Cancel the connection check when tryLock() is interrupted The `!connected` and the `TimeoutException` paths both cancel the probe future, but the `InterruptedException` path did not. The default `mutexTaskExecutor` is a `ThreadPoolTaskExecutor` with `corePoolSize` of `1`, so an abandoned `checkExists()` stays runnable in the Curator retry loop for the whole `connectionTimeoutMs` (15 s by default) and holds the only worker. Every subsequent `tryLock()` on that registry then queues behind it and returns `false` on its own timeout, whatever the real connection state is. --- .../integration/zookeeper/lock/ZookeeperLockRegistry.java | 3 +++ 1 file changed, 3 insertions(+) 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 929d97eb31..85708444d4 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 @@ -354,6 +354,9 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { 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; } From 78d3bb0b12a54733828ea42e6ae182180828a0a2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Aug 2026 16:51:38 -0400 Subject: [PATCH 13/14] Polish the ZkLock comments and test wrapping * keep two lines within 120 columns - there is no `LineLength` module in the Checkstyle configuration, so neither is caught by the build * collapse the nested try-with-resources in `voidLockFailsWhenServerDown()` into one: the resources are still closed in reverse order, and the whole method loses an indentation level * say that the `Math.max(0, ...)` is defensive: Curator breaks its lock loop on `millisToWait <= 0`, so a negative deadline behaves exactly like a zero one and the previous comment implied a defect which is not there * assert what the interrupt test reaches: `Thread.start()` establishes happens-before, so the guard on the `lockInterruptibly()` entry wins the race and the retry loop is not entered * rename `lock4` to `interruptibleLock`: it is not a part of the numbered sequence --- .../zookeeper/lock/ZookeeperLockRegistry.java | 5 +- .../zookeeper/lock/ZkLockRegistryTests.java | 105 +++++++++--------- 2 files changed, 55 insertions(+), 55 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 85708444d4..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 @@ -343,14 +343,15 @@ public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { return false; } else { - // A slow connection check must not pass a negative deadline to the mutex. + // 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); + LOGGER.debug(() -> + "Timed out while checking the Zookeeper connection to acquire mutex at " + this.path); return false; } catch (InterruptedException 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 704feb60dc..677da5b11e 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 @@ -303,78 +303,77 @@ public void testLockWithBoundedStrategy() throws Exception { public void voidLockFailsWhenServerDown() throws Exception { // 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()) { - try (var client = CuratorFrameworkFactory.newClient(server.getConnectString(), - new BoundedExponentialBackoffRetry(100, 1000, 3))) { + try (var server = new TestingServer(); + var client = CuratorFrameworkFactory.newClient(server.getConnectString(), + new BoundedExponentialBackoffRetry(100, 1000, 3))) { - client.start(); + client.start(); - ZookeeperLockRegistry registry = new ZookeeperLockRegistry(client); + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(client); - Lock lock1 = registry.obtain("foo"); - lock1.lock(); + Lock lock1 = registry.obtain("foo"); + lock1.lock(); - server.stop(); + server.stop(); - Lock lock2 = registry.obtain("bar"); + Lock lock2 = registry.obtain("bar"); - long startTime = System.currentTimeMillis(); + long startTime = System.currentTimeMillis(); - assertThat(lock2.tryLock(1, TimeUnit.SECONDS)) - .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); + assertThat(lock2.tryLock(1, TimeUnit.SECONDS)) + .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); - assertThat(System.currentTimeMillis() - startTime) - .as("The tryLock() must not block far beyond the time requested!").isLessThan(3_000); + assertThat(System.currentTimeMillis() - startTime) + .as("The tryLock() must not block far beyond the time requested!").isLessThan(3_000); - // The `lockInterruptibly()` retries until it succeeds, so with the server down - // an interrupt is the only way out of its loop. - Lock lock4 = registry.obtain("interruptible"); - CountDownLatch lockAttemptLatch = new CountDownLatch(1); - AtomicReference lockException = new AtomicReference<>(); + // 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<>(); - Thread lockThread = new Thread(() -> { - try { - lock4.lockInterruptibly(); - } - catch (Exception ex) { - lockException.set(ex); - } - finally { - lockAttemptLatch.countDown(); - } - }); - lockThread.start(); + Thread lockThread = new Thread(() -> { + try { + interruptibleLock.lockInterruptibly(); + } + catch (Exception ex) { + lockException.set(ex); + } + finally { + lockAttemptLatch.countDown(); + } + }); + lockThread.start(); - lockThread.interrupt(); + 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 swallow it in its loop") - .isInstanceOf(InterruptedException.class); + 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(); + // Otherwise a failed assertion above would leave this thread racing for the lock below. + lockThread.join(10_000); + assertThat(lockThread.isAlive()).isFalse(); - server.restart(); + server.restart(); - assertThat(lock2.tryLock(10, TimeUnit.SECONDS)) - .as("Should have been able to lock with zookeeper server restarted!").isTrue(); + assertThat(lock2.tryLock(10, TimeUnit.SECONDS)) + .as("Should have been able to lock with zookeeper server restarted!").isTrue(); - assertThat(lock1.tryLock(1, TimeUnit.SECONDS)).as("Should have still held lock1").isTrue(); + assertThat(lock1.tryLock(1, TimeUnit.SECONDS)).as("Should have still held lock1").isTrue(); - Lock lock3 = registry.obtain("foobar"); + Lock lock3 = registry.obtain("foobar"); - assertThat(lock3.tryLock(1, TimeUnit.SECONDS)).as("Should have been able to a obtain new lock!").isTrue(); + assertThat(lock3.tryLock(1, TimeUnit.SECONDS)).as("Should have been able to a obtain new lock!").isTrue(); - lock1.unlock(); - lock1.unlock(); - lock2.unlock(); - lock3.unlock(); - registry.destroy(); - } + lock1.unlock(); + lock1.unlock(); + lock2.unlock(); + lock3.unlock(); + registry.destroy(); } } From b8c41c5914f2ac9c545ba0cb1c0480b16f5a4097 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Aug 2026 17:30:15 -0400 Subject: [PATCH 14/14] Replace the foo/bar lock keys in ZkLockRegistryTests * use `orders`, `invoices` and `shipments` for the lock keys, keeping every same-key and different-key relation the assertions rely on: the reentrancy tests still obtain one key twice, `testTwoLocks()` still obtains two distinct ones, and the capacity tests keep their `orders:` ordering * name the locks in `voidLockFailsWhenServerDown()` after their keys: the `lock1`, `lock2`, `lock3` sequence stopped matching anything once `lock4` became `interruptibleLock` * fix the two assertion descriptions which named those variables or read ungrammatically --- .../zookeeper/lock/ZkLockRegistryTests.java | 86 ++++++++++--------- 1 file changed, 44 insertions(+), 42 deletions(-) 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 677da5b11e..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 @@ -55,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); @@ -75,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); @@ -91,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(); @@ -110,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(); @@ -129,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(); @@ -147,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 { @@ -177,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); @@ -185,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(); @@ -214,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); @@ -222,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(); @@ -251,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); @@ -281,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); @@ -311,16 +311,16 @@ public void voidLockFailsWhenServerDown() throws Exception { ZookeeperLockRegistry registry = new ZookeeperLockRegistry(client); - Lock lock1 = registry.obtain("foo"); - lock1.lock(); + Lock ordersLock = registry.obtain("orders"); + ordersLock.lock(); server.stop(); - Lock lock2 = registry.obtain("bar"); + Lock invoicesLock = registry.obtain("invoices"); long startTime = System.currentTimeMillis(); - assertThat(lock2.tryLock(1, TimeUnit.SECONDS)) + assertThat(invoicesLock.tryLock(1, TimeUnit.SECONDS)) .as("Should not have been able to lock with zookeeper server stopped!").isFalse(); assertThat(System.currentTimeMillis() - startTime) @@ -360,19 +360,21 @@ public void voidLockFailsWhenServerDown() throws Exception { server.restart(); - assertThat(lock2.tryLock(10, TimeUnit.SECONDS)) + assertThat(invoicesLock.tryLock(10, TimeUnit.SECONDS)) .as("Should have been able to lock with zookeeper server restarted!").isTrue(); - assertThat(lock1.tryLock(1, TimeUnit.SECONDS)).as("Should have still held lock1").isTrue(); + assertThat(ordersLock.tryLock(1, TimeUnit.SECONDS)) + .as("Should have still held the lock obtained before the server went down!").isTrue(); - Lock lock3 = registry.obtain("foobar"); + Lock shipmentsLock = registry.obtain("shipments"); - assertThat(lock3.tryLock(1, TimeUnit.SECONDS)).as("Should have been able to a obtain new lock!").isTrue(); + assertThat(shipmentsLock.tryLock(1, TimeUnit.SECONDS)) + .as("Should have been able to obtain a new lock!").isTrue(); - lock1.unlock(); - lock1.unlock(); - lock2.unlock(); - lock3.unlock(); + ordersLock.unlock(); + ordersLock.unlock(); + invoicesLock.unlock(); + shipmentsLock.unlock(); registry.destroy(); } } @@ -381,7 +383,7 @@ public void voidLockFailsWhenServerDown() throws Exception { 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) { @@ -417,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(); @@ -451,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(); } @@ -466,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(); @@ -487,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); @@ -497,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(); } @@ -517,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(); @@ -538,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(); }