ARTEMIS-5852 Leader/Follower pattern on acceptors

I'm adding a LeaderManager to the broker, that will use DistributedLock
to determine the node that will be the lead.

You can associate the leaderManager with acceptors and an acceptor
serving only clients would then be activated in only one of the brokers.

Author: clebertsuconic

artemis-bom/pom.xml

@@ -191,6 +191,11 @@
             <artifactId>artemis-lockmanager-ri</artifactId>
             <version>${project.version}</version>
          </dependency>
+         <dependency>
+            <groupId>org.apache.artemis</groupId>
+            <artifactId>artemis-lockmanager-etcd</artifactId>
+            <version>${project.version}</version>
+         </dependency>
          <dependency>
             <groupId>org.apache.artemis</groupId>
             <artifactId>artemis-ra</artifactId>
@@ -393,6 +398,11 @@
             <artifactId>artemis-lockmanager-ri</artifactId>
             <version>${project.version}</version>
          </dependency>
+         <dependency>
+            <groupId>org.apache.activemq</groupId>
+            <artifactId>artemis-lockmanager-etcd</artifactId>
+            <version>${project.version}</version>
+         </dependency>
          <dependency>
             <groupId>org.apache.activemq</groupId>
             <artifactId>artemis-ra</artifactId>

artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/TransportConfiguration.java

@@ -51,6 +51,8 @@ public class TransportConfiguration implements Serializable {
 
    private String name;
 
+   private String leaderManager;
+
    private String factoryClassName = "null";
 
    private Map<String, Object> params;
@@ -413,6 +415,15 @@ public void decode(final ActiveMQBuffer buffer) {
       }
    }
 
+   public String getLeaderManager() {
+      return leaderManager;
+   }
+
+   public TransportConfiguration setLeaderManager(String leaderManager) {
+      this.leaderManager = leaderManager;
+      return this;
+   }
+
    private static String replaceWildcardChars(final String str) {
       return str.replace('.', '-');
    }

artemis-distribution/pom.xml

@@ -126,6 +126,10 @@
          <groupId>org.apache.artemis</groupId>
          <artifactId>artemis-lockmanager-ri</artifactId>
       </dependency>
+      <dependency>
+         <groupId>org.apache.artemis</groupId>
+         <artifactId>artemis-lockmanager-etcd</artifactId>
+      </dependency>
 
       <!--TODO: no other modules seem to use this, is it equivalent to something else they do use ? -->
       <dependency>

artemis-lockmanager/artemis-lockmanager-api/pom.xml

@@ -36,6 +36,25 @@
          <groupId>org.apache.artemis</groupId>
          <artifactId>artemis-commons</artifactId>
       </dependency>
+
+      <!-- tests -->
+      <dependency>
+         <groupId>org.junit.jupiter</groupId>
+         <artifactId>junit-jupiter-api</artifactId>
+         <scope>test</scope>
+      </dependency>
+      <dependency>
+         <groupId>org.junit.jupiter</groupId>
+         <artifactId>junit-jupiter-engine</artifactId>
+         <scope>test</scope>
+      </dependency>
+      <dependency>
+         <groupId>org.apache.artemis</groupId>
+         <artifactId>artemis-unit-test-support</artifactId>
+         <version>${project.version}</version>
+         <scope>test</scope>
+      </dependency>
+
    </dependencies>
 
 </project>

artemis-lockmanager/artemis-lockmanager-api/src/main/java/org/apache/activemq/artemis/lockmanager/DistributedLock.java

