backup - will be squashed later

Author: clebertsuconic

artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java

@@ -44,9 +44,9 @@ public abstract class ActiveMQScheduledComponent implements ActiveMQComponent, R
    /**
     * initialDelay < 0 would mean no initial delay, use the period instead
     */
-   private long initialDelay;
-   private long period;
-   private TimeUnit timeUnit;
+   protected long initialDelay;
+   protected long period;
+   protected TimeUnit timeUnit;
    protected final Executor executor;
    private volatile boolean isStarted;
    private ScheduledFuture future;

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzer.java

@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.artemis.utils.critical;
 
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.activemq.artemis.core.server.ActiveMQComponent;
@@ -48,4 +49,10 @@ default int getNumberOfComponents() {
    CriticalAnalyzer addAction(CriticalAction action);
 
    void check();
+
+   /** This will kick one verification at the configured timeout. If it does not finish before then
+    *  the actions will be fired. */
+   default ScheduledFuture<?> kickIt(CriticalComponent criticalComponent) {
+      return null;
+   }
 }

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java

@@ -21,6 +21,9 @@
 import java.util.ConcurrentModificationException;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
@@ -39,7 +42,7 @@ public class CriticalAnalyzerImpl implements CriticalAnalyzer {
    // one minute by default.. the server will change it for sure
    private volatile long checkTimeNanoSeconds = TimeUnit.SECONDS.toNanos(60);
 
-   private final ActiveMQScheduledComponent scheduledComponent;
+   private final CriticalAnalyzerScheduledComponent scheduledComponent;
 
    public CriticalAnalyzerImpl() {
       // this will make the scheduled component to start its own pool
@@ -49,28 +52,53 @@ public CriticalAnalyzerImpl() {
        *  otherwise in case of a deadlock, or a starvation of the server the analyzer won't pick up any
        *  issues and won't be able to shutdown the server or halt the VM
        */
-      this.scheduledComponent = new ActiveMQScheduledComponent(null, null, checkTimeNanoSeconds, TimeUnit.NANOSECONDS, false) {
+      this.scheduledComponent = new CriticalAnalyzerScheduledComponent(this, null, null, checkTimeNanoSeconds, checkTimeNanoSeconds, TimeUnit.NANOSECONDS, false);
 
-         @Override
-         public void run() {
-            logger.trace("Checking critical analyzer");
-            check();
-         }
+   }
 
-         @Override
-         protected ActiveMQThreadFactory getThreadFactory() {
-            return new ActiveMQThreadFactory("critical-analyzer", true, getThisClassLoader());
-         }
+   private static class CriticalAnalyzerScheduledComponent extends ActiveMQScheduledComponent {
 
-         private ClassLoader getThisClassLoader() {
-            return AccessController.doPrivileged((PrivilegedAction<ClassLoader>) () -> CriticalAnalyzerImpl.this.getClass().getClassLoader());
+      private final CriticalAnalyzerImpl criticalAnalyzer;
 
-         }
+      public CriticalAnalyzerScheduledComponent(CriticalAnalyzerImpl criticalAnalyzer,
+                                                ScheduledExecutorService scheduledExecutorService,
+                                                Executor executor,
+                                                long initialDelay,
+                                                long checkPeriod,
+                                                TimeUnit timeUnit,
+                                                boolean onDemand) {
+         super(scheduledExecutorService, executor, initialDelay, checkPeriod, timeUnit, onDemand);
+         this.criticalAnalyzer = criticalAnalyzer;
+      }
+
+
+      @Override
+      public void run() {
+         logger.trace("Checking critical analyzer");
+         criticalAnalyzer.check();
+      }
+
+      @Override
+      protected ActiveMQThreadFactory getThreadFactory() {
+         return new ActiveMQThreadFactory("critical-analyzer", true, getThisClassLoader());
+      }
+
+      private ClassLoader getThisClassLoader() {
+         return AccessController.doPrivileged((PrivilegedAction<ClassLoader>) () -> criticalAnalyzer.getClass().getClassLoader());
+      }
+
+      /** Schedule an action to kick in the critical analyzer.
+       *  If this future is not cancelled in time, the critical analyzer will kick in. */
+      public ScheduledFuture<?> kickIt(CriticalComponent component) {
+         return scheduledExecutorService.schedule(() -> {
+            criticalAnalyzer.fireActions(component);
+         }, period, timeUnit);
+      }
 
-      };
 
    }
 
+
    @Override
    public void clear() {
       actions.clear();
@@ -188,4 +216,9 @@ public void stop() {
    public boolean isStarted() {
       return scheduledComponent.isStarted();
    }
+
+   @Override
+   public ScheduledFuture<?> kickIt(CriticalComponent criticalComponent) {
+      return scheduledComponent.kickIt(criticalComponent);
+   }
 }

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalMeasure.java

@@ -57,7 +57,7 @@ public void close() {
       }
    };
 
-   protected static final CriticalCloseable dummyCloseable = new CriticalCloseable() {
+   public static final CriticalCloseable dummyCloseable = new CriticalCloseable() {
       @Override
       public void beforeClose(ArtemisCloseable runnable) {
          throw new IllegalStateException("The dummy closeable does not support beforeClose. Check before CriticalMeasure.isDummy(closeable) before you call beforeClose(runnable)");

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

@@ -724,8 +724,6 @@ public static String getDefaultHapolicyBackupStrategy() {
 
    private static final int DEFAULT_CLUSTER_TOPOLOGY_SCANNER_ATTEMPTS = 30;
 
-   private static final int DEFAULT_CLOSE_TIMEOUT = 30_000;
-
    /**
     * If {@code true} then the ActiveMQ Artemis Server will make use of any Protocol Managers that are in available on
     * the classpath. If false then only the core protocol will be available, unless in Embedded mode where users can
@@ -2036,8 +2034,4 @@ public static int getInitialQueueBufferSize() {
    public static int getClusterTopologyScannerAttempts() {
       return DEFAULT_CLUSTER_TOPOLOGY_SCANNER_ATTEMPTS;
    }
-
-   public static int getCloseTimeout() {
-      return DEFAULT_CLOSE_TIMEOUT;
-   }
 }

artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java

@@ -919,7 +919,6 @@ public void appendAddRecord(final long id,
                                final boolean sync,
                                final IOCompletion callback) throws Exception {
       checkJournalIsLoaded();
-      lineUpContext(callback);
 
       if (logger.isTraceEnabled()) {
          logger.trace("scheduling appendAddRecord::id={}, userRecordType={}, record = {}", id, recordType, record);
@@ -930,6 +929,8 @@ public void appendAddRecord(final long id,
 
       checkRecordSize(addRecordEncodeSize, record);
 
+      lineUpContext(callback);
+
       final SimpleFuture<Boolean> result = newSyncAndCallbackResult(sync, callback);
       appendExecutor.execute(() -> {
          journalLock.readLock().lock();
@@ -966,7 +967,6 @@ public void appendAddEvent(final long id,
                               final boolean sync,
                               final IOCompletion callback) throws Exception {
       checkJournalIsLoaded();
-      lineUpContext(callback);
 
       if (logger.isTraceEnabled()) {
          logger.trace("scheduling appendAddEvent::id={}, userRecordType={}, record = {}", id, recordType, record);
@@ -976,6 +976,8 @@ public void appendAddEvent(final long id,
 
       checkRecordSize(addRecord.getEncodeSize(), record);
 
+      lineUpContext(callback);
+
       final SimpleFuture<Boolean> result = newSyncAndCallbackResult(sync, callback);
       appendExecutor.execute(() -> {
          journalLock.readLock().lock();
@@ -1022,7 +1024,6 @@ public void appendUpdateRecord(final long id,
                                   final boolean sync,
                                   final IOCompletion callback) throws Exception {
       checkJournalIsLoaded();
-      lineUpContext(callback);
 
       if (logger.isTraceEnabled()) {
          logger.trace("scheduling appendUpdateRecord::id={}, userRecordType={}", id, recordType);
@@ -1036,6 +1037,8 @@ public void appendUpdateRecord(final long id,
          onFoundAddInfo = new SimpleFutureImpl<>();
       }
 
+      lineUpContext(callback);
+
       if (onFoundAddInfo == null) {
          internalAppendUpdateRecord(id, recordType, persister, record, false, false, null, callback);
       } else {
@@ -1142,7 +1145,6 @@ public void appendDeleteRecord(final long id, final boolean sync, final IOComple
       }
 
       checkJournalIsLoaded();
-      lineUpContext(callback);
 
       final SimpleFuture<Boolean> onFoundAddInfo;
 
@@ -1152,6 +1154,8 @@ public void appendDeleteRecord(final long id, final boolean sync, final IOComple
          onFoundAddInfo = new SimpleFutureImpl<>();
       }
 
+      lineUpContext(callback);
+
       if (onFoundAddInfo == null) {
          internalAppendDeleteRecord(id, false, null, callback);
       } else {

artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java

@@ -1538,9 +1538,4 @@ default String resolvePropertiesSources(String propertiesFileUrl) {
    default boolean isUsingDatabasePersistence() {
       return getStoreConfiguration() != null && getStoreConfiguration().getStoreType() == StoreConfiguration.StoreType.DATABASE;
    }
-
-   /** Timeout applied when closing sessions or connections. */
-   Configuration setCloseTimeout(int timeout);
-
-   int getCloseTimeout();
 }

artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java

@@ -474,8 +474,6 @@ public class ConfigurationImpl implements Configuration, Serializable {
 
    private boolean purgePageFolders = ActiveMQDefaultConfiguration.getPurgePageFolders();
 
-   private int closeTimeout = ActiveMQDefaultConfiguration.getCloseTimeout();
-
    /**
     * Parent folder for all data folders.
     */
@@ -3460,17 +3458,6 @@ public Configuration setMirrorPageTransaction(boolean ignorePageTransactions) {
       return this;
    }
 
-   @Override
-   public int getCloseTimeout() {
-      return closeTimeout;
-   }
-
-   @Override
-   public ConfigurationImpl setCloseTimeout(int closeTimeout) {
-      this.closeTimeout = closeTimeout;
-      return this;
-   }
-
    // extend property utils with ability to auto-fill and locate from collections
    // collection entries are identified by the name() property
    private static class CollectionAutoFillPropertiesUtil extends PropertyUtilsBean {

artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java

@@ -117,8 +117,6 @@ default SequentialFileFactory getJournalSequentialFileFactory() {
     */
    OperationContext getContext();
 
-   void lineUpContext();
-
    /**
     * It just creates an OperationContext without associating it
     */

artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java

@@ -1763,16 +1763,6 @@ private void insertPersistedKeyValuePair(final PersistedKeyValuePair keyValuePai
       persistedKeyValuePairs.put(keyValuePair.getKey(), keyValuePair);
    }
 
-   @Override
-   public void lineUpContext() {
-      try (ArtemisCloseable lock = closeableReadLock()) {
-         messageJournal.lineUpContext(getContext());
-      }
-   }
-
-   // ActiveMQComponent implementation
-   // ------------------------------------------------------
-
    protected abstract void beforeStart() throws Exception;
 
    @Override