ARTEMIS-3340 layered over ARTEMIS-2716 - activation sequence tracking to protect the journal

artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/Run.java

@@ -19,7 +19,7 @@
 import java.io.File;
 import java.util.Timer;
 import java.util.TimerTask;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import io.airlift.airline.Command;
 import io.airlift.airline.Option;
@@ -71,6 +71,7 @@ public static void setEmbedded(boolean embedded) {
    public Object execute(ActionContext context) throws Exception {
       super.execute(context);
 
+      AtomicReference<Throwable> serverActivationFailed = new AtomicReference<>();
       try {
          BrokerDTO broker = getBrokerDTO();
          ActiveMQSecurityManager securityManager = SecurityManagerFactory.create(broker.security);
@@ -110,8 +111,7 @@ public void deActivate() {
          server = BrokerFactory.createServer(broker.server, securityManager, activateCallback);
 
          server.createComponents();
-         AtomicBoolean serverActivationFailed = new AtomicBoolean(false);
-         server.getServer().registerActivationFailureListener(exception -> serverActivationFailed.set(true));
+         server.getServer().registerActivationFailureListener(exception -> serverActivationFailed.set(exception));
          server.start();
          server.getServer().addExternalComponent(managementContext, false);
 
@@ -126,14 +126,16 @@ public void deActivate() {
             server.getServer().addExternalComponent(component, true);
             assert component.isStarted();
          }
-
-         if (serverActivationFailed.get()) {
-            stop();
-         }
       } catch (Throwable t) {
          t.printStackTrace();
+         serverActivationFailed.set(t);
+      }
+
+      if (serverActivationFailed.get() != null) {
          stop();
+         return serverActivationFailed.get();
       }
+
       return new Pair<>(managementContext, server.getServer());
    }
 

artemis-commons/src/main/java/org/apache/activemq/artemis/logs/AuditLogger.java

@@ -2843,4 +2843,12 @@ static void isAutoCreated(Object source) {
    @LogMessage(level = Logger.Level.INFO)
    @Message(id = 601747, value = "User {0} is getting auto-created property on target resource: {1} {2}", format = Message.Format.MESSAGE_FORMAT)
    void isAutoCreated(String user, Object source, Object... args);
+
+   static void getActivationSequence(Object source) {
+      BASE_LOGGER.getActivationSequence(getCaller(), source);
+   }
+
+   @LogMessage(level = Logger.Level.INFO)
+   @Message(id = 601748, value = "User {0} is getting activation sequence on target resource: {1} {2}", format = Message.Format.MESSAGE_FORMAT)
+   void getActivationSequence(String user, Object source, Object... args);
 }

artemis-commons/src/test/java/org/apache/activemq/artemis/utils/ThreadLeakCheckRule.java

@@ -270,6 +270,18 @@ private boolean isExpectedThread(Thread thread) {
       } else if (threadName.contains("ObjectCleanerThread")) {
          // Required since upgrade to Netty 4.1.22 maybe because https://github.com/netty/netty/commit/739e70398ccb6b11ffa97c6b5f8d55e455a2165e
          return true;
+      } else if (threadName.contains("RMI TCP")) {
+         return true;
+      } else if (threadName.contains("RMI Scheduler")) {
+         return true;
+      } else if (threadName.contains("RMI RenewClean")) {
+         return true;
+      } else if (threadName.contains("Signal Dispatcher")) {
+         return true;
+      } else if (threadName.contains("ForkJoinPool.commonPool")) {
+         return true;
+      } else if (threadName.contains("GC Daemon")) {
+         return true;
       } else {
          for (StackTraceElement element : thread.getStackTrace()) {
             if (element.getClassName().contains("org.jboss.byteman.agent.TransformListener")) {

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

@@ -264,6 +264,9 @@ public static String getDefaultHapolicyBackupStrategy() {
    // the directory to store the journal files in
    private static String DEFAULT_JOURNAL_DIR = "data/journal";
 
+   // the directory to store the data files in
+   private static String DEFAULT_DATA_DIR = "data";
+
    // true means that the journal directory will be created
    private static boolean DEFAULT_CREATE_JOURNAL_DIR = true;
 
@@ -627,6 +630,8 @@ public static String getDefaultHapolicyBackupStrategy() {
 
    public static final String DEFAULT_TEMPORARY_QUEUE_NAMESPACE = "";
 
+   private static final String DEFAULT_DISTRIBUTED_PRIMITIVE_MANAGER_CLASS_NAME = "org.apache.activemq.artemis.quorum.zookeeper.CuratorDistributedPrimitiveManager";
+
    // Number of concurrent workers for a core bridge
    public static int DEFAULT_BRIDGE_CONCURRENCY = 1;
 
@@ -938,6 +943,13 @@ public static String getDefaultJournalDir() {
       return DEFAULT_JOURNAL_DIR;
    }
 
+   /**
+    * the directory to store the journal files in
+    */
+   public static String getDefaultDataDir() {
+      return DEFAULT_DATA_DIR;
+   }
+
    /**
     * true means that the journal directory will be created
     */
@@ -1721,6 +1733,10 @@ public static String getDefaultTemporaryQueueNamespace() {
       return DEFAULT_TEMPORARY_QUEUE_NAMESPACE;
    }
 
+   public static String getDefaultDistributedPrimitiveManagerClassName() {
+      return DEFAULT_DISTRIBUTED_PRIMITIVE_MANAGER_CLASS_NAME;
+   }
+
    public static int getDefaultBridgeConcurrency() {
       return DEFAULT_BRIDGE_CONCURRENCY;
    }

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

@@ -321,6 +321,15 @@ public interface ActiveMQServerControl {
    @Attribute(desc = "Node ID of this server")
    String getNodeID();
 
+
+   /**
+    * Returns the current activation sequence number of this server.
+    * <br>
+    * When replicated, peers may coordinate activation with this monotonic sequence
+    */
+   @Attribute(desc = "Activation sequence of this server instance")
+   long getActivationSequence();
+
    /**
     * Returns the management notification address of this server.
     * <br>

artemis-distribution/pom.xml

@@ -231,6 +231,17 @@
          <version>${project.version}</version>
          <classifier>javadoc</classifier>
       </dependency>
+      <!-- quorum -->
+      <dependency>
+         <groupId>org.apache.activemq</groupId>
+         <artifactId>artemis-quorum-api</artifactId>
+         <version>${project.version}</version>
+      </dependency>
+      <dependency>
+         <groupId>org.apache.activemq</groupId>
+         <artifactId>artemis-quorum-ri</artifactId>
+         <version>${project.version}</version>
+      </dependency>
       <dependency>
          <groupId>io.netty</groupId>
          <artifactId>netty-all</artifactId>

artemis-distribution/src/main/assembly/dep.xml

@@ -62,6 +62,9 @@
             <include>org.apache.activemq.rest:artemis-rest</include>
             <include>org.apache.qpid:qpid-jms-client</include>
             <include>io.micrometer:micrometer-core</include>
+            <!-- quorum -->
+            <include>org.apache.activemq:artemis-quorum-api</include>
+            <include>org.apache.activemq:artemis-quorum-ri</include>
 
             <!-- dependencies -->
             <include>jakarta.jms:jakarta.jms-api</include>
@@ -97,6 +100,12 @@
             <include>com.sun.xml.bind:jaxb-impl</include>
             <include>jakarta.activation:jakarta.activation-api</include>
             <include>jakarta.security.auth.message:jakarta.security.auth.message-api</include>
+            <!-- quorum -->
+            <include>org.apache.curator:curator-recipes</include>
+            <include>org.apache.curator:curator-client</include>
+            <include>org.apache.curator:curator-framework</include>
+            <include>org.apache.zookeeper:zookeeper</include>
+            <include>org.apache.zookeeper:zookeeper-jute</include>
          </includes>
          <!--excludes>
             <exclude>org.apache.activemq:artemis-website</exclude>

artemis-features/src/main/resources/features.xml

@@ -81,6 +81,7 @@
 		<!--bundle dependency="true">mvn:io.micrometer/micrometer-core/${version.micrometer}</bundle-->
 
 		<bundle>mvn:org.apache.activemq/activemq-artemis-native/${activemq-artemis-native-version}</bundle>
+		<bundle>mvn:org.apache.activemq/artemis-quorum-api/${pom.version}</bundle>
 		<bundle>mvn:org.apache.activemq/artemis-server-osgi/${pom.version}</bundle>
 	</feature>
 

artemis-quorum-api/pom.xml

@@ -0,0 +1,41 @@
+<!--
+  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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+   <modelVersion>4.0.0</modelVersion>
+
+   <parent>
+      <groupId>org.apache.activemq</groupId>
+      <artifactId>artemis-pom</artifactId>
+      <version>2.18.0-SNAPSHOT</version>
+   </parent>
+
+   <artifactId>artemis-quorum-api</artifactId>
+   <packaging>bundle</packaging>
+   <name>ActiveMQ Artemis Quorum API</name>
+
+   <properties>
+      <activemq.basedir>${project.basedir}/..</activemq.basedir>
+   </properties>
+
+   <dependencies>
+      <dependency>
+         <groupId>com.google.errorprone</groupId>
+         <artifactId>error_prone_core</artifactId>
+      </dependency>
+   </dependencies>
+</project>

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

@@ -0,0 +1,87 @@
+/*
+ * 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.quorum;
+
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
+
+public interface DistributedLock extends AutoCloseable {
+
+   String getLockId();
+
+   boolean isHeldByCaller() throws UnavailableStateException;
+
+   boolean tryLock() throws UnavailableStateException, InterruptedException;
+
+   default boolean tryLock(long timeout, TimeUnit unit) throws UnavailableStateException, InterruptedException {
+      // it doesn't make sense to be super fast
+      final long TARGET_FIRE_PERIOD_NS = TimeUnit.MILLISECONDS.toNanos(250);
+      if (timeout < 0) {
+         throw new IllegalArgumentException("timeout cannot be negative");
+      }
+      Objects.requireNonNull(unit);
+      if (timeout == 0) {
+         return tryLock();
+      }
+      final Thread currentThread = Thread.currentThread();
+      final long timeoutNs = unit.toNanos(timeout);
+      final long start = System.nanoTime();
+      final long deadline = start + timeoutNs;
+      long expectedNextFireTime = start;
+      while (!currentThread.isInterrupted()) {
+         long parkNs = expectedNextFireTime - System.nanoTime();
+         while (parkNs > 0) {
+            LockSupport.parkNanos(parkNs);
+            if (currentThread.isInterrupted()) {
+               throw new InterruptedException();
+            }
+            final long now = System.nanoTime();
+            parkNs = expectedNextFireTime - now;
+         }
+         if (tryLock()) {
+            return true;
+         }
+         final long now = System.nanoTime();
+         final long remainingTime = deadline - now;
+         if (remainingTime <= 0) {
+            return false;
+         }
+         if (remainingTime < TARGET_FIRE_PERIOD_NS) {
+            expectedNextFireTime = now;
+         } else {
+            expectedNextFireTime += TARGET_FIRE_PERIOD_NS;
+         }
+      }
+      throw new InterruptedException();
+   }
+
+   void unlock() throws UnavailableStateException;
+
+   void addListener(UnavailableLockListener listener);
+
+   void removeListener(UnavailableLockListener listener);
+
+   @FunctionalInterface
+   interface UnavailableLockListener {
+
+      void onUnavailableLockEvent();
+   }
+
+   @Override
+   void close();
+}

artemis-quorum-api/src/main/java/org/apache/activemq/artemis/quorum/DistributedPrimitiveManager.java

@@ -0,0 +1,56 @@
+/*
+ * 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.quorum;
+
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public interface DistributedPrimitiveManager extends AutoCloseable {
+
+   static DistributedPrimitiveManager newInstanceOf(String className, Map<String, String> properties) throws Exception {
+      return (DistributedPrimitiveManager) Class.forName(className).getDeclaredConstructor(Map.class).newInstance(properties);
+   }
+
+   @FunctionalInterface
+   interface UnavailableManagerListener {
+
+      void onUnavailableManagerEvent();
+   }
+
+   void addUnavailableManagerListener(UnavailableManagerListener listener);
+
+   void removeUnavailableManagerListener(UnavailableManagerListener listener);
+
+   boolean start(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException;
+
+   void start() throws InterruptedException, ExecutionException;
+
+   boolean isStarted();
+
+   void stop();
+
+   DistributedLock getDistributedLock(String lockId) throws InterruptedException, ExecutionException, TimeoutException;
+
+   MutableLong getMutableLong(String mutableLongId) throws InterruptedException, ExecutionException, TimeoutException;
+
+   @Override
+   default void close() {
+      stop();
+   }
+}