@@ -24,7 +24,7 @@ public interface DistributedLock extends AutoCloseable {
 
    String getLockId();
 
-   boolean isHeldByCaller() throws UnavailableStateException;
+   boolean isLockValid() throws UnavailableStateException;
 
    boolean tryLock() throws UnavailableStateException, InterruptedException;
 

artemis-lockmanager/artemis-lockmanager-api/src/main/java/org/apache/activemq/artemis/lockmanager/DistributedLockManager.java

@@ -21,16 +21,14 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-import org.apache.activemq.artemis.utils.ClassloadingUtil;
-
 public interface DistributedLockManager extends AutoCloseable {
 
    static DistributedLockManager newInstanceOf(String className, Map<String, String> properties) throws Exception {
-      return (DistributedLockManager) ClassloadingUtil.getInstanceForParamsWithTypeCheck(className,
-                                                                                         DistributedLockManager.class,
-                                                                                         DistributedLockManager.class.getClassLoader(),
-                                                                                         new Class[]{Map.class},
-                                                                                         properties);
+      DistributedLockManagerFactory factory = Registry.getInstance().getFactoryWithClassName(className);
+      if (factory == null) {
+         throw new IllegalArgumentException(className + " not found");
+      }
+      return factory.build(properties);
    }
 
    @FunctionalInterface

artemis-lockmanager/artemis-lockmanager-api/src/main/java/org/apache/activemq/artemis/lockmanager/DistributedLockManagerFactory.java

@@ -0,0 +1,43 @@
+/*
+ * 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.lockmanager;
+
+import java.util.Map;
+import java.util.Set;
+
+public interface DistributedLockManagerFactory {
+   DistributedLockManager build(Map<String, String> properties);
+
+   String getName();
+
+   String getImplName();
+
+   default Map<String, String> validateParameters(Map<String, String> config) {
+      config.forEach((parameterName, ignore) -> validateParameter(parameterName));
+      return config;
+   }
+
+   Set<String> getValidParametersList();
+
+   default void validateParameter(String parameterName) {
+      Set<String> validList = getValidParametersList();
+      if (!validList.contains(parameterName)) {
+         throw new IllegalArgumentException("Invalid parameter '" + parameterName + "'. Accepted parameters: " + String.join(", ", validList));
+      }
+   }
+}

artemis-lockmanager/artemis-lockmanager-api/src/main/java/org/apache/activemq/artemis/lockmanager/Registry.java

@@ -0,0 +1,82 @@
+/*
+ * 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.lockmanager;
+
+import java.util.HashMap;
+import java.util.ServiceLoader;
+
+public class Registry {
+
+   private final HashMap<String, DistributedLockManagerFactory> factories = new HashMap<>();
+   private final HashMap<String, DistributedLockManagerFactory> factoriesWithImpl = new HashMap<>();
+
+   private volatile boolean serviceLoaded = false;
+
+   private static final Registry INSTANCE = new Registry();
+
+   private Registry() {
+   }
+
+   public static Registry getInstance() {
+      return INSTANCE;
+   }
+
+   public synchronized void register(DistributedLockManagerFactory factory) {
+      factories.put(factory.getName(), factory);
+      factoriesWithImpl.put(factory.getImplName(), factory);
+   }
+
+   public synchronized void unregisterWithType(String type) {
+      DistributedLockManagerFactory factory = factories.get(type);
+      unregister(factory);
+   }
+
+   public synchronized void unregisterWithClassName(String name) {
+      DistributedLockManagerFactory factory = factoriesWithImpl.get(name);
+      unregister(factory);
+   }
+
+   private void unregister(DistributedLockManagerFactory factory) {
+      factories.remove(factory.getName());
+      factoriesWithImpl.remove(factory.getImplName());
+   }
+
+   public synchronized DistributedLockManagerFactory getFactoryWithClassName(String className) {
+      checkService();
+      DistributedLockManagerFactory factory = factoriesWithImpl.get(className);
+      if (factory == null) {
+         throw new IllegalArgumentException("factory " + className + " not found");
+      }
+      return factory;
+   }
+
+   public synchronized DistributedLockManagerFactory getFactory(String name) {
+      checkService();
+      return factories.get(name);
+   }
+
+   public synchronized void checkService() {
+      if (serviceLoaded) {
+         return;
+      }
+      ServiceLoader<DistributedLockManagerFactory> services = ServiceLoader.load(DistributedLockManagerFactory.class);
+      services.forEach(this::register);
+      serviceLoaded = true;
+   }
+
+}

artemis-lockmanager/artemis-lockmanager-api/src/test/java/org/apache/activemq/artemis/lockmanager/RegistryTest.java

@@ -0,0 +1,73 @@
+/*
+ * 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.lockmanager;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class RegistryTest {
+
+
+   @Test
+   public void testRegistryUnregister() {
+      Registry.getInstance().register(new FakeDistributedLockManagerFactory());
+      assertInstanceOf(FakeDistributedLockManagerFactory.class, Registry.getInstance().getFactory("fake"));
+      assertInstanceOf(FakeDistributedLockManagerFactory.class, Registry.getInstance().getFactoryWithClassName("Fake"));
+      Registry.getInstance().unregisterWithType("fake");
+      assertNull(Registry.getInstance().getFactory("fake"));
+      assertThrows(IllegalArgumentException.class, () -> Registry.getInstance().getFactoryWithClassName("Fake"));
+      Registry.getInstance().register(new FakeDistributedLockManagerFactory());
+      assertInstanceOf(FakeDistributedLockManagerFactory.class, Registry.getInstance().getFactory("fake"));
+      assertInstanceOf(FakeDistributedLockManagerFactory.class, Registry.getInstance().getFactoryWithClassName("Fake"));
+      Registry.getInstance().unregisterWithClassName("Fake");
+      assertNull(Registry.getInstance().getFactory("fake"));
+      assertNull(Registry.getInstance().getFactory("Fake"));
+      assertThrows(IllegalArgumentException.class, () -> Registry.getInstance().getFactoryWithClassName("Fake"));
+
+   }
+
+   public static class FakeDistributedLockManagerFactory implements DistributedLockManagerFactory  {
+
+      @Override
+      public DistributedLockManager build(Map<String, String> properties) {
+         return null;
+      }
+
+      @Override
+      public String getName() {
+         return "fake";
+      }
+
+      @Override
+      public String getImplName() {
+         return "Fake";
+      }
+
+      @Override
+      public Set<String> getValidParametersList() {
+         return Set.of();
+      }
+   }
+
+}