From e2de2e3883cfbfcabb79a0422f03cc4165a223fc Mon Sep 17 00:00:00 2001 From: Jonathan Oddy Date: Fri, 31 Aug 2018 12:57:46 -0400 Subject: [PATCH] Avoid leaking Zookeeper znodes in tryLock Creating a EPHEMERAL_SEQUENTIAL node was being used to ensure the Zookeeper connection had been established. These were never explicitly removed, which led to tryLock effectively leaking zNodes until the client was terminated. This in turn lead to an extremely large number of ephemeral nodes being deleted whenever a long running service terminated! This fix replaces the creation of a node with a stat of "/" instead. This has the desired effect of re-establishing the Zookeeper connection, but is a read-only operation. * Remove unused import (cherry picked from commit 47a74abd306b8f0d98e15727f80c514aee993035) --- .../zookeeper/lock/ZookeeperLockRegistry.java | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java index 2807aa92e5..7291c7884d 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 @@ -29,7 +29,6 @@ import java.util.concurrent.locks.Lock; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.recipes.locks.InterProcessMutex; -import org.apache.zookeeper.CreateMode; import org.springframework.beans.factory.DisposableBean; import org.springframework.core.task.AsyncTaskExecutor; @@ -281,28 +280,24 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { - Future future = null; + Future future = null; try { long startTime = System.currentTimeMillis(); - future = this.mutexTaskExecutor.submit(new Callable() { + future = this.mutexTaskExecutor.submit(new Callable() { @Override - public String call() throws Exception { - return ZkLock.this.client.create() - .creatingParentContainersIfNeeded() - .withProtection() - .withMode(CreateMode.EPHEMERAL_SEQUENTIAL) - .forPath(ZkLock.this.path); + public Boolean call() throws Exception { + return ZkLock.this.client.checkExists().forPath("/") != null; } }); long waitTime = unit.toMillis(time); - String ourPath = future.get(waitTime, TimeUnit.MILLISECONDS); + boolean connected = future.get(waitTime, TimeUnit.MILLISECONDS); - if (ourPath == null) { + if (!connected) { future.cancel(true); return false; }