This is an automated email from the ASF dual-hosted git repository.
mattrpav pushed a commit to branch activemq-6.3.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-6.3.x by this push:
new 3762e0898a Fix network bridge local-side close (#2475) (#2503)
3762e0898a is described below
commit 3762e0898ae4f60a5a4f1199a5cff35b9394ccbd
Author: Matt Pavlovich <[email protected]>
AuthorDate: Fri Aug 28 12:30:07 2026 -0500
Fix network bridge local-side close (#2475) (#2503)
* Fix: Perform full lifecycle network bridge shutdown on an administrative
stop
* Update transport connection stop to pass exception (#7)
* Update transport connection stop to pass exception
This updates the transport stop to pass any transport exception to the
transport during the stop() method call. The transport can optionally
decide to use it. In this case the VMTransport has been updated to
include the exception as part of the ShutdownInfo command so that
bridges know whether stop() was called normally or because of an error.
The use of this could be expanded to other transports in the future
Test fixes:
---------
(cherry picked from commit 9156668b2e4ce207d58cedcf64bce5b928ca7704)
Co-authored-by: Christopher L. Shannon <[email protected]>
---
.../activemq/broker/TransportConnection.java | 2 +-
.../network/DemandForwardingBridgeSupport.java | 11 +-
.../apache/activemq/transport/vm/VMTransport.java | 7 +-
.../org/apache/activemq/command/ShutdownInfo.java | 21 +-
.../org/apache/activemq/transport/Transport.java | 1 +
.../apache/activemq/transport/TransportFilter.java | 5 +
.../activemq/transport/TransportSupport.java | 5 +
.../transport/failover/FailoverTransport.java | 5 +
.../activemq/transport/fanout/FanoutTransport.java | 5 +
.../activemq/transport/mock/MockTransport.java | 5 +
...dgeDuplexLocalConnectionCloseReconnectTest.java | 230 ++++++++++++++
...orkBridgeLocalConnectionCloseReconnectTest.java | 331 +++++++++++++++++++++
...etworkBridgeSlowConsumerAbortReconnectTest.java | 164 ++++++++++
13 files changed, 786 insertions(+), 6 deletions(-)
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
index 26a14c8cfe..22f3d972f7 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
@@ -1260,7 +1260,7 @@ public class TransportConnection implements Connection,
Task, CommandVisitor {
LOG.trace("Exception caught stopping. This exception is ignored.",
ignore);
}
try {
- transport.stop();
+ transport.stop(transportException.get());
LOG.debug("Stopped transport: {}", transport.getRemoteAddress());
} catch (Exception e) {
LOG.debug("Could not stop transport to {}. This exception is
ignored.", transport.getRemoteAddress(), e);
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
index 86dda092d4..9db3fc8b56 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
@@ -98,6 +98,7 @@ import
org.apache.activemq.transport.TransportDisposedIOException;
import org.apache.activemq.transport.TransportFilter;
import org.apache.activemq.transport.failover.FailoverTransport;
import org.apache.activemq.transport.tcp.TcpTransport;
+import org.apache.activemq.util.IOExceptionSupport;
import org.apache.activemq.util.IdGenerator;
import org.apache.activemq.util.IntrospectionSupport;
import org.apache.activemq.util.LongSequenceGenerator;
@@ -1328,8 +1329,14 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
} else if (command.isBrokerInfo()) {
futureLocalBrokerInfo.set((BrokerInfo) command);
} else if (command.isShutdownInfo()) {
- LOG.info("{} Shutting down {}",
configuration.getBrokerName(), configuration.getName());
- stop();
+ ShutdownInfo info = (ShutdownInfo) command;
+ if (brokerService.isStopping() ||
brokerService.isStopped() || info.getError() == null) {
+ LOG.info("{} Shutting down {}",
configuration.getBrokerName(), configuration.getName());
+ stop();
+ } else {
+ // Administrative shutdown via .stop() or
SlowConsumerStrategy needs lifecycle clean-up
+
serviceLocalException(IOExceptionSupport.create(info.getError()));
+ }
} else if (command.getClass() == ConnectionError.class) {
ConnectionError ce = (ConnectionError) command;
serviceLocalException(ce.getException());
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
b/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
index 9ef7860ecb..972c344898 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
@@ -196,6 +196,11 @@ public class VMTransport implements Transport, Task {
@Override
public void stop() throws Exception {
+ stop(null);
+ }
+
+ @Override
+ public void stop(Throwable exception) throws Exception {
// Only need to do this once, all future oneway calls will now
// fail as will any asnyc jobs in the task runner.
if (disposed.compareAndSet(false, true)) {
@@ -224,7 +229,7 @@ public class VMTransport implements Transport, Task {
// to cleanly shutdown the async tasks so that this is the last
// command it see's.
try {
- peer.transportListener.onCommand(new ShutdownInfo());
+ peer.transportListener.onCommand(new
ShutdownInfo(exception));
} catch (Exception ignore) {
}
diff --git
a/activemq-client/src/main/java/org/apache/activemq/command/ShutdownInfo.java
b/activemq-client/src/main/java/org/apache/activemq/command/ShutdownInfo.java
index 1a74317c23..dfc81141cd 100644
---
a/activemq-client/src/main/java/org/apache/activemq/command/ShutdownInfo.java
+++
b/activemq-client/src/main/java/org/apache/activemq/command/ShutdownInfo.java
@@ -19,14 +19,24 @@ package org.apache.activemq.command;
import org.apache.activemq.state.CommandVisitor;
/**
- *
+ *
* @openwire:marshaller code="11"
- *
+ *
*/
public class ShutdownInfo extends BaseCommand {
public static final byte DATA_STRUCTURE_TYPE = CommandTypes.SHUTDOWN_INFO;
+ private transient Throwable error;
+
+ public ShutdownInfo() {
+
+ }
+
+ public ShutdownInfo(Throwable error) {
+ this.error = error;
+ }
+
public byte getDataStructureType() {
return DATA_STRUCTURE_TYPE;
}
@@ -39,4 +49,11 @@ public class ShutdownInfo extends BaseCommand {
return true;
}
+ public Throwable getError() {
+ return error;
+ }
+
+ public void setError(Throwable error) {
+ this.error = error;
+ }
}
diff --git
a/activemq-client/src/main/java/org/apache/activemq/transport/Transport.java
b/activemq-client/src/main/java/org/apache/activemq/transport/Transport.java
index 2067c14412..ee4c0f379f 100644
--- a/activemq-client/src/main/java/org/apache/activemq/transport/Transport.java
+++ b/activemq-client/src/main/java/org/apache/activemq/transport/Transport.java
@@ -164,4 +164,5 @@ public interface Transport extends Service {
*/
WireFormat getWireFormat();
+ void stop(Throwable exception) throws Exception;
}
diff --git
a/activemq-client/src/main/java/org/apache/activemq/transport/TransportFilter.java
b/activemq-client/src/main/java/org/apache/activemq/transport/TransportFilter.java
index ce02a7ab1a..9034c68b17 100644
---
a/activemq-client/src/main/java/org/apache/activemq/transport/TransportFilter.java
+++
b/activemq-client/src/main/java/org/apache/activemq/transport/TransportFilter.java
@@ -72,6 +72,11 @@ public class TransportFilter implements TransportListener,
Transport {
next.stop();
}
+ @Override
+ public void stop(Throwable exception) throws Exception {
+ next.stop(exception);
+ }
+
@Override
public void onCommand(Object command) {
transportListener.onCommand(command);
diff --git
a/activemq-client/src/main/java/org/apache/activemq/transport/TransportSupport.java
b/activemq-client/src/main/java/org/apache/activemq/transport/TransportSupport.java
index cdf1174b56..583a5987b1 100644
---
a/activemq-client/src/main/java/org/apache/activemq/transport/TransportSupport.java
+++
b/activemq-client/src/main/java/org/apache/activemq/transport/TransportSupport.java
@@ -135,4 +135,9 @@ public abstract class TransportSupport extends
ServiceSupport implements Transpo
return isStarted();
}
+ @Override
+ public void stop(Throwable exception) throws Exception {
+ stop();
+ }
+
}
diff --git
a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
index 7ab3b18818..379e8aa371 100644
---
a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
+++
b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
@@ -373,6 +373,11 @@ public class FailoverTransport implements
CompositeTransport {
}
}
+ @Override
+ public void stop(Throwable exception) throws Exception {
+ stop();
+ }
+
@Override
public void stop() throws Exception {
Transport transportToStop = null;
diff --git
a/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
b/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
index fa618e0889..d2e2f30cd4 100644
---
a/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
+++
b/activemq-client/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
@@ -295,6 +295,11 @@ public class FanoutTransport implements CompositeTransport
{
}
}
+ @Override
+ public void stop(Throwable exception) throws Exception {
+ stop();
+ }
+
@Override
public void stop() throws Exception {
try {
diff --git
a/activemq-client/src/main/java/org/apache/activemq/transport/mock/MockTransport.java
b/activemq-client/src/main/java/org/apache/activemq/transport/mock/MockTransport.java
index 8b00e27bc0..567f411f3f 100644
---
a/activemq-client/src/main/java/org/apache/activemq/transport/mock/MockTransport.java
+++
b/activemq-client/src/main/java/org/apache/activemq/transport/mock/MockTransport.java
@@ -71,6 +71,11 @@ public class MockTransport extends DefaultTransportListener
implements Transport
getNext().stop();
}
+ @Override
+ public void stop(Throwable exception) throws Exception {
+ getNext().stop(exception);
+ }
+
@Override
public void onCommand(Object command) {
getTransportListener().onCommand(command);
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkBridgeDuplexLocalConnectionCloseReconnectTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkBridgeDuplexLocalConnectionCloseReconnectTest.java
new file mode 100644
index 0000000000..f14b4bdd29
--- /dev/null
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkBridgeDuplexLocalConnectionCloseReconnectTest.java
@@ -0,0 +1,230 @@
+/**
+ * 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.network;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.jms.Connection;
+import jakarta.jms.Session;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnection;
+import org.apache.activemq.broker.region.RegionBroker;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.test.annotations.ParallelTest;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ShutdownInfo handling for a duplex network bridge, on both sides.
+ *
+ * A duplex bridge has two halves, each with its own local vm:// connection:
the initiator's
+ * bridge on the initiating broker, and the responder's bridge embedded in the
inbound
+ * TransportConnection on the accepting broker. Closing either vm://
connection server-side while
+ * its broker keeps running delivers a ShutdownInfo to that half, and both
halves should fail and
+ * recover so message flow resumes in both directions. An orderly broker
shutdown, which sends
+ * ShutdownInfo over the remote side of the shared connection, should still
stop the bridge without
+ * trying to reconnect to the stopped broker.
+ */
+@Category(ParallelTest.class)
+public class NetworkBridgeDuplexLocalConnectionCloseReconnectTest {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(NetworkBridgeDuplexLocalConnectionCloseReconnectTest.class);
+
+ // one queue per direction, consumed only on the receiving side; a
consumer on the producing
+ // side would compete with the bridge's subscription for the same messages
+ private static final String QUEUE_TO_RESPONDER =
"BRIDGE.DUPLEX.TO.RESPONDER";
+ private static final String QUEUE_TO_INITIATOR =
"BRIDGE.DUPLEX.TO.INITIATOR";
+
+ private BrokerService initiatorBroker;
+ private BrokerService responderBroker;
+ private Connection initiatorConnection;
+ private Connection responderConnection;
+ private final AtomicInteger initiatorReceived = new AtomicInteger();
+ private final AtomicInteger responderReceived = new AtomicInteger();
+ private NetworkConnector nc;
+
+ @Before
+ public void setUp() throws Exception {
+ responderBroker = new BrokerService();
+ responderBroker.setBrokerName("responder");
+ responderBroker.setPersistent(false);
+ responderBroker.setUseJmx(false);
+ responderBroker.addConnector("tcp://127.0.0.1:0");
+ responderBroker.start();
+ responderBroker.waitUntilStarted();
+
+ initiatorBroker = new BrokerService();
+ initiatorBroker.setBrokerName("initiator");
+ initiatorBroker.setPersistent(false);
+ initiatorBroker.setUseJmx(false);
+ initiatorBroker.addConnector("tcp://127.0.0.1:0");
+ initiatorBroker.start();
+ initiatorBroker.waitUntilStarted();
+
+ var responderUri =
responderBroker.getTransportConnectors().get(0).getPublishableConnectURI();
+ nc = initiatorBroker.addNetworkConnector("static:(" + responderUri +
")");
+ nc.setName("duplex-to-responder");
+ nc.setDuplex(true);
+ // static inclusion is mirrored onto the responder half via the
BrokerInfo config, so
+ // forwarding works in both directions; this test covers lifecycle,
not demand propagation
+ nc.addStaticallyIncludedDestination(new
ActiveMQQueue(QUEUE_TO_RESPONDER));
+ nc.addStaticallyIncludedDestination(new
ActiveMQQueue(QUEUE_TO_INITIATOR));
+ nc.start();
+ assertTrue("duplex bridge should establish",
+ Wait.waitFor(() -> !nc.activeBridges().isEmpty(), 20_000, 10));
+
+ // consumers on both brokers create demand in both directions across
the one duplex bridge
+ responderConnection = new
ActiveMQConnectionFactory(responderUri).createConnection();
+ responderConnection.start();
+ responderConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)
+ .createConsumer(new ActiveMQQueue(QUEUE_TO_RESPONDER))
+ .setMessageListener(m -> responderReceived.incrementAndGet());
+
+ var initiatorUri =
initiatorBroker.getTransportConnectors().get(0).getPublishableConnectURI();
+ initiatorConnection = new
ActiveMQConnectionFactory(initiatorUri).createConnection();
+ initiatorConnection.start();
+ initiatorConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)
+ .createConsumer(new ActiveMQQueue(QUEUE_TO_INITIATOR))
+ .setMessageListener(m -> initiatorReceived.incrementAndGet());
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ for (var c : new Connection[] { initiatorConnection,
responderConnection }) {
+ if (c != null) {
+ try { c.close(); } catch (Exception ignored) {}
+ }
+ }
+ if (initiatorBroker != null) {
+ initiatorBroker.stop();
+ initiatorBroker.waitUntilStopped();
+ }
+ if (responderBroker != null) {
+ responderBroker.stop();
+ responderBroker.waitUntilStopped();
+ }
+ }
+
+ /** Server-side close of the initiator half's local vm:// connection. */
+ @Test(timeout = 120_000)
+ public void testDuplexBridgeReconnectsAfterInitiatorLocalConnectionClose()
throws Exception {
+ assertFlowBothWays("before initiator-side close");
+
+ var oldBridge = nc.activeBridges().iterator().next();
+ var vmConnection = findVmConnection(initiatorBroker);
+ assertNotNull("expected the initiator bridge's local vm://
connection", vmConnection);
+ LOG.info("stopping initiator-side local connection server-side: {}",
vmConnection);
+ vmConnection.serviceException(new IOException("stopping initiator-side
local connection server-side"));
+
+ awaitNewBridge(oldBridge, "after initiator-side close");
+ assertFlowBothWays("after initiator-side close and reconnect");
+ }
+
+ /** Server-side close of the responder half's local vm:// connection. */
+ @Test(timeout = 120_000)
+ public void testDuplexBridgeReconnectsAfterResponderLocalConnectionClose()
throws Exception {
+ assertFlowBothWays("before responder-side close");
+
+ var oldBridge = nc.activeBridges().iterator().next();
+ var vmConnection = findVmConnection(responderBroker);
+ assertNotNull("expected the responder bridge half's local vm://
connection", vmConnection);
+ LOG.info("stopping responder-side local connection server-side: {}",
vmConnection);
+ vmConnection.stop();
+
+ awaitNewBridge(oldBridge, "after responder-side close");
+ assertFlowBothWays("after responder-side close and reconnect");
+ }
+
+ /**
+ * An orderly responder broker shutdown sends ShutdownInfo over the remote
side. The initiator
+ * bridge should stop and stay stopped, not linger in activeBridges
retrying against the
+ * stopped broker.
+ */
+ @Test(timeout = 120_000)
+ public void testDuplexBridgeStopsOnResponderBrokerShutdown() throws
Exception {
+ assertFlowBothWays("before responder broker shutdown");
+
+ responderBroker.stop();
+ responderBroker.waitUntilStopped();
+
+ assertTrue("initiator bridge should be gone after an orderly responder
broker shutdown",
+ Wait.waitFor(() -> nc.activeBridges().isEmpty(), 20_000, 10));
+ }
+
+ private void awaitNewBridge(NetworkBridge oldBridge, String label) throws
Exception {
+ assertTrue("duplex bridge must reconnect with a new instance " + label,
+ Wait.waitFor(() -> {
+ for (var bridge : nc.activeBridges()) {
+ if (bridge != oldBridge) {
+ return true;
+ }
+ }
+ return false;
+ }, 30_000, 10));
+ assertTrue("old duplex bridge instance must be deregistered " + label,
+ Wait.waitFor(() -> nc.activeBridges().size() == 1
+ && !nc.activeBridges().contains(oldBridge), 20_000,
10));
+ LOG.info("duplex bridge reconnected {}", label);
+ }
+
+ /** Messages must cross both ways over the duplex bridge. */
+ private void assertFlowBothWays(String label) throws Exception {
+ var responderBefore = responderReceived.get();
+ produce(initiatorBroker, QUEUE_TO_RESPONDER, 5);
+ assertTrue("initiator->responder flow " + label + " (responder had " +
responderBefore + ")",
+ Wait.waitFor(() -> responderReceived.get() >= responderBefore
+ 5, 30_000, 10));
+
+ var initiatorBefore = initiatorReceived.get();
+ produce(responderBroker, QUEUE_TO_INITIATOR, 5);
+ assertTrue("responder->initiator flow " + label + " (initiator had " +
initiatorBefore + ")",
+ Wait.waitFor(() -> initiatorReceived.get() >= initiatorBefore
+ 5, 30_000, 10));
+ }
+
+ private void produce(BrokerService broker, String queueName, int n) throws
Exception {
+ try (var c = new ActiveMQConnectionFactory(
+
broker.getTransportConnectors().get(0).getPublishableConnectURI()).createConnection();
+ var session = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ var producer = session.createProducer(new
ActiveMQQueue(queueName))) {
+ c.start();
+
+ for (var i = 0; i < n; i++) {
+ producer.send(session.createTextMessage("m-" + i));
+ }
+ }
+ }
+
+ private TransportConnection findVmConnection(BrokerService broker) throws
Exception {
+ for (var connection : ((RegionBroker)
broker.getRegionBroker()).getClients()) {
+ var address = connection.getRemoteAddress();
+ if (address != null && address.startsWith("vm:") && connection
instanceof TransportConnection) {
+ return (TransportConnection) connection;
+ }
+ }
+ return null;
+ }
+}
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkBridgeLocalConnectionCloseReconnectTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkBridgeLocalConnectionCloseReconnectTest.java
new file mode 100644
index 0000000000..6c03f50b25
--- /dev/null
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkBridgeLocalConnectionCloseReconnectTest.java
@@ -0,0 +1,331 @@
+/**
+ * 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.network;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.jms.Connection;
+import jakarta.jms.Session;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnection;
+import org.apache.activemq.broker.region.RegionBroker;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ShutdownInfo;
+import org.apache.activemq.test.annotations.ParallelTest;
+import org.apache.activemq.transport.InactivityIOException;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * When a network bridge's local connection is closed server-side while the
broker keeps running,
+ * the bridge should fail and reconnect.
+ */
+@Category(ParallelTest.class)
+public class NetworkBridgeLocalConnectionCloseReconnectTest {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(NetworkBridgeLocalConnectionCloseReconnectTest.class);
+
+ private static final String QUEUE_NAME = "BRIDGE.RECONNECT.TEST";
+
+ private BrokerService localBroker;
+ private BrokerService remoteBroker;
+ private Connection remoteConnection;
+ private final AtomicInteger remoteReceived = new AtomicInteger();
+
+ @Before
+ public void setUp() throws Exception {
+ remoteBroker = new BrokerService();
+ remoteBroker.setBrokerName("remote");
+ remoteBroker.setPersistent(false);
+ remoteBroker.setUseJmx(false);
+ remoteBroker.addConnector("tcp://127.0.0.1:0");
+ remoteBroker.start();
+ remoteBroker.waitUntilStarted();
+
+ localBroker = new BrokerService();
+ localBroker.setBrokerName("local");
+ localBroker.setPersistent(false);
+ localBroker.setUseJmx(false);
+ localBroker.start();
+ localBroker.waitUntilStarted();
+
+ var remoteUri =
remoteBroker.getTransportConnectors().get(0).getPublishableConnectURI();
+ remoteConnection = new
ActiveMQConnectionFactory(remoteUri).createConnection();
+ remoteConnection.start();
+ remoteConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)
+ .createConsumer(new ActiveMQQueue(QUEUE_NAME))
+ .setMessageListener(m -> remoteReceived.incrementAndGet());
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (remoteConnection != null) {
+ try { remoteConnection.close(); } catch (Exception ignored) {}
+ }
+ if (localBroker != null) {
+ localBroker.stop();
+ localBroker.waitUntilStopped();
+ }
+ if (remoteBroker != null) {
+ remoteBroker.stop();
+ remoteBroker.waitUntilStopped();
+ }
+ }
+
+ @Test(timeout = 120_000)
+ public void testBridgeReconnectsAfterLocalConnectionClosedServerSide()
throws Exception {
+ var nc = startBridge();
+
+ produce(5);
+ assertTrue("bridge should forward before the close",
+ Wait.waitFor(() -> remoteReceived.get() >= 5, 20_000, 10));
+
+ // The bridge's local side is a vm:// client of the local broker.
+ var oldBridge = nc.activeBridges().iterator().next();
+ var bridgeLocalConnection = findVmConnection(localBroker);
+ assertNotNull("expected the bridge's local vm:// connection on the
local broker", bridgeLocalConnection);
+ LOG.info("stopping the bridge's local connection server-side: {}",
bridgeLocalConnection);
+ bridgeLocalConnection.serviceException(new IOException("stopping the
bridge's local connection server-side"));
+
+ // a connection stop is not a broker stop
+ var reconnected = Wait.waitFor(() -> {
+ for (var bridge : nc.activeBridges()) {
+ if (bridge != oldBridge) {
+ return true;
+ }
+ }
+ return false;
+ }, 30_000, 10);
+ assertTrue("bridge must reconnect after local-side connection closed
and not a broker shutdown", reconnected);
+ assertTrue("stopped bridge instance should be removed from
activeBridges",
+ Wait.waitFor(() -> !nc.activeBridges().contains(oldBridge),
20_000, 10));
+
+ var before = remoteReceived.get();
+ produce(5);
+ assertTrue("messages produced after the reconnect must flow across the
new bridge "
+ + "(remote had " + before + ", waiting for " + (before
+ 5) + ")",
+ Wait.waitFor(() -> remoteReceived.get() >= before + 5, 20_000,
10));
+ LOG.info("bridge reconnected and resumed forwarding after server-side
local connection stop");
+ nc.stop();
+ }
+
+ /**
+ * A local broker shutdown should stop the bridge cleanly, with no failure
handling or
+ * reconnect attempt.
+ */
+ @Test(timeout = 120_000)
+ public void testBridgeStopsCleanlyOnLocalBrokerShutdown() throws Exception
{
+ var nc = startBridge();
+ produce(5);
+ assertTrue("bridge should forward before the shutdown",
+ Wait.waitFor(() -> remoteReceived.get() >= 5, 20_000, 10));
+
+ localBroker.stop();
+ localBroker.waitUntilStopped();
+
+ assertTrue("bridge should be gone after a local broker shutdown",
+ Wait.waitFor(() -> nc.activeBridges().isEmpty(), 20_000, 10));
+ // the remote broker should be left with only the test's consumer
connection - a
+ // reconnect attempt during shutdown would show up as an extra inbound
client
+ assertTrue("no bridge connection should linger or reconnect on the
remote broker",
+ Wait.waitFor(() -> ((RegionBroker)
remoteBroker.getRegionBroker()).getClients().length == 1, 20_000, 10));
+ Thread.sleep(2000);
+ assertTrue("and none may appear afterwards (no reconnect attempts
after broker stop)",
+ ((RegionBroker)
remoteBroker.getRegionBroker()).getClients().length == 1);
+ }
+
+ /**
+ * Stopping the connector delivers a ShutdownInfo to the bridge's own
local transport as part
+ * of its shutdown. That echo must not be treated as a failure and
reconnect, since the bridge
+ * is already disposed.
+ */
+ @Test(timeout = 120_000)
+ public void testConnectorStopStaysStopped() throws Exception {
+ var nc = startBridge();
+ produce(5);
+ assertTrue("bridge should forward before the stop",
+ Wait.waitFor(() -> remoteReceived.get() >= 5, 20_000, 10));
+
+ nc.stop();
+ assertTrue("bridge should be gone after connector stop",
+ Wait.waitFor(() -> nc.activeBridges().isEmpty(), 20_000, 10));
+
+ var receivedAtStop = remoteReceived.get();
+ produce(5);
+ Thread.sleep(3000);
+ assertTrue("bridge must not reconnect ", nc.activeBridges().isEmpty());
+ assertTrue("no messages move after the stop (remote had " +
receivedAtStop
+ + ", now " + remoteReceived.get() + ")", remoteReceived.get()
== receivedAtStop);
+ }
+
+ /**
+ * Every server-side close should be recovered from, not just the first.
After each close the
+ * bridge reconnects, only one bridge stays registered, and forwarding
still works at the end.
+ */
+ @Test(timeout = 120_000)
+ public void testBridgeReconnectsAfterRepeatedLocalConnectionCloses()
throws Exception {
+ var nc = startBridge();
+ produce(5);
+ assertTrue("bridge should forward before the closes",
+ Wait.waitFor(() -> remoteReceived.get() >= 5, 20_000, 10));
+
+ for (var round = 1; round <= 3; round++) {
+ var oldBridge = nc.activeBridges().iterator().next();
+ // after a reconnect the new bridge's vm connection may not be
registered yet
+ assertTrue("round " + round + ": bridge's local vm:// connection
should be present",
+ Wait.waitFor(() -> findVmConnection(localBroker) != null,
20_000, 10));
+ var bridgeLocalConnection = findVmConnection(localBroker);
+ bridgeLocalConnection.serviceException(new IOException("stopping
the bridge's local connection server-side"));
+
+ final var expectRound = round;
+ assertTrue("round " + expectRound + ": bridge must reconnect with
a new instance",
+ Wait.waitFor(() -> {
+ for (var bridge : nc.activeBridges()) {
+ if (bridge != oldBridge) {
+ return true;
+ }
+ }
+ return false;
+ }, 30_000, 10));
+ assertTrue("round " + expectRound + ": old bridge instance must be
deregistered, only one live bridge",
+ Wait.waitFor(() -> nc.activeBridges().size() == 1
+ && !nc.activeBridges().contains(oldBridge),
20_000, 10));
+ }
+
+ var before = remoteReceived.get();
+ produce(5);
+ assertTrue("forwarding must still work after repeated close/reconnect
cycles",
+ Wait.waitFor(() -> remoteReceived.get() >= before + 5, 20_000,
10));
+ nc.stop();
+ }
+
+ /**
+ * Once the bridge is stopped, a late ShutdownInfo must be a no-op:
serviceLocalException() on
+ * a disposed bridge must not fire bridgeFailed() or schedule a reconnect.
+ */
+ @Test(timeout = 120_000)
+ public void testStoppedBridgeIgnoresShutdownInfo() throws Exception {
+ var nc = startBridge();
+ produce(5);
+ assertTrue("bridge should forward before the stop",
+ Wait.waitFor(() -> remoteReceived.get() >= 5, 20_000, 10));
+
+ var bridge = (DemandForwardingBridgeSupport)
nc.activeBridges().iterator().next();
+ nc.stop();
+ assertTrue("bridge should be gone after connector stop",
+ Wait.waitFor(() -> nc.activeBridges().isEmpty(), 20_000, 10));
+
+ // deliver a late ShutdownInfo to the stopped bridge; the broker is
still running, so
+ // without the disposed guard this would take the failure path and
reconnect
+ bridge.serviceLocalCommand(new ShutdownInfo());
+
+ Thread.sleep(3000);
+ assertTrue("a ShutdownInfo delivered to a stopped bridge must not
resurrect it",
+ nc.activeBridges().isEmpty());
+ assertTrue("and no bridge connection may reappear on the remote
broker",
+ ((RegionBroker)
remoteBroker.getRegionBroker()).getClients().length == 1);
+ }
+
+ /**
+ * AbortSlowConsumerStrategy and AbortSlowAckConsumerStrategy with
abortConnection=true call
+ * connection.serviceException(InactivityIOException) on the bridge's
local connection. The
+ * bridge must restart and resume forwarding.
+ */
+ @Test(timeout = 120_000)
+ public void testBridgeRestartsAfterConnectionServiceException() throws
Exception {
+ var nc = startBridge();
+ produce(5);
+ assertTrue("bridge should forward before the abort",
+ Wait.waitFor(() -> remoteReceived.get() >= 5, 20_000, 10));
+
+ var oldBridge = nc.activeBridges().iterator().next();
+ var bridgeLocalConnection = findVmConnection(localBroker);
+ assertNotNull("expected the bridge's local vm:// connection",
bridgeLocalConnection);
+ LOG.info("servicing an InactivityIOException on the bridge's local
connection: {}", bridgeLocalConnection);
+ bridgeLocalConnection.serviceException(
+ new InactivityIOException("1 Consumers was slow too often or
too long"));
+
+ var restarted = Wait.waitFor(() -> {
+ for (var bridge : nc.activeBridges()) {
+ if (bridge != oldBridge) {
+ return true;
+ }
+ }
+ return false;
+ }, 30_000, 10);
+ assertTrue("bridge must restart after connection.serviceException(..)
on its local connection", restarted);
+ assertTrue("aborted bridge instance must be deregistered from
activeBridges",
+ Wait.waitFor(() -> !nc.activeBridges().contains(oldBridge),
20_000, 10));
+
+ var before = remoteReceived.get();
+ produce(5);
+ assertTrue("messages must flow across the restarted bridge (remote had
" + before + ")",
+ Wait.waitFor(() -> remoteReceived.get() >= before + 5, 20_000,
10));
+ LOG.info("bridge restarted and resumed forwarding after
serviceException abort");
+ nc.stop();
+ }
+
+ private NetworkConnector startBridge() throws Exception {
+ var remoteUri =
remoteBroker.getTransportConnectors().get(0).getPublishableConnectURI();
+ var nc = localBroker.addNetworkConnector("static:(" + remoteUri + ")");
+ nc.setName("to-remote");
+ nc.setDuplex(false);
+ nc.addStaticallyIncludedDestination(new ActiveMQQueue(QUEUE_NAME));
+ nc.start();
+ assertTrue("bridge should establish",
+ Wait.waitFor(() -> !nc.activeBridges().isEmpty(), 20_000, 10));
+ return nc;
+ }
+
+ private void produce(int n) throws Exception {
+ try (var c = new ActiveMQConnectionFactory(
+ localBroker.getTransportConnectors().isEmpty()
+ ? localBroker.getVmConnectorURI()
+ :
localBroker.getTransportConnectors().get(0).getPublishableConnectURI()).createConnection();
+ var session = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ var producer = session.createProducer(new
ActiveMQQueue(QUEUE_NAME))) {
+ c.start();
+
+ for (var i = 0; i < n; i++) {
+ producer.send(session.createTextMessage("m-" + i));
+ }
+ }
+ }
+
+ // The bridge's local side is the only vm:// client connection on the
local broker
+ private TransportConnection findVmConnection(BrokerService broker) throws
Exception {
+ for (var connection :
+ ((RegionBroker) broker.getRegionBroker()).getClients()) {
+ var address = connection.getRemoteAddress();
+ if (address != null && address.startsWith("vm:") && connection
instanceof TransportConnection) {
+ return (TransportConnection) connection;
+ }
+ }
+ return null;
+ }
+}
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkBridgeSlowConsumerAbortReconnectTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkBridgeSlowConsumerAbortReconnectTest.java
new file mode 100644
index 0000000000..c3187d7fbf
--- /dev/null
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkBridgeSlowConsumerAbortReconnectTest.java
@@ -0,0 +1,164 @@
+/**
+ * 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.network;
+
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.jms.Connection;
+import jakarta.jms.Session;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.policy.AbortSlowAckConsumerStrategy;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.test.annotations.ParallelTest;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * AbortSlowAckConsumerStrategy with ignoreNetworkConsumers=false and
abortConnection=true aborts
+ * a network bridge's subscription. The abort closes the bridge's local
connection server-side, so
+ * the bridge should fail, reconnect, and resume forwarding.
+ *
+ * The strategy marks the bridge's subscription slow from timeSinceLastAck,
which starts at
+ * subscription creation, so an idle bridge is aborted after
maxTimeSinceLastAck without simulating
+ * a link fault.
+ */
+@Category(ParallelTest.class)
+public class NetworkBridgeSlowConsumerAbortReconnectTest {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(NetworkBridgeSlowConsumerAbortReconnectTest.class);
+
+ private static final String QUEUE_NAME = "BRIDGE.ABORT.RECONNECT.TEST";
+ private static final long MAX_TIME_SINCE_LAST_ACK = 2000;
+ private static final long MAX_SLOW_DURATION = 2000;
+
+ private BrokerService localBroker;
+ private BrokerService remoteBroker;
+ private Connection remoteConnection;
+ private final AtomicInteger remoteReceived = new AtomicInteger();
+
+ @Before
+ public void setUp() throws Exception {
+ remoteBroker = new BrokerService();
+ remoteBroker.setBrokerName("remote");
+ remoteBroker.setPersistent(false);
+ remoteBroker.setUseJmx(false);
+ remoteBroker.addConnector("tcp://127.0.0.1:0");
+ remoteBroker.start();
+ remoteBroker.waitUntilStarted();
+
+ var strategy = new AbortSlowAckConsumerStrategy();
+ strategy.setIgnoreNetworkConsumers(false); // default true never
touches a bridge
+ strategy.setIgnoreIdleConsumers(false); // reach the bridge even
when it is idle
+ strategy.setMaxTimeSinceLastAck(MAX_TIME_SINCE_LAST_ACK);
+ strategy.setMaxSlowDuration(MAX_SLOW_DURATION);
+ strategy.setCheckPeriod(500);
+ strategy.setAbortConnection(true); // server-side stop of
the bridge's connection
+
+ var entry = new PolicyEntry();
+ entry.setQueue(">");
+ entry.setSlowConsumerStrategy(strategy);
+ var policyMap = new PolicyMap();
+ policyMap.setDefaultEntry(entry);
+
+ localBroker = new BrokerService();
+ localBroker.setBrokerName("local");
+ localBroker.setPersistent(false);
+ localBroker.setUseJmx(false);
+ localBroker.setDestinationPolicy(policyMap);
+ localBroker.start();
+ localBroker.waitUntilStarted();
+
+ var remoteUri =
remoteBroker.getTransportConnectors().get(0).getPublishableConnectURI();
+ remoteConnection = new
ActiveMQConnectionFactory(remoteUri).createConnection();
+ remoteConnection.start();
+ remoteConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)
+ .createConsumer(new ActiveMQQueue(QUEUE_NAME))
+ .setMessageListener(m -> remoteReceived.incrementAndGet());
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (remoteConnection != null) {
+ try { remoteConnection.close(); } catch (Exception ignored) {}
+ }
+ if (localBroker != null) {
+ localBroker.stop();
+ localBroker.waitUntilStopped();
+ }
+ if (remoteBroker != null) {
+ remoteBroker.stop();
+ remoteBroker.waitUntilStopped();
+ }
+ }
+
+ @Test(timeout = 120_000)
+ public void testBridgeReconnectsAfterSlowConsumerStrategyAbort() throws
Exception {
+ var remoteUri =
remoteBroker.getTransportConnectors().get(0).getPublishableConnectURI();
+ var nc = localBroker.addNetworkConnector("static:(" + remoteUri + ")");
+ nc.setName("to-remote");
+ nc.setDuplex(false);
+ nc.addStaticallyIncludedDestination(new ActiveMQQueue(QUEUE_NAME));
+ nc.start();
+ assertTrue("bridge should establish",
+ Wait.waitFor(() -> !nc.activeBridges().isEmpty(), 20_000, 10));
+ var firstBridge = nc.activeBridges().iterator().next();
+
+ // the strategy marks the idle bridge subscription slow after
maxTimeSinceLastAck and
+ // aborts its connection after maxSlowDuration; the bridge should then
come back
+ assertTrue("a new bridge instance must replace the aborted one (abort
expected ~"
+ + (MAX_TIME_SINCE_LAST_ACK + MAX_SLOW_DURATION) + "ms after
establish)",
+ Wait.waitFor(() -> {
+ for (var bridge : nc.activeBridges()) {
+ if (bridge != firstBridge) {
+ return true;
+ }
+ }
+ return false;
+ }, 60_000, 10));
+ assertTrue("the aborted bridge instance must be deregistered from
activeBridges",
+ Wait.waitFor(() -> !nc.activeBridges().contains(firstBridge),
20_000, 10));
+ LOG.info("bridge aborted by AbortSlowAckConsumerStrategy and
reconnected");
+
+ // an idle bridge keeps being recycled, but once messages flow and are
acked the active
+ // bridge is no longer slow, so forwarding converges
+ try (var c = new
ActiveMQConnectionFactory(localBroker.getVmConnectorURI()).createConnection();
+ var session = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ var producer = session.createProducer(new
ActiveMQQueue(QUEUE_NAME));) {
+ c.start();
+
+
+ for (var i = 0; i < 5; i++) {
+ producer.send(session.createTextMessage("m-" + i));
+ }
+ }
+ assertTrue("messages must flow across a live bridge after the strategy
aborts",
+ Wait.waitFor(() -> remoteReceived.get() >= 5, 60_000, 10));
+ LOG.info("forwarding resumed after the strategy abort");
+ nc.stop();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
