From 21ae61e7ae759178fc10a62edfebe707661e695a Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 20 Aug 2026 20:50:45 -0400 Subject: [PATCH 1/2] Reapply "ARTEMIS-6142 Ensure InVM connections are removed on graceful close" This reverts commit ded1b54c538de60b4df8f4953c5e1979ac64b5db. --- .../core/remoting/impl/invm/InVMAcceptor.java | 10 +- .../remoting/impl/invm/InVMConnection.java | 21 ++-- .../remoting/impl/invm/InVMConnector.java | 10 +- .../server/impl/RemotingServiceImpl.java | 3 +- .../InVMConnectionLeakStressTest.java | 107 ++++++++++++++++++ .../impl/invm/InVMConnectionTest.java | 88 ++++++++++++++ 6 files changed, 219 insertions(+), 20 deletions(-) create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/connection/InVMConnectionLeakStressTest.java diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java index 44dcb821a012..95f905c5ded8 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java @@ -238,7 +238,7 @@ 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; } @@ -246,7 +246,11 @@ public void disconnect(final String connectionID) { Connection conn = connections.get(connectionID); if (conn != null) { - conn.disconnect(); + if (failed) { + 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); }); } } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java index 699f34e7b959..957770fcc7ba 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java +++ b/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,8 +62,6 @@ public class InVMConnection implements Connection { private final ArtemisExecutor executor; - private volatile boolean closing; - private final ActiveMQPrincipal defaultActiveMQPrincipal; private RemotingConnection protocolConnection; @@ -146,18 +144,15 @@ public void close() { } 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; } } diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnector.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnector.java index 5fe5c860667b..db44466c53fa 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnector.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnector.java @@ -214,7 +214,7 @@ public BufferHandler getHandler() { return handler; } - public void disconnect(final String connectionID) { + public void disconnect(final String connectionID, final boolean failed) { if (!started) { return; } @@ -222,7 +222,11 @@ public void disconnect(final String connectionID) { Connection conn = connections.get(connectionID); if (conn != null) { - conn.close(); + if (failed) { + conn.disconnect(); + } else { + conn.close(); + } } } @@ -267,7 +271,7 @@ public void connectionCreated(final ActiveMQComponent component, public void connectionDestroyed(final Object connectionID, boolean failed) { if (connections.remove(connectionID) != null) { // Close the corresponding connection on the other side - acceptor.disconnect((String) connectionID); + acceptor.disconnect((String) connectionID, failed); // Execute on different thread to avoid deadlocks closeExecutor.execute(() -> listener.connectionDestroyed(connectionID, failed)); diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java index ef30d9b4064f..3cff2bdc4e48 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/server/impl/RemotingServiceImpl.java +++ b/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 { diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/connection/InVMConnectionLeakStressTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/connection/InVMConnectionLeakStressTest.java new file mode 100644 index 000000000000..a68a12735249 --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/connection/InVMConnectionLeakStressTest.java @@ -0,0 +1,107 @@ +/* + * 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.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +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 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. + *

+ * Two independent defects can cause this and are guarded here: + *

    + *
  1. A race in {@code InVMConnection.internalClose} where a concurrent close could return before + * {@code connectionDestroyed} fired (fixed by making the close fully atomic).
    2. + *
  2. {@code InVMAcceptor}/{@code InVMConnector} reporting every 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.
    3. + *
+ * 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 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; + final int threads = 100; + + List> tasks = new ArrayList<>(numConnections); + for (int i = 0; i < numConnections; i++) { + tasks.add(() -> { + Connection connection = floodCf.createConnection(); + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + session.createProducer(ActiveMQJMSClient.createQueue("stress-queue")); + // Graceful close + connection.close(); + return null; + }); + } + + ExecutorService executor = Executors.newFixedThreadPool(threads); + try { + for (Future future : executor.invokeAll(tasks)) { + future.get(); + } + } finally { + executor.shutdown(); + assertTrue(executor.awaitTermination(2, TimeUnit.MINUTES)); + } + + // 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); + } +} diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/invm/InVMConnectionTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/invm/InVMConnectionTest.java index dc710389686f..48b97d0f573e 100644 --- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/invm/InVMConnectionTest.java +++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/invm/InVMConnectionTest.java @@ -16,17 +16,25 @@ */ package org.apache.activemq.artemis.tests.unit.core.remoting.impl.invm; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import org.apache.activemq.artemis.api.core.ActiveMQException; import org.apache.activemq.artemis.api.core.TransportConfiguration; import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnection; import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnectorFactory; import org.apache.activemq.artemis.core.remoting.impl.invm.TransportConstants; +import org.apache.activemq.artemis.core.server.ActiveMQComponent; +import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager; +import org.apache.activemq.artemis.spi.core.remoting.BaseConnectionLifeCycleListener; +import org.apache.activemq.artemis.spi.core.remoting.Connection; import org.junit.jupiter.api.Test; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.atomic.AtomicInteger; public class InVMConnectionTest { @@ -55,4 +63,84 @@ public void testIsTargetNode() throws Exception { assertTrue(conn.isSameTarget(tf2, tf0)); assertFalse(conn.isSameTarget(tf2, tf1)); } + + @Test + public void testConcurrentCloseFiresConnectionDestroyedExactlyOnce() throws Exception { + final int threads = 16; + // Repeat several rounds to widen the window for catching the race. + for (int round = 0; round < 50; round++) { + final CountingLifeCycleListener listener = new CountingLifeCycleListener(); + final InVMConnection conn = new InVMConnection(0, null, listener, null); + + final CyclicBarrier barrier = new CyclicBarrier(threads); + final Thread[] workers = new Thread[threads]; + final AtomicInteger prematureReturns = new AtomicInteger(); + + for (int i = 0; i < threads; i++) { + final boolean disconnect = (i % 2 == 0); + workers[i] = new Thread(() -> { + try { + // Line up all threads so they hit close()/disconnect() together. + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + if (disconnect) { + conn.disconnect(); + } else { + conn.close(); + } + // by the time any close()/disconnect() call returns, connectionDestroyed must already have fired. + if (!listener.destroyFired) { + prematureReturns.incrementAndGet(); + } + }); + } + + for (Thread worker : workers) { + worker.start(); + } + for (Thread worker : workers) { + worker.join(); + } + + assertEquals(1, listener.destroyedCount.get(), + "connectionDestroyed must be fired exactly once per connection (round " + round + ")"); + assertEquals(0, prematureReturns.get(), + "close()/disconnect() must not return before connectionDestroyed has fired (round " + round + ")"); + } + } + + private static final class CountingLifeCycleListener implements BaseConnectionLifeCycleListener { + + private final AtomicInteger destroyedCount = new AtomicInteger(); + + private volatile boolean destroyFired; + + @Override + public void connectionCreated(ActiveMQComponent component, Connection connection, ProtocolManager protocol) { + } + + @Override + public void connectionDestroyed(Object connectionID, boolean failed) { + // Hold inside the callback for a moment to widen the window in which a + // concurrent, non-atomic close() could wrongly observe the connection + // as "closing" and return early before this callback completes. + try { + Thread.sleep(5); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + destroyedCount.incrementAndGet(); + destroyFired = true; + } + + @Override + public void connectionException(Object connectionID, ActiveMQException me) { + } + + @Override + public void connectionReadyForWrites(Object connectionID, boolean ready) { + } + } } From 0bb110eb5a77d45a9a85b79a4e3660df0669ca43 Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Wed, 19 Aug 2026 19:01:03 -0400 Subject: [PATCH 2/2] ARTEMIS-6142 FollowUp to fix tests Topology discovery on InVM (for collocated stuff, that some tests still use). are using serverLocator.connect(); for those cases I'm keeping the previous semantic on calling close and disconnect. With that we will have the best scenario on each case. keeping the previous semantics for clustering while fixing the regular client usage. assisted by Claude --- .../client/impl/ClientSessionFactoryImpl.java | 2 + .../core/client/impl/ServerLocatorImpl.java | 9 ++++ .../client/impl/ServerLocatorInternal.java | 2 + .../artemis/spi/core/remoting/Connection.java | 8 +++ .../core/remoting/impl/invm/InVMAcceptor.java | 2 +- .../remoting/impl/invm/InVMConnection.java | 13 +++++ .../remoting/impl/invm/InVMConnector.java | 7 +-- .../InVMConnectionLeakStressTest.java | 50 +++++++++++-------- .../impl/invm/InVMConnectionTest.java | 42 +++++++++------- 9 files changed, 92 insertions(+), 43 deletions(-) diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionFactoryImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionFactoryImpl.java index 267d38f03b60..837587456e1a 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionFactoryImpl.java +++ b/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; diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java index b8d05736b420..5ccc5d94d476 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java +++ b/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; diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorInternal.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorInternal.java index 4338d7a9b14d..cf0b02aa8fba 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorInternal.java +++ b/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 selectNextConnectorPair(); long getNextRetryInterval(long retryInterval, double retryIntervalMultiplier, long maxRetryInterval); + + boolean isConnected(); } diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Connection.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Connection.java index a9695cc42c33..7bf3f4e8ed94 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Connection.java +++ b/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} */ diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java index 95f905c5ded8..83dbe48713ea 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMAcceptor.java @@ -246,7 +246,7 @@ public void disconnect(final String connectionID, final boolean failed) { Connection conn = connections.get(connectionID); if (conn != null) { - if (failed) { + if (failed || conn.isConnected()) { conn.disconnect(); } else { conn.close(); diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java index 957770fcc7ba..e92ce9f22359 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java @@ -66,6 +66,9 @@ public class InVMConnection implements Connection { 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; @@ -138,6 +141,16 @@ 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); diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnector.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnector.java index db44466c53fa..f14ca2f9b480 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnector.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnector.java @@ -222,7 +222,7 @@ public void disconnect(final String connectionID, final boolean failed) { Connection conn = connections.get(connectionID); if (conn != null) { - if (failed) { + if (failed || conn.isConnected()) { conn.disconnect(); } else { conn.close(); @@ -269,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, failed); + acceptor.disconnect((String) connectionID, failed || removed.isConnected()); // Execute on different thread to avoid deadlocks closeExecutor.execute(() -> listener.connectionDestroyed(connectionID, failed)); diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/connection/InVMConnectionLeakStressTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/connection/InVMConnectionLeakStressTest.java index a68a12735249..af936f5261dc 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/connection/InVMConnectionLeakStressTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/connection/InVMConnectionLeakStressTest.java @@ -19,13 +19,12 @@ import javax.jms.Connection; import javax.jms.Session; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.Callable; +import java.lang.invoke.MethodHandles; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.Future; 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; @@ -35,7 +34,10 @@ 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; /** @@ -59,6 +61,8 @@ */ public class InVMConnectionLeakStressTest extends JMSTestBase { + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + private ActiveMQConnectionFactory floodCf; @Override @@ -76,29 +80,31 @@ public void setUp() throws Exception { @Test public void testConcurrentGracefulCloseRemovesAllConnections() throws Exception { final int numConnections = 20_000; - final int threads = 100; - List> tasks = new ArrayList<>(numConnections); + 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++) { - tasks.add(() -> { - Connection connection = floodCf.createConnection(); - Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); - session.createProducer(ActiveMQJMSClient.createQueue("stress-queue")); - // Graceful close - connection.close(); - return null; + 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(); + } }); } - ExecutorService executor = Executors.newFixedThreadPool(threads); - try { - for (Future future : executor.invokeAll(tasks)) { - future.get(); - } - } finally { - executor.shutdown(); - assertTrue(executor.awaitTermination(2, TimeUnit.MINUTES)); - } + 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. diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/invm/InVMConnectionTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/invm/InVMConnectionTest.java index 48b97d0f573e..780fae8ea4f9 100644 --- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/invm/InVMConnectionTest.java +++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/remoting/impl/invm/InVMConnectionTest.java @@ -29,14 +29,19 @@ import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager; import org.apache.activemq.artemis.spi.core.remoting.BaseConnectionLifeCycleListener; import org.apache.activemq.artemis.spi.core.remoting.Connection; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; import org.junit.jupiter.api.Test; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -public class InVMConnectionTest { +public class InVMConnectionTest extends ActiveMQTestBase { @Test public void testIsTargetNode() throws Exception { @@ -67,42 +72,45 @@ public void testIsTargetNode() throws Exception { @Test public void testConcurrentCloseFiresConnectionDestroyedExactlyOnce() throws Exception { final int threads = 16; + + ExecutorService service = Executors.newFixedThreadPool(threads); + runAfter(service::shutdownNow); + // Repeat several rounds to widen the window for catching the race. for (int round = 0; round < 50; round++) { + CountDownLatch done = new CountDownLatch(threads); final CountingLifeCycleListener listener = new CountingLifeCycleListener(); final InVMConnection conn = new InVMConnection(0, null, listener, null); final CyclicBarrier barrier = new CyclicBarrier(threads); - final Thread[] workers = new Thread[threads]; final AtomicInteger prematureReturns = new AtomicInteger(); for (int i = 0; i < threads; i++) { final boolean disconnect = (i % 2 == 0); - workers[i] = new Thread(() -> { + service.execute(() -> { try { // Line up all threads so they hit close()/disconnect() together. barrier.await(); } catch (Exception e) { throw new RuntimeException(e); } - if (disconnect) { - conn.disconnect(); - } else { - conn.close(); - } - // by the time any close()/disconnect() call returns, connectionDestroyed must already have fired. - if (!listener.destroyFired) { - prematureReturns.incrementAndGet(); + try { + if (disconnect) { + conn.disconnect(); + } else { + conn.close(); + } + // by the time any close()/disconnect() call returns, connectionDestroyed must already have fired. + if (!listener.destroyFired) { + prematureReturns.incrementAndGet(); + } + } finally { + done.countDown(); } }); } - for (Thread worker : workers) { - worker.start(); - } - for (Thread worker : workers) { - worker.join(); - } + assertTrue(done.await(10, TimeUnit.SECONDS)); assertEquals(1, listener.destroyedCount.get(), "connectionDestroyed must be fired exactly once per connection (round " + round + ")");