ARTEMIS-TBD Closing session after a configured timeout

Author: clebertsuconic

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

@@ -35,6 +35,10 @@
 public final class ThreadDumpUtil {
    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
+   public static String threadDump() {
+      return threadDump("");
+   }
+
    public static String threadDump(final String msg) {
 
       try (

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

@@ -724,6 +724,8 @@ 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
@@ -2034,4 +2036,8 @@ public static int getInitialQueueBufferSize() {
    public static int getClusterTopologyScannerAttempts() {
       return DEFAULT_CLUSTER_TOPOLOGY_SCANNER_ATTEMPTS;
    }
+
+   public static int getCloseTimeout() {
+      return DEFAULT_CLOSE_TIMEOUT;
+   }
 }

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java

@@ -378,6 +378,8 @@ public class TransportConstants {
 
    public static final String SHUTDOWN_TIMEOUT = "shutdownTimeout";
 
+   public static final String SESSION_CLOSE_TIMEOUT = "sessionCloseTimeout";
+
    /**
     * We let this to be defined as a System Variable, as we need a different timeout over our testsuite. When running on
     * a real server, this is the default we want. When running on a test suite, we need it to be 0, You should see a
@@ -484,6 +486,7 @@ private static int parseDefaultVariable(String variableName, int defaultValue) {
       allowableAcceptorKeys.add(TransportConstants.AUTO_START);
       allowableAcceptorKeys.add(TransportConstants.ROUTER);
       allowableAcceptorKeys.add(TransportConstants.PROXY_PROTOCOL_ENABLED_PROP_NAME);
+      allowableAcceptorKeys.add(TransportConstants.SESSION_CLOSE_TIMEOUT);
 
       ALLOWABLE_ACCEPTOR_KEYS = Collections.unmodifiableSet(allowableAcceptorKeys);
 

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

@@ -1467,9 +1467,6 @@ public void appendCommitRecord(final long txID,
                                   final IOCompletion callback,
                                   final boolean lineUpContext) throws Exception {
       checkJournalIsLoaded();
-      if (lineUpContext) {
-         lineUpContext(callback);
-      }
 
       if (logger.isTraceEnabled()) {
          logger.trace("scheduling appendCommitRecord::txID={}", txID);
@@ -1480,6 +1477,9 @@ public void appendCommitRecord(final long txID,
          txcheck.checkErrorCondition();
       }
 
+      if (lineUpContext) {
+         lineUpContext(callback);
+      }
 
       final SimpleFuture<JournalTransaction> result = newSyncAndCallbackResult(sync, callback);
 
@@ -3523,4 +3523,9 @@ public void testCompact() {
    public int getCompactCount() {
       return compactCount;
    }
+
+   public void markTXError(long txID, Throwable t) {
+      JournalTransaction tx = transactions.get(txID);
+      tx.onError(-1, t.getMessage());
+   }
 }

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

@@ -1538,4 +1538,9 @@ 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,6 +474,8 @@ public class ConfigurationImpl implements Configuration, Serializable {
 
    private boolean purgePageFolders = ActiveMQDefaultConfiguration.getPurgePageFolders();
 
+   private int closeTimeout = ActiveMQDefaultConfiguration.getCloseTimeout();
+
    /**
     * Parent folder for all data folders.
     */
@@ -3458,6 +3460,17 @@ 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/remoting/impl/invm/TransportConstants.java

@@ -34,6 +34,10 @@ public final class TransportConstants {
 
    public static final String DIRECT_DELIVER = "directDeliver";
 
+   public static final String SESSION_CLOSE_TIMEOUT = "sessionCloseTimeout";
+
+   public static final int DEFAULT_SESSION_CLOSE_TIMEOUT = 30_000;
+
    private TransportConstants() {
       // Utility class
    }

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

@@ -1519,4 +1519,8 @@ 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);
+
+   //Closing session <this.name> with <conumsers.size()> consumers and transaction <tx> after <timeoutSessionClose> milliseconds
+   @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);
 }

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

@@ -32,6 +32,8 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.activemq.artemis.Closeable;
 import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
@@ -107,6 +109,7 @@
 import org.apache.activemq.artemis.utils.CompositeAddress;
 import org.apache.activemq.artemis.utils.JsonLoader;
 import org.apache.activemq.artemis.utils.PrefixUtil;
+import org.apache.activemq.artemis.utils.ThreadDumpUtil;
 import org.apache.activemq.artemis.utils.collections.TypedProperties;
 import org.apache.activemq.artemis.utils.runnables.AtomicRunnable;
 import org.apache.activemq.artemis.utils.runnables.RunnableList;
@@ -202,6 +205,9 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
    // server. Both the request and failure listener will
    // try to close one session from different threads
    // concurrently.
+   private volatile boolean closing = false;
+
+   // When the doClose is called, we will make it actually closed
    private volatile boolean closed = false;
 
    private boolean prefixEnabled = false;
@@ -339,7 +345,7 @@ public void disableSecurity() {
 
    @Override
    public boolean isClosed() {
-      return closed;
+      return closing;
    }
 
    @Override
@@ -404,6 +410,10 @@ protected void doClose(final boolean failed) throws Exception {
          callback.close(failed);
       }
       synchronized (this) {
+         if (closed) {
+            return;
+         }
+         closed = true;
          if (server.hasBrokerSessionPlugins()) {
             server.callBrokerSessionPlugins(plugin -> plugin.beforeCloseSession(this, failed));
          }
@@ -609,7 +619,7 @@ public ServerConsumer createConsumer(final long consumerID,
 
       ServerConsumer consumer;
       synchronized (this) {
-         if (closed) {
+         if (closing) {
             throw ActiveMQMessageBundle.BUNDLE.cannotCreateConsumerOnClosedSession(queueName);
          }
          consumer = new ServerConsumerImpl(consumerID, this, (QueueBinding) binding, filter, priority, started, browseOnly, storageManager, callback, preAcknowledge, strictUpdateDeliveryCount, managementService, supportLargeMessage, credits, server);
@@ -1777,37 +1787,47 @@ public void close(final boolean failed) {
    @Override
    public void close(final boolean failed, final boolean force) {
       synchronized (this) {
-         if (closed) {
+         if (closing) {
             return;
          }
-         closed = true;
+         closing = true;
       }
 
       if (force) {
          context.reset();
       }
 
+      final ScheduledFuture<?> scheduledFuture = server.getScheduledPool().schedule(() -> {
+         long txID = tx != null ? tx.getID() : -1L;
+
+         ActiveMQServerLogger.LOGGER.sessionCloseTimeout(name, server.getConfiguration().getCloseTimeout(), consumers.size(), txID, remotingConnection, ThreadDumpUtil.threadDump());
+
+         callDoClose(failed);
+      }, server.getConfiguration().getCloseTimeout(), TimeUnit.MILLISECONDS);
+
       context.executeOnCompletion(new IOCallback() {
          @Override
          public void onError(int errorCode, String errorMessage) {
-            callDoClose();
+            scheduledFuture.cancel(true);
+            callDoClose(failed);
          }
 
          @Override
          public void done() {
-            callDoClose();
-         }
-
-         private void callDoClose() {
-            try {
-               doClose(failed);
-            } catch (Exception e) {
-               ActiveMQServerLogger.LOGGER.errorClosingSession(e);
-            }
+            scheduledFuture.cancel(true);
+            callDoClose(failed);
          }
       });
    }
 
+   private void callDoClose(boolean failed) {
+      try {
+         doClose(failed);
+      } catch (Exception e) {
+         ActiveMQServerLogger.LOGGER.errorClosingSession(e);
+      }
+   }
+
    @Override
    public void closeConsumer(final long consumerID) throws Exception {
       final ServerConsumer consumer = locateConsumer(consumerID);
@@ -2220,7 +2240,7 @@ public void connectionFailed(final ActiveMQException me, boolean failedOver) {
        * This can be invoked from Netty (via channelInactive) when the connection has already been closed causing
        * spurious logging about clearing up resources for failed client connections.
        */
-      if (closed)
+      if (closing)
          return;
 
       try {

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/journal/NIOJournalCompactTest.java

@@ -496,6 +496,47 @@ public void testCompactPrepareRestart() throws Exception {
       loadAndCheck();
    }
 
+   @Test
+   public void testCommitOnRolledBack() throws Exception {
+
+      ExecutorService executorService = Executors.newSingleThreadExecutor();
+      runAfter(executorService::shutdownNow);
+
+         setup(2, 60 * 1024, false);
+
+         createJournal();
+
+         startJournal();
+
+         load();
+
+         addTx(1, 2);
+         rollback(1);
+         startCompact();
+
+         OperationContextImpl context = new OperationContextImpl(executorService);
+         journal.appendCommitRecord(1, false, context);
+         CountDownLatch latch = new CountDownLatch(1);
+         context.executeOnCompletion(new IOCallback() {
+            @Override
+            public void done() {
+               latch.countDown();
+            }
+
+            @Override
+            public void onError(int errorCode, String errorMessage) {
+               latch.countDown();
+            }
+         });
+
+         assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+         finishCompact();
+
+         stopJournal();
+   }
+
+
    @Test
    public void testCompactPrepareRestart2() throws Exception {
       setup(2, 60 * 1024, false);