backup - will be squashed later

clebertsuconic · clebertsuconic · commit 9326858d8df7 · 2025-10-07T17:43:11.000-04:00
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java b/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;
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzer.java b/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;
+   }
 }
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java b/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);
+   }
 }
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalMeasure.java b/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)");
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java b/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;
-   }
 }
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java b/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 {
@@ -1448,13 +1452,13 @@ public void lineUpContext(IOCompletion callback) {
       }
    }
 
-   private void setErrorCondition(IOCallback otherCallback, JournalTransaction jt, Throwable t) {
+   private void setErrorCondition(IOCallback ioCallback, JournalTransaction jt, Throwable t) {
       if (jt != null) {
          jt.onError(ActiveMQExceptionType.IO_ERROR.getCode(), t.getMessage());
       }
 
-      if (otherCallback != null) {
-         otherCallback.onError(ActiveMQExceptionType.IO_ERROR.getCode(), t.getMessage());
+      if (ioCallback != null) {
+         ioCallback.onError(ActiveMQExceptionType.IO_ERROR.getCode(), t.getMessage());
       }
    }
 
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/Configuration.java b/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();
 }
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImpl.java b/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 {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java b/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
     */
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java b/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
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
@@ -647,10 +647,6 @@ public long storePageCounterInc(final long queueID, final int add, final long si
    public void commit(final long txID, final boolean lineUpContext) throws Exception {
    }
 
-   @Override
-   public void lineUpContext() {
-   }
-
    @Override
    public void deletePendingLargeMessage(final long recordID) throws Exception {
    }
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
@@ -1519,7 +1519,4 @@ void slowConsumerDetected(String sessionID,
 
    @LogMessage(id = 224152, value = "Closing connection from {} for PROXY Protocol violation. Acceptor {} uses proxyProtocolEnabled={}, but connection {} PROXY Protocol.", level = LogMessage.Level.WARN)
    void proxyProtocolViolation(String remoteAddress, String acceptorName, boolean proxyProtocolEnabled, String actualUsage);
-
-   @LogMessage(id = 224153, value = "Timed out closing ServerSession {} after {} milliseconds with {} consumers, txID={}, connection={},\n{}", level = LogMessage.Level.WARN)
-   void sessionCloseTimeout(String name, long timeout, int consumers, long txID, Object connection, String threadDump);
 }
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/JournalImplTestBase.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/JournalImplTestBase.java
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/SessionCloseTimeoutTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/persistence/SessionCloseTimeoutTest.java
diff --git a/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/journal/JournalImplTestUnit.java b/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/journal/JournalImplTestUnit.java

Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,7 @@ public void close() {`
`57`	`57`	`}`
`58`	`58`	`};`
`59`	`59`
`60`		`- protected static final CriticalCloseable dummyCloseable = new CriticalCloseable() {`
	`60`	`+ public static final CriticalCloseable dummyCloseable = new CriticalCloseable() {`
`61`	`61`	`@Override`
`62`	`62`	`public void beforeClose(ArtemisCloseable runnable) {`
`63`	`63`	`throw new IllegalStateException("The dummy closeable does not support beforeClose. Check before CriticalMeasure.isDummy(closeable) before you call beforeClose(runnable)");`
Original file line number	Diff line number	Diff line change
`@@ -1538,9 +1538,4 @@ default String resolvePropertiesSources(String propertiesFileUrl) {`
`1538`	`1538`	`default boolean isUsingDatabasePersistence() {`
`1539`	`1539`	`return getStoreConfiguration() != null && getStoreConfiguration().getStoreType() == StoreConfiguration.StoreType.DATABASE;`
`1540`	`1540`	`}`
`1541`		`-`
`1542`		`- /** Timeout applied when closing sessions or connections. */`
`1543`		`- Configuration setCloseTimeout(int timeout);`
`1544`		`-`
`1545`		`- int getCloseTimeout();`
`1546`	`1541`	`}`