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 44dcb821a012..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 @@ -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.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); }); } } 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..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 @@ -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; } } 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..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 @@ -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.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)); 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..af936f5261dc --- /dev/null +++ b/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. + *

+ * 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 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); + } +} 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..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 @@ -16,19 +16,32 @@ */ 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.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 { @@ -55,4 +68,87 @@ 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; + + 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 AtomicInteger prematureReturns = new AtomicInteger(); + + for (int i = 0; i < threads; i++) { + final boolean disconnect = (i % 2 == 0); + service.execute(() -> { + try { + // Line up all threads so they hit close()/disconnect() together. + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + 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(); + } + }); + } + + assertTrue(done.await(10, TimeUnit.SECONDS)); + + 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) { + } + } }