ARTEMIS-6162 Only start lock coordinators once the broker is activated

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);
+
 }

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.}
     */

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<String, String> properties;
 
    public LockCoordinatorConfiguration() {
@@ -85,8 +86,17 @@ public Map<String, String> 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 + '}';
    }
 }

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<String, String> 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);
    }
 

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;
    }

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);

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();

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;
    }

artemis-server/src/main/resources/schema/artemis-configuration.xsd

@@ -3314,6 +3314,16 @@
          </xsd:annotation>
       </xsd:attribute>
 
+      <xsd:attribute name="auto-start" type="xsd:boolean" default="true">
+         <xsd:annotation>
+            <xsd:documentation>
+               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).
+            </xsd:documentation>
+         </xsd:annotation>
+      </xsd:attribute>
+
    </xsd:complexType>
 
    <xsd:element name="lock-coordinators">

artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/FileConfigurationParserTest.java

@@ -153,6 +153,10 @@ public class FileConfigurationParserTest extends ServerTestBase {
                <property key='test2' value='value2'/>
             </properties>
          </lock-coordinator>
+         <lock-coordinator name="my-lock-no-autostart" auto-start="false">
+            <lock-id>sausage-factory-2</lock-id>
+            <class-name>some.class.somewhere</class-name>
+         </lock-coordinator>
       </lock-coordinators>""";
 
    private static final String BROKER_CONNECTION_PART = """
@@ -464,16 +468,24 @@ public void testLockCoordinatorParse() throws Exception {
 
       Collection<LockCoordinatorConfiguration> 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<String, String> properties = lockConfiguration.getProperties();
-         assertEquals(2, properties.size());
-         assertEquals("value1", properties.get("test1"));
-         assertEquals("value2", properties.get("test2"));
-      }
+      assertEquals(2, lockConfigurations.size());
+      Map<String, LockCoordinatorConfiguration> 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<String, String> 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());