diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java index 41a4f0bc069..d34c3d8f34d 100644 --- a/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java +++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java @@ -2912,4 +2912,18 @@ static void listLockCoordinatorsAsJSON(Object source) { @LogMessage(id = 601804, value = "User {} is listing lock coordinators as json on target resource: {}", level = LogMessage.Level.INFO) void listLockCoordinatorsAsJSON(String user, Object source); + static void startLockCoordinator(Object source, Object... args) { + BASE_LOGGER.startLockCoordinator(getCaller(), source, parametersList(args)); + } + + @LogMessage(id = 601805, value = "User {} is starting a lock coordinator on target resource: {} {}", level = LogMessage.Level.INFO) + void startLockCoordinator(String user, Object source, String args); + + static void stopLockCoordinator(Object source, Object... args) { + BASE_LOGGER.stopLockCoordinator(getCaller(), source, parametersList(args)); + } + + @LogMessage(id = 601806, value = "User {} is stopping a lock coordinator on target resource: {} {}", level = LogMessage.Level.INFO) + void stopLockCoordinator(String user, Object source, String args); + } diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java index eba2bdd15b6..00fd405421a 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/ActiveMQServerControl.java @@ -62,6 +62,12 @@ public interface ActiveMQServerControl { @Operation(desc = "Return the status of lock coordinators") String listLockCoordinatorsAsJSON(); + @Operation(desc = "Start polling for the distributed lock of a lock coordinator that isn't already started, e.g. one configured with auto-start=false", impact = MBeanOperationInfo.ACTION) + void startLockCoordinator(@Parameter(name = "name", desc = "The name of the lock coordinator") String name) throws Exception; + + @Operation(desc = "Stop polling for the distributed lock of a lock coordinator, releasing it if currently held", impact = MBeanOperationInfo.ACTION) + void stopLockCoordinator(@Parameter(name = "name", desc = "The name of the lock coordinator") String name) throws Exception; + /** * {@return the number of clients connected to this server.} */ diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/LockCoordinatorConfiguration.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/LockCoordinatorConfiguration.java index 1c742e2c6eb..bb18b514961 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/LockCoordinatorConfiguration.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/LockCoordinatorConfiguration.java @@ -26,6 +26,7 @@ public class LockCoordinatorConfiguration { String lockId; String className; int checkPeriod; + boolean autoStart = true; Map properties; public LockCoordinatorConfiguration() { @@ -85,8 +86,17 @@ public Map getProperties() { return properties; } + public boolean isAutoStart() { + return autoStart; + } + + public LockCoordinatorConfiguration setAutoStart(boolean autoStart) { + this.autoStart = autoStart; + return this; + } + @Override public String toString() { - return "LockCoordinatorConfiguration{" + "name='" + name + '\'' + ", lockId='" + lockId + '\'' + ", className='" + className + '\'' + ", checkPeriod=" + checkPeriod + ", properties=" + properties + '}'; + return "LockCoordinatorConfiguration{" + "name='" + name + '\'' + ", lockId='" + lockId + '\'' + ", className='" + className + '\'' + ", checkPeriod=" + checkPeriod + ", autoStart=" + autoStart + ", properties=" + properties + '}'; } } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java index 4d5824b15c5..b89948b6e49 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java @@ -958,6 +958,7 @@ private void parseLockCoordinator(final Element lockCoordinatorElement, final Co String lockId = getString(lockCoordinatorElement, "lock-id", name, NO_CHECK); String className = getString(lockCoordinatorElement, "class-name", null, NOT_NULL_OR_EMPTY); int checkPeriod = getInteger(lockCoordinatorElement, "check-period", LockCoordinator.DEFAULT_CHECK_PERIOD, NO_CHECK); + boolean autoStart = getBooleanAttribute(lockCoordinatorElement, "auto-start", true); HashMap properties = new HashMap<>(); @@ -973,7 +974,7 @@ private void parseLockCoordinator(final Element lockCoordinatorElement, final Co } } - LockCoordinatorConfiguration lockCoordinatorConfiguration = new LockCoordinatorConfiguration(properties).setName(name).setLockId(lockId).setClassName(className).setCheckPeriod(checkPeriod); + LockCoordinatorConfiguration lockCoordinatorConfiguration = new LockCoordinatorConfiguration(properties).setName(name).setLockId(lockId).setClassName(className).setCheckPeriod(checkPeriod).setAutoStart(autoStart); mainConfig.addLockCoordinatorConfiguration(lockCoordinatorConfiguration); } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java index 94f9873d4f9..b2edbd58e31 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java @@ -4787,6 +4787,44 @@ public String listLockCoordinatorsAsJSON() { } } + @Override + public void startLockCoordinator(String name) throws Exception { + if (AuditLogger.isBaseLoggingEnabled()) { + AuditLogger.startLockCoordinator(this.server, name); + } + checkStarted(); + + clearIO(); + try { + LockCoordinator lockCoordinator = server.getLockCoordinator(name); + if (lockCoordinator == null) { + throw new IllegalArgumentException("No lock coordinator found with name: " + name); + } + lockCoordinator.start(); + } finally { + blockOnIO(); + } + } + + @Override + public void stopLockCoordinator(String name) throws Exception { + if (AuditLogger.isBaseLoggingEnabled()) { + AuditLogger.stopLockCoordinator(this.server, name); + } + checkStarted(); + + clearIO(); + try { + LockCoordinator lockCoordinator = server.getLockCoordinator(name); + if (lockCoordinator == null) { + throw new IllegalArgumentException("No lock coordinator found with name: " + name); + } + lockCoordinator.stop(); + } finally { + blockOnIO(); + } + } + public ActiveMQServer getServer() { return server; } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java index 0967cfe46ef..4f336900776 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java @@ -1525,7 +1525,7 @@ void slowConsumerDetected(String sessionID, void lockCoordinatorNotFoundOnAcceptor(String lockName, String acceptorName); @LogMessage(id = 224156, value = "LockCoordinator {} starting with className={} and lockID={} with checkPeriod={} milliseconds", level = LogMessage.Level.INFO) - void lockCoordinatorStarting(String lockName, String className, String lockID, int checkPeriod); + void lockCoordinatorStarting(String lockName, String className, String lockID, long checkPeriod); @LogMessage(id = 224157, value = "At least one of the components failed to start under the lockCoordinator {}. A retry will be executed", level = LogMessage.Level.INFO) void retryLockCoordinator(String name); diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java index 4fb63b0b006..ee7e22817a0 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java @@ -761,7 +761,7 @@ private void internalStart() throws Exception { ActiveMQServerLogger.LOGGER.serverStarting((haPolicy.isBackup() ? "Backup" : "Primary"), configuration); - startLockCoordinators(); + createLockCoordinators(); final boolean wasPrimary = !haPolicy.isBackup(); if (!haPolicy.isBackup()) { @@ -821,7 +821,15 @@ private void internalStart() throws Exception { } } - private void startLockCoordinators() { + /** + * Instantiates the configured lock managers and creates the {@link LockCoordinator}s, making them available + * through {@link #getLockCoordinator(String)} for acceptors and broker connections to look up by name. This runs + * unconditionally on every {@link #internalStart()}, well before the server is activated, so the coordinators + * must exist here for that lookup to succeed. It does NOT start polling for the lock: that only happens once the + * server actually becomes active, in {@link #startLockCoordinators()}, called from {@link #completeActivation}. + * Otherwise a passive backup (or any node that hasn't been chosen to activate) would contend for the lock too. + */ + private void createLockCoordinators() { for (LockCoordinatorConfiguration lockCoordinatorConfiguration : configuration.getLockCoordinatorConfigurations()) { String className = lockCoordinatorConfiguration.getClassName(); String name = lockCoordinatorConfiguration.getName(); @@ -838,9 +846,24 @@ private void startLockCoordinators() { } LockCoordinator lockCoordinator = new LockCoordinator(scheduledPool, executorFactory.getExecutor(), checkPeriod, lockManager, lockId, name); + lockCoordinator.setAutoStart(lockCoordinatorConfiguration.isAutoStart()); lockCoordinators.put(name, lockCoordinator); - ActiveMQServerLogger.LOGGER.lockCoordinatorStarting(name, className, lockId, checkPeriod); - lockCoordinator.start(); + } + } + + /** + * Starts polling for the distributed lock on every created {@link LockCoordinator} configured with + * {@code auto-start=true} (the default) that isn't already started. Called from {@link #completeActivation} so + * that only the node that actually became active contends for the lock; a coordinator configured with + * {@code auto-start=false} is left stopped until it's started explicitly through management + * ({@code startLockCoordinator}). + */ + private void startLockCoordinators() { + for (LockCoordinator lockCoordinator : lockCoordinators.values()) { + if (lockCoordinator.isAutoStart() && !lockCoordinator.isStarted()) { + ActiveMQServerLogger.LOGGER.lockCoordinatorStarting(lockCoordinator.getName(), lockCoordinator.getLockManager().getClass().getName(), lockCoordinator.getLockId(), lockCoordinator.getPeriod()); + lockCoordinator.start(); + } } } @@ -3733,6 +3756,7 @@ public void completeActivation(boolean replicated) throws Exception { } } } + startLockCoordinators(); getRemotingService().startAcceptors(); activationLatch.countDown(); callActivationCompleteCallbacks(); 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..3bccfb9d394 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 @@ -79,6 +79,21 @@ public LockCoordinator setDebugInfo(String debugInfo) { DistributedLock distributedLock; volatile boolean locked; + /** + * Whether this coordinator should start polling for the distributed lock as soon as the server is activated. + * When {@code false}, the coordinator stays stopped until it is started explicitly, e.g. through management. + */ + private volatile boolean autoStart = true; + + public boolean isAutoStart() { + return autoStart; + } + + public LockCoordinator setAutoStart(boolean autoStart) { + this.autoStart = autoStart; + return this; + } + public DistributedLockManager getLockManager() { return lockManager; } diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd index cbb95bdc0ff..8567d5fe764 100644 --- a/artemis-server/src/main/resources/schema/artemis-configuration.xsd +++ b/artemis-server/src/main/resources/schema/artemis-configuration.xsd @@ -3314,6 +3314,16 @@ + + + + should the lock coordinator start polling for the distributed lock as soon as the broker is + activated. When false, the lock is not contended for until it is started explicitly through + management (startLockCoordinator). + + + + diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationParserTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationParserTest.java index 04dc87e104c..516b22bee04 100644 --- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationParserTest.java +++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationParserTest.java @@ -153,6 +153,10 @@ public class FileConfigurationParserTest extends ServerTestBase { + + sausage-factory-2 + some.class.somewhere + """; private static final String BROKER_CONNECTION_PART = """ @@ -464,16 +468,24 @@ public void testLockCoordinatorParse() throws Exception { Collection lockConfigurations = configuration.getLockCoordinatorConfigurations(); lockConfigurations.forEach(f -> logger.info("lockConfiguration={}", f)); - assertEquals(1, lockConfigurations.size()); - for (LockCoordinatorConfiguration lockConfiguration : lockConfigurations) { - assertEquals("my-lock", lockConfiguration.getName()); - assertEquals("sausage-factory", lockConfiguration.getLockId()); - assertEquals("some.class.somewhere", lockConfiguration.getClassName()); - Map properties = lockConfiguration.getProperties(); - assertEquals(2, properties.size()); - assertEquals("value1", properties.get("test1")); - assertEquals("value2", properties.get("test2")); - } + assertEquals(2, lockConfigurations.size()); + Map lockConfigurationsByName = lockConfigurations.stream().collect(Collectors.toMap(LockCoordinatorConfiguration::getName, Function.identity())); + + LockCoordinatorConfiguration lockConfiguration = lockConfigurationsByName.get("my-lock"); + assertNotNull(lockConfiguration); + assertEquals("my-lock", lockConfiguration.getName()); + assertEquals("sausage-factory", lockConfiguration.getLockId()); + assertEquals("some.class.somewhere", lockConfiguration.getClassName()); + assertTrue(lockConfiguration.isAutoStart(), "auto-start must default to true when the attribute is absent"); + Map properties = lockConfiguration.getProperties(); + assertEquals(2, properties.size()); + assertEquals("value1", properties.get("test1")); + assertEquals("value2", properties.get("test2")); + + LockCoordinatorConfiguration lockConfigurationNoAutoStart = lockConfigurationsByName.get("my-lock-no-autostart"); + assertNotNull(lockConfigurationNoAutoStart); + assertFalse(lockConfigurationNoAutoStart.isAutoStart(), "auto-start=\"false\" must be honored by the parser"); + configuration.getAcceptorConfigurations().stream().filter(f -> f.getName().equals("netty-with-lock")).forEach(f -> assertEquals("my-lock", f.getLockCoordinator())); assertEquals(1, configuration.getAMQPConnection().size()); assertEquals("my-lock", configuration.getAMQPConnection().get(0).getLockCoordinator()); diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/lock/LockCoordinatorAutoStartTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/lock/LockCoordinatorAutoStartTest.java new file mode 100644 index 00000000000..2e1d51db526 --- /dev/null +++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/lock/LockCoordinatorAutoStartTest.java @@ -0,0 +1,124 @@ +/* + * 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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; + +import org.apache.activemq.artemis.core.config.LockCoordinatorConfiguration; +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.junit.jupiter.api.Test; + +/** + * Unit-level coverage for the {@code auto-start} plumbing of lock coordinators: the config default/setter and the + * {@link LockCoordinator} flag itself. Whether a coordinator actually stays stopped until the broker is activated + * (or until started through management) is exercised by integration tests, since it depends on the server's + * activation lifecycle. + */ +public class LockCoordinatorAutoStartTest extends ArtemisTestCase { + + private static final int CHECK_PERIOD = 60_000; + + @Test + public void testLockCoordinatorConfigurationAutoStartDefaultsToTrue() { + LockCoordinatorConfiguration configuration = new LockCoordinatorConfiguration(); + assertTrue(configuration.isAutoStart(), "auto-start must default to true"); + } + + @Test + public void testLockCoordinatorConfigurationAutoStartSetter() { + LockCoordinatorConfiguration configuration = new LockCoordinatorConfiguration(); + LockCoordinatorConfiguration returned = configuration.setAutoStart(false); + assertSame(configuration, returned, "setAutoStart must return this for fluent chaining"); + assertFalse(configuration.isAutoStart()); + } + + @Test + public void testLockCoordinatorAutoStartDefaultsToTrue() { + LockCoordinator coordinator = newCoordinator(); + assertTrue(coordinator.isAutoStart(), "a LockCoordinator must be auto-starting unless configured otherwise"); + } + + @Test + public void testLockCoordinatorAutoStartSetter() { + LockCoordinator coordinator = newCoordinator(); + LockCoordinator returned = coordinator.setAutoStart(false); + assertSame(coordinator, returned, "setAutoStart must return this for fluent chaining"); + assertFalse(coordinator.isAutoStart()); + } + + private LockCoordinator newCoordinator() { + // scheduledExecutor is only needed once start() is called, which none of these tests do. + final Executor directExecutor = Runnable::run; + return new LockCoordinator(null, directExecutor, CHECK_PERIOD, new NoopLockManager(), "theLock", "theLock"); + } + + /** + * A DistributedLockManager whose methods are never expected to be exercised by these tests: they only check the + * autoStart flag on the LockCoordinator itself, without ever calling start()/run(). + */ + private static class NoopLockManager implements DistributedLockManager { + + private volatile boolean started; + + @Override + public void addUnavailableManagerListener(UnavailableManagerListener listener) { + } + + @Override + public void removeUnavailableManagerListener(UnavailableManagerListener listener) { + } + + @Override + public boolean start(long timeout, TimeUnit unit) { + started = true; + return true; + } + + @Override + public void start() { + started = true; + } + + @Override + public boolean isStarted() { + return started; + } + + @Override + public void stop() { + started = false; + } + + @Override + public DistributedLock getDistributedLock(String lockId) { + throw new UnsupportedOperationException("not needed by these tests"); + } + + @Override + public MutableLong getMutableLong(String mutableLongId) { + throw new UnsupportedOperationException("not needed by these tests"); + } + } +} diff --git a/docs/user-manual/lock-coordination.adoc b/docs/user-manual/lock-coordination.adoc index 92c8679876d..76a4d1e2902 100644 --- a/docs/user-manual/lock-coordination.adoc +++ b/docs/user-manual/lock-coordination.adoc @@ -133,8 +133,15 @@ The following elements are configured on every lock-coordinator regardless of im |No |5000 |How often to check if the lock is still valid, in milliseconds + +|auto-start +|No +|true +|Whether the lock coordinator starts polling for the distributed lock as soon as the broker is activated. When `false`, the coordinator stays stopped, and the lock is not contended for, until it is started explicitly through management (see <>). |=== +NOTE: In a journal HA setup with either shared storage or journal replication, lock coordinators will stay inactive and will not poll for the lock until the server is activated. + === File-Based Lock Manager The file-based lock manager uses the file system to manage distributed locks through file locking mechanisms. @@ -202,3 +209,12 @@ This is the recommended approach for production deployments as it provides bette |1000 |Delay in milliseconds between retry attempts |=== + +== Management Operations + +The `ActiveMQServerControl` management interface exposes the following operations for lock coordinators, on top of the existing `listLockCoordinatorsAsJSON` that reports their current status: + +* `startLockCoordinator(name)` - starts polling for the distributed lock of a lock coordinator that isn't already started, most commonly one configured with `auto-start="false"`. Calling it on an already started lock coordinator is a no-op. An unknown `name` throws `IllegalArgumentException`. +* `stopLockCoordinator(name)` - stops polling for the distributed lock. If the lock is currently held, it is released, and any element bound to the lock coordinator (acceptors, broker connections) is paused as if the lock had been lost. An unknown `name` throws `IllegalArgumentException`. + +These operations allow an administrator to manually control which broker owns a lock coordinator's elements, e.g. to force a controlled fail-over, or to defer contention for a coordinator configured with `auto-start="false"` until some external condition is met. diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/LockCoordinatorAutoStartManagementTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/LockCoordinatorAutoStartManagementTest.java new file mode 100644 index 00000000000..da995262084 --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/LockCoordinatorAutoStartManagementTest.java @@ -0,0 +1,120 @@ +/* + * 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.amqp.connect; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.activemq.artemis.api.core.TransportConfiguration; +import org.apache.activemq.artemis.api.core.management.ActiveMQServerControl; +import org.apache.activemq.artemis.core.config.Configuration; +import org.apache.activemq.artemis.core.config.LockCoordinatorConfiguration; +import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.lock.LockCoordinator; +import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager; +import org.apache.activemq.artemis.spi.core.remoting.Acceptor; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.tests.util.Wait; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +/** + * A lock coordinator configured with {@code auto-start="false"} must stay stopped after the server activates, and + * the elements bound to it (in this test, an acceptor) must stay closed, until it's started explicitly through + * management. This also exercises {@code startLockCoordinator}/{@code stopLockCoordinator}, including a start + * following a previous stop, an already-started coordinator being a no-op, and an unknown name being rejected. + */ +public class LockCoordinatorAutoStartManagementTest extends ActiveMQTestBase { + + private static final String LOCK_NAME = "theLock"; + private static final String ACCEPTOR_NAME = "locked"; + private static final int ACCEPTOR_PORT = 61617; + + private ActiveMQServer server; + + @AfterEach + public void stopServer() throws Exception { + if (server != null) { + server.stop(); + } + } + + @Test + public void testAutoStartFalseIsControllableThroughManagement() throws Exception { + Configuration config = createDefaultConfig(false); + config.clearAcceptorConfigurations(); + + Map params = new HashMap<>(); + params.put(TransportConstants.PORT_PROP_NAME, ACCEPTOR_PORT); + TransportConfiguration acceptorConfig = new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, params, ACCEPTOR_NAME, new HashMap<>()); + acceptorConfig.setLockCoordinator(LOCK_NAME); + config.addAcceptorConfiguration(acceptorConfig); + + Map lockProperties = new HashMap<>(); + lockProperties.put("locks-folder", getTemporaryDir()); + LockCoordinatorConfiguration lockCoordinatorConfiguration = new LockCoordinatorConfiguration(lockProperties) + .setName(LOCK_NAME).setLockId(LOCK_NAME).setClassName(FileBasedLockManager.class.getName()).setCheckPeriod(100).setAutoStart(false); + config.addLockCoordinatorConfiguration(lockCoordinatorConfiguration); + + server = addServer(createServer(false, config)); + server.start(); + Wait.assertTrue(server::isActive); + + // auto-start=false: the coordinator must stay stopped after activation, and the acceptor bound to it must + // stay closed as a consequence. + LockCoordinator lockCoordinator = server.getLockCoordinator(LOCK_NAME); + assertNotNull(lockCoordinator, "the coordinator must still be created even with auto-start=false"); + assertEquals("Stopped", lockCoordinator.getStatus()); + + Acceptor acceptor = server.getRemotingService().getAcceptor(ACCEPTOR_NAME); + assertNotNull(acceptor); + assertFalse(acceptor.isStarted(), "the acceptor must stay closed while its lock coordinator hasn't acquired the lock"); + + ActiveMQServerControl serverControl = server.getActiveMQServerControl(); + + // an unknown lock coordinator name must be rejected + assertThrows(IllegalArgumentException.class, () -> serverControl.startLockCoordinator("does-not-exist")); + assertThrows(IllegalArgumentException.class, () -> serverControl.stopLockCoordinator("does-not-exist")); + + // start it explicitly through management + serverControl.startLockCoordinator(LOCK_NAME); + Wait.assertEquals("Locked", lockCoordinator::getStatus, 10_000, 100); + Wait.assertTrue(acceptor::isStarted, 10_000, 100); + + // calling start again on an already started coordinator is a no-op + serverControl.startLockCoordinator(LOCK_NAME); + assertEquals("Locked", lockCoordinator.getStatus()); + assertTrue(acceptor.isStarted()); + + // stop it: the lock is released and the acceptor closes again + serverControl.stopLockCoordinator(LOCK_NAME); + Wait.assertEquals("Stopped", lockCoordinator::getStatus, 10_000, 100); + Wait.assertFalse(acceptor::isStarted, 10_000, 100); + + // and a fresh start after a stop must work just as well + serverControl.startLockCoordinator(LOCK_NAME); + Wait.assertEquals("Locked", lockCoordinator::getStatus, 10_000, 100); + Wait.assertTrue(acceptor::isStarted, 10_000, 100); + } +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/LockCoordinatorHaActivationTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/LockCoordinatorHaActivationTest.java new file mode 100644 index 00000000000..1e8fa5f7561 --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/LockCoordinatorHaActivationTest.java @@ -0,0 +1,109 @@ +/* + * 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.cluster.failover; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.activemq.artemis.api.core.TransportConfiguration; +import org.apache.activemq.artemis.core.config.Configuration; +import org.apache.activemq.artemis.core.config.LockCoordinatorConfiguration; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.NodeManager; +import org.apache.activemq.artemis.core.server.impl.InVMNodeManager; +import org.apache.activemq.artemis.core.server.lock.LockCoordinator; +import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.artemis.tests.util.ReplicatedBackupUtils; +import org.apache.activemq.artemis.tests.util.TransportConfigurationUtils; +import org.apache.activemq.artemis.tests.util.Wait; +import org.junit.jupiter.api.Test; + +/** + * A lock coordinator must only start polling for its distributed lock once the broker that owns it is actually + * active. Before this was fixed, {@code ActiveMQServerImpl.internalStart()} started every configured lock + * coordinator unconditionally, regardless of primary/backup role. As a consequence a passive HA backup (replication + * ha-policy) would compete for the lock with the currently active primary and, if it happened to win it, elements + * bound to the lock coordinator (acceptors, broker connections) would stay dead on the primary while the passive + * backup - which isn't accepting any traffic - silently held the lock. + */ +public class LockCoordinatorHaActivationTest extends ActiveMQTestBase { + + private static final String LOCK_NAME = "theLock"; + + @Test + public void testPassiveBackupDoesNotContendForLock() throws Exception { + final String locksFolder = getTemporaryDir(); + + final TransportConfiguration primaryConnector = TransportConfigurationUtils.getInVMConnector(true); + final TransportConfiguration backupConnector = TransportConfigurationUtils.getInVMConnector(false); + final TransportConfiguration backupAcceptor = TransportConfigurationUtils.getInVMAcceptor(false); + + Configuration backupConfig = createDefaultInVMConfig(); + Configuration primaryConfig = createDefaultInVMConfig(); + + ReplicatedBackupUtils.configureReplicationPair(backupConfig, backupConnector, backupAcceptor, primaryConfig, primaryConnector, null); + primaryConfig.clearAcceptorConfigurations().addAcceptorConfiguration(TransportConfigurationUtils.getInVMAcceptor(true)); + + // both nodes are configured with the very same lock-coordinator, pointing at the very same locks-folder, so + // they really do contend for the same file lock if both were to start polling for it. + addLockCoordinator(primaryConfig, locksFolder); + addLockCoordinator(backupConfig, locksFolder); + + NodeManager primaryNodeManager = new InVMNodeManager(true, primaryConfig.getJournalLocation()); + ActiveMQServer primaryServer = createInVMFailoverServer(true, primaryConfig, primaryNodeManager, 1); + + NodeManager backupNodeManager = new InVMNodeManager(true, backupConfig.getJournalLocation()); + ActiveMQServer backupServer = createInVMFailoverServer(true, backupConfig, backupNodeManager, 2); + + primaryServer.start(); + Wait.assertTrue(primaryServer::isActive); + Wait.assertEquals("Locked", () -> primaryServer.getLockCoordinator(LOCK_NAME).getStatus()); + + backupServer.start(); + Wait.assertTrue(backupServer::isReplicaSync, 30_000, 100); + + // the backup is passive: its lock coordinator must have been created (so acceptors/broker connections can + // look it up by name) but must stay stopped, never contending for the lock with the active primary. + LockCoordinator backupCoordinator = backupServer.getLockCoordinator(LOCK_NAME); + assertNotNull(backupCoordinator, "the backup must create its lock coordinators even while passive"); + assertEquals("Stopped", backupCoordinator.getStatus()); + + // give the backup's coordinator a few check-periods worth of time to prove it really isn't polling + Thread.sleep(500); + assertEquals("Stopped", backupCoordinator.getStatus(), "a passive backup must not contend for the lock"); + assertEquals("Locked", primaryServer.getLockCoordinator(LOCK_NAME).getStatus(), "the active primary must still be holding the lock"); + + // the primary goes away: the backup activates and, only now, is allowed to start polling for (and acquire) + // the lock. + primaryServer.stop(); + + Wait.assertTrue(backupServer::isActive, 30_000, 100); + Wait.assertEquals("Locked", () -> backupServer.getLockCoordinator(LOCK_NAME).getStatus(), 30_000, 100); + } + + private void addLockCoordinator(Configuration configuration, String locksFolder) { + Map properties = new HashMap<>(); + properties.put("locks-folder", locksFolder); + LockCoordinatorConfiguration lockCoordinatorConfiguration = new LockCoordinatorConfiguration(properties) + .setName(LOCK_NAME).setLockId(LOCK_NAME).setClassName(FileBasedLockManager.class.getName()).setCheckPeriod(100); + configuration.addLockCoordinatorConfiguration(lockCoordinatorConfiguration); + } +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java index f7496a4b0ea..03467a19753 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ActiveMQServerControlUsingCoreTest.java @@ -1892,6 +1892,16 @@ public String listLockCoordinatorsAsJSON() { return null; } } + + @Override + public void startLockCoordinator(String name) throws Exception { + proxy.invokeOperation("startLockCoordinator", name); + } + + @Override + public void stopLockCoordinator(String name) throws Exception { + proxy.invokeOperation("stopLockCoordinator", name); + } }; }