ARTEMIS-6142 FollowUp to fix tests

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionFactoryImpl.java

@@ -1201,6 +1201,8 @@ protected Connection openTransportConnection(final Connector connector) {
             connector.close();
          } catch (Throwable t) {
          }
+      } else if (serverLocator.isConnected() || serverLocator.isHA()) {
+         transportConnection.setConnected();
       }
 
       return transportConnection;

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java

@@ -131,6 +131,9 @@ private enum STATE {
     */
    private volatile boolean disableDiscoveryRetries = false;
 
+   // set when connect() is called, meaning this locator is used for clustering or topology discovery
+   private volatile boolean connected = false;
+
    // if the system should shutdown the pool when shutting down
    private transient boolean shutdownPool;
 
@@ -552,6 +555,7 @@ private ClientSessionFactoryInternal connect(final boolean skipWarnings) throws
       // if we used connect, we should control UDP reconnections at a different path.
       // and this belongs to a cluster connection, not client
       disableDiscoveryRetries = true;
+      connected = true;
       ClientSessionFactoryInternal returnFactory = null;
 
       synchronized (this) {
@@ -1382,6 +1386,11 @@ public boolean isClusterConnection() {
       return clusterConnection;
    }
 
+   @Override
+   public boolean isConnected() {
+      return connected;
+   }
+
    @Override
    public TransportConfiguration getClusterTransportConfiguration() {
       return clusterTransportConfiguration;

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorInternal.java

@@ -99,4 +99,6 @@ default void notifyNodeDown(long uniqueEventID, String nodeID) {
    Pair<TransportConfiguration, TransportConfiguration> selectNextConnectorPair();
 
    long getNextRetryInterval(long retryInterval, double retryIntervalMultiplier, long maxRetryInterval);
+
+   boolean isConnected();
 }

artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Connection.java

@@ -134,6 +134,14 @@ default void disconnect() {
       close();
    }
 
+   // Marks this connection as being used for clustering or topology discovery
+   default void setConnected() {
+   }
+
+   default boolean isConnected() {
+      return false;
+   }
+
    /**
     * {@return the string representation of the remote address this connection is connected to}
     */

artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java

@@ -238,15 +238,19 @@ public void connect(final String connectionID,
       connectionListener.connectionCreated(this, inVMConnection, protocolMap.get(ActiveMQClient.DEFAULT_CORE_PROTOCOL));
    }
 
-   public void disconnect(final String connectionID) {
+   public void disconnect(final String connectionID, final boolean failed) {
       if (!started) {
          return;
       }
 
       Connection conn = connections.get(connectionID);
 
       if (conn != null) {
-         conn.disconnect();
+         if (failed || conn.isConnected()) {
+            conn.disconnect();
+         } else {
+            conn.close();
+         }
       }
    }
 
@@ -301,7 +305,7 @@ public void connectionDestroyed(final Object connectionID, boolean failed) {
             // Execute on different thread after all the packets are sent, to avoid deadlocks
             connection.getExecutor().execute(() -> {
                // Remove on the other side too
-               connector.disconnect((String) connectionID);
+               connector.disconnect((String) connectionID, failed);
             });
          }
       }

artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java

@@ -53,7 +53,7 @@ public class InVMConnection implements Connection {
 
    private final String id;
 
-   private boolean closed;
+   private volatile boolean closed;
 
    // Used on tests
    private static boolean flushEnabled = true;
@@ -62,12 +62,13 @@ public class InVMConnection implements Connection {
 
    private final ArtemisExecutor executor;
 
-   private volatile boolean closing;
-
    private final ActiveMQPrincipal defaultActiveMQPrincipal;
 
    private RemotingConnection protocolConnection;
 
+   // set when this connection is used for clustering or topology discovery
+   private volatile boolean connected;
+
    private boolean bufferPoolingEnabled = TransportConstants.DEFAULT_BUFFER_POOLING;
 
    private boolean directDeliver = TransportConstants.DEFAULT_DIRECT_DELIVER;
@@ -140,24 +141,31 @@ public void setProtocolConnection(RemotingConnection connection) {
       this.protocolConnection = connection;
    }
 
+   @Override
+   public void setConnected() {
+      this.connected = true;
+   }
+
+   @Override
+   public boolean isConnected() {
+      return connected;
+   }
+
    @Override
    public void close() {
       internalClose(false);
    }
 
    private void internalClose(boolean failed) {
-      if (closing) {
-         return;
-      }
-
-      closing = true;
-
+      // guarantee connectionDestroyed  is fired exactly once
       synchronized (this) {
-         if (!closed) {
-            listener.connectionDestroyed(id, failed);
-
-            closed = true;
+         if (closed) {
+            return;
          }
+
+         listener.connectionDestroyed(id, failed);
+
+         closed = true;
       }
    }
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnector.java

@@ -214,15 +214,19 @@ public BufferHandler getHandler() {
       return handler;
    }
 
-   public void disconnect(final String connectionID) {
+   public void disconnect(final String connectionID, final boolean failed) {
       if (!started) {
          return;
       }
 
       Connection conn = connections.get(connectionID);
 
       if (conn != null) {
-         conn.close();
+         if (failed || conn.isConnected()) {
+            conn.disconnect();
+         } else {
+            conn.close();
+         }
       }
    }
 
@@ -265,9 +269,10 @@ public void connectionCreated(final ActiveMQComponent component,
 
       @Override
       public void connectionDestroyed(final Object connectionID, boolean failed) {
-         if (connections.remove(connectionID) != null) {
+         Connection removed = connections.remove(connectionID);
+         if (removed != null) {
             // Close the corresponding connection on the other side
-            acceptor.disconnect((String) connectionID);
+            acceptor.disconnect((String) connectionID, failed || removed.isConnected());
 
             // Execute on different thread to avoid deadlocks
             closeExecutor.execute(() -> listener.connectionDestroyed(connectionID, failed));

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

@@ -815,7 +815,8 @@ private void issueFailure(Object connectionID, ActiveMQException e) {
    private void issueClose(Object connectionID) {
       ConnectionEntry conn = connections.get(connectionID);
 
-      if (conn != null && !conn.connection.isSupportReconnect()) {
+      // always remove connection on graceful close
+      if (conn != null) {
          RemotingConnection removedConnection = removeConnection(connectionID);
          if (removedConnection != null) {
             try {

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/connection/InVMConnectionLeakStressTest.java

@@ -0,0 +1,113 @@
+/*
+ * 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.tests.integration.jms.connection;
+
+import javax.jms.Connection;
+import javax.jms.Session;
+
+import java.lang.invoke.MethodHandles;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.activemq.artemis.api.core.TransportConfiguration;
+import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
+import org.apache.activemq.artemis.api.jms.JMSFactoryType;
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.apache.activemq.artemis.tests.util.JMSTestBase;
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Attempts to reproduce a server-side InVM connection leak where {@code RemotingServiceImpl.removeConnection} is not
+ * invoked even though {@code ActiveMQConnection.close()} was called.
+ * <p>
+ * Two independent defects can cause this and are guarded here:
+ * <ol>
+ *    <li>A race in {@code InVMConnection.internalClose} where a concurrent close could return before
+ *    {@code connectionDestroyed} fired (fixed by making the close fully atomic).</li>
+ *    <li>{@code InVMAcceptor}/{@code InVMConnector} reporting <em>every</em> close (including a graceful
+ *    {@code close()}) to the server as a failure, which routed it through {@code issueFailure} where the
+ *    {@code isSupportReconnect()} guard could skip removal. When the client enables a confirmation window the
+ *    server-side connection reports {@code isSupportReconnect() == true}; if the session channel is still present at
+ *    transport-teardown time the connection was never removed, and because the InVM connection-ttl is -1 the
+ *    failure-check reaper never removes it either - a permanent leak.</li>
+ * </ol>
+ * The bug is timing dependent (it depends on the ordering of {@code SESS_CLOSE} processing versus transport teardown),
+ * so this test floods the broker with many short-lived reconnect-capable connections to widen the race window. It may
+ * not fail on every machine, but on hardware/timing where the race is hit it will leave server connections behind.
+ */
+public class InVMConnectionLeakStressTest extends JMSTestBase {
+
+   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+   private ActiveMQConnectionFactory floodCf;
+
+   @Override
+   @BeforeEach
+   public void setUp() throws Exception {
+      super.setUp();
+
+      floodCf = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(INVM_CONNECTOR_FACTORY));
+      // A positive confirmation window makes the server-side connection report isSupportReconnect() == true, which is
+      // what used to make issueFailure()/issueClose() skip removeConnection().
+      floodCf.setConfirmationWindowSize(1024 * 1024);
+      floodCf.setReconnectAttempts(-1);
+   }
+
+   @Test
+   public void testConcurrentGracefulCloseRemovesAllConnections() throws Exception {
+      final int numConnections = 20_000;
+
+      AtomicInteger error = new AtomicInteger(0);
+      CountDownLatch latch = new CountDownLatch(numConnections);
+      ExecutorService executor = Executors.newFixedThreadPool(100);
+      runAfter(executor::shutdownNow);
+
+      for (int i = 0; i < numConnections; i++) {
+         executor.execute(() -> {
+            try {
+               Connection connection = floodCf.createConnection();
+               Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+               session.createProducer(ActiveMQJMSClient.createQueue("stress-queue"));
+               // Graceful close
+               connection.close();
+            } catch (Exception e) {
+               logger.warn(e.getMessage(), e);
+               error.incrementAndGet();
+            } finally {
+               latch.countDown();
+            }
+         });
+      }
+
+      assertTrue(latch.await(10, TimeUnit.SECONDS));
+      assertEquals(0, error.get());
+
+      // Every gracefully-closed connection must be removed from the server. InVM connection-ttl is -1 so the
+      // failure-check reaper never removes them; if this never reaches 0 the connections have leaked.
+      Wait.assertEquals(0, () -> server.getRemotingService().getConnectionCount(), 10_000);
+   }
+}