From b4481631f9fc1fc80d415632b8e769b0f4b2f5de Mon Sep 17 00:00:00 2001 From: michael Date: Fri, 17 Jul 2026 16:33:09 +0200 Subject: [PATCH] ARTEMIS-6159 Bound lock coordinator start so an unreachable ZooKeeper does not hang the broker LockCoordinator.run() started the lock manager with the unbounded start() overload. Against CuratorDistributedLockManager this delegates to start(-1, null), which blocks forever in CuratorFramework.blockUntilConnected while ZooKeeper is unreachable. Because run() executes on the coordinator's own ordered executor, a stuck start() permanently occupies that executor. stop() queues its cleanup on the same executor and waits on it with no timeout, so broker stop/restart hangs indefinitely. This also happens for a lock coordinator that isn't referenced by any acceptor or broker connection, since every configured coordinator is started unconditionally. Bound the start call to the coordinator's own check period instead: on timeout it is simply retried on the next scheduled run, at the same cadence it already runs at. Once run() can no longer block forever, stop()'s cleanup task gets its turn within one check period and returns on its own, so stop() itself did not need to change. Added LockCoordinatorTest (artemis-server) with a lock manager that blocks on start like an unreachable ZooKeeper would, and UnreachableLockManagerTest (integration-tests) with a real CuratorDistributedLockManager pointed at a closed port and an unassigned coordinator. Both hang and fail without the fix, pass with it. --- .../core/server/lock/LockCoordinator.java | 7 +- .../core/server/lock/LockCoordinatorTest.java | 182 ++++++++++++++++++ .../UnreachableLockManagerTest.java | 95 +++++++++ 3 files changed, 283 insertions(+), 1 deletion(-) create mode 100644 artemis-server/src/test/java/org/apache/activemq/artemis/core/server/lock/LockCoordinatorTest.java create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/lockmanager/UnreachableLockManagerTest.java diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/lock/LockCoordinator.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/lock/LockCoordinator.java index f722c5354f5..3abac52ba6a 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/lock/LockCoordinator.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/lock/LockCoordinator.java @@ -313,7 +313,12 @@ public void run() { try { if (!locked) { if (!lockManager.isStarted()) { - lockManager.start(); + // a bounded start: an unreachable lock manager must not hold this executor forever, + // it is simply retried on the next scheduled run + if (!lockManager.start(getPeriod(), getTimeUnit())) { + logger.debug("Not able to start the lock manager for {}, lockID={}", name, lockID); + return; + } } DistributedLock lock = lockManager.getDistributedLock(lockID); if (lock.tryLock(1, TimeUnit.SECONDS)) { diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/lock/LockCoordinatorTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/lock/LockCoordinatorTest.java new file mode 100644 index 00000000000..24e2207aa3d --- /dev/null +++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/lock/LockCoordinatorTest.java @@ -0,0 +1,182 @@ +/* + * 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.activemq.artemis.core.server.lock; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import org.apache.activemq.artemis.lockmanager.DistributedLock; +import org.apache.activemq.artemis.lockmanager.DistributedLockManager; +import org.apache.activemq.artemis.lockmanager.MutableLong; +import org.apache.activemq.artemis.tests.util.ArtemisTestCase; +import org.apache.activemq.artemis.utils.ActiveMQThreadFactory; +import org.apache.activemq.artemis.utils.actors.OrderedExecutor; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +public class LockCoordinatorTest extends ArtemisTestCase { + + private static final int CHECK_PERIOD = 100; + + /** + * A DistributedLockManager blocking forever on start, the same way CuratorDistributedLockManager does + * when ZooKeeper is not reachable: start() calls start(-1, null), which parks on + * CuratorFramework::blockUntilConnected with no timeout. + */ + private static class BlockedOnStartLockManager implements DistributedLockManager { + + private final CountDownLatch startCalled; + private final CountDownLatch releaseStart; + private volatile boolean started; + + BlockedOnStartLockManager(CountDownLatch startCalled, CountDownLatch releaseStart) { + this.startCalled = startCalled; + this.releaseStart = releaseStart; + } + + @Override + public void addUnavailableManagerListener(UnavailableManagerListener listener) { + } + + @Override + public void removeUnavailableManagerListener(UnavailableManagerListener listener) { + } + + @Override + public boolean start(long timeout, TimeUnit unit) throws InterruptedException { + startCalled.countDown(); + if (timeout >= 0) { + return releaseStart.await(timeout, unit) && markStarted(); + } + releaseStart.await(); + return markStarted(); + } + + @Override + public void start() throws InterruptedException { + start(-1, null); + } + + private boolean markStarted() { + started = true; + return true; + } + + @Override + public boolean isStarted() { + return started; + } + + @Override + public void stop() { + started = false; + } + + @Override + public DistributedLock getDistributedLock(String lockId) { + return new NeverAcquiredLock(lockId); + } + + @Override + public MutableLong getMutableLong(String mutableLongId) { + throw new UnsupportedOperationException(); + } + } + + private static class NeverAcquiredLock implements DistributedLock { + + private final String lockId; + + NeverAcquiredLock(String lockId) { + this.lockId = lockId; + } + + @Override + public String getLockId() { + return lockId; + } + + @Override + public boolean isHeldByCaller() { + return false; + } + + @Override + public boolean tryLock() { + return false; + } + + @Override + public void unlock() { + } + + @Override + public void addListener(UnavailableLockListener listener) { + } + + @Override + public void removeListener(UnavailableLockListener listener) { + } + + @Override + public void close() { + } + } + + /** + * A lock manager unable to connect must not make LockCoordinator::stop hang: the periodic task and the + * cleanup share the same ordered executor, hence a start blocking forever would keep the broker from + * being stopped or restarted. + */ + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + public void testStopWithLockManagerBlockedOnStart() throws Exception { + final CountDownLatch startCalled = new CountDownLatch(1); + final CountDownLatch releaseStart = new CountDownLatch(1); + final ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor(ActiveMQThreadFactory.defaultThreadFactory(getClass().getName())); + final ExecutorService executorService = Executors.newSingleThreadExecutor(ActiveMQThreadFactory.defaultThreadFactory(getClass().getName())); + try { + final DistributedLockManager lockManager = new BlockedOnStartLockManager(startCalled, releaseStart); + final LockCoordinator coordinator = new LockCoordinator(scheduledExecutor, new OrderedExecutor(executorService), CHECK_PERIOD, lockManager, "theLock", "theLock"); + coordinator.start(); + + assertTrue(startCalled.await(10, TimeUnit.SECONDS), "the lock coordinator never tried to start the lock manager"); + + final CountDownLatch stopped = new CountDownLatch(1); + final Thread stopper = new Thread(() -> { + coordinator.stop(); + stopped.countDown(); + }, "lock-coordinator-stopper"); + stopper.start(); + try { + assertTrue(stopped.await(30, TimeUnit.SECONDS), "LockCoordinator::stop is hanging while the lock manager is blocked on start"); + } finally { + releaseStart.countDown(); + stopper.join(TimeUnit.SECONDS.toMillis(10)); + } + } finally { + releaseStart.countDown(); + executorService.shutdownNow(); + scheduledExecutor.shutdownNow(); + } + } +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/lockmanager/UnreachableLockManagerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/lockmanager/UnreachableLockManagerTest.java new file mode 100644 index 00000000000..20c2e496bab --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/lockmanager/UnreachableLockManagerTest.java @@ -0,0 +1,95 @@ +/* + * 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.activemq.artemis.tests.integration.lockmanager; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.net.ServerSocket; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.activemq.artemis.core.config.Configuration; +import org.apache.activemq.artemis.core.config.LockCoordinatorConfiguration; +import org.apache.activemq.artemis.core.server.ActiveMQServers; +import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.utils.Wait; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +/** + * A lock coordinator unable to reach ZooKeeper must not make the broker unstoppable. + */ +public class UnreachableLockManagerTest extends ActiveMQTestBase { + + private static final String CURATOR_LOCK_MANAGER = "org.apache.activemq.artemis.lockmanager.zookeeper.CuratorDistributedLockManager"; + private static final int CHECK_PERIOD = 100; + + /** + * The lock coordinator is not referenced by any acceptor or broker connection, still it is started with the + * broker: with an unreachable ZooKeeper its periodic task used to park forever on the lock manager start, + * holding the ordered executor the stop is relying on, hence the broker could not be stopped or restarted. + */ + @Test + @Timeout(value = 120, unit = TimeUnit.SECONDS) + public void testStopWithUnreachableZooKeeper() throws Exception { + final Configuration config = createDefaultInVMConfig(); + + final Map properties = new HashMap<>(); + // nothing is listening on this port: the ZooKeeper ensemble is unreachable + properties.put("connect-string", "localhost:" + unusedPort()); + properties.put("namespace", "activemq-artemis"); + properties.put("session-ms", "2000"); + properties.put("connection-ms", "2000"); + final LockCoordinatorConfiguration lockCoordinatorConfiguration = new LockCoordinatorConfiguration(properties); + lockCoordinatorConfiguration.setName("zk").setClassName(CURATOR_LOCK_MANAGER).setCheckPeriod(CHECK_PERIOD).setLockId("zk"); + config.addLockCoordinatorConfiguration(lockCoordinatorConfiguration); + + // the server is created outside of the test base on purpose: an unstoppable broker would hang the tear down + final ActiveMQServerImpl server = (ActiveMQServerImpl) ActiveMQServers.newActiveMQServer(config, false); + server.start(); + Wait.assertTrue(() -> server.getLockCoordinators().size() == 1, 10_000, 100); + // let the periodic task run and try to connect + Wait.assertTrue(() -> "Unlocked".equals(server.getLockCoordinator("zk").getStatus()), 10_000, 100); + + final CountDownLatch stopped = new CountDownLatch(1); + final Thread stopper = new Thread(() -> { + try { + server.stop(); + } catch (Exception e) { + e.printStackTrace(); + } finally { + stopped.countDown(); + } + }, "unreachable-lock-manager-server-stopper"); + stopper.start(); + try { + assertTrue(stopped.await(60, TimeUnit.SECONDS), "the broker cannot be stopped while the lock coordinator cannot reach ZooKeeper"); + } finally { + stopper.join(TimeUnit.SECONDS.toMillis(10)); + } + } + + private static int unusedPort() throws IOException { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } +}