You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by jb...@apache.org on 2026/10/17 19:01:58 UTC
(activemq) branch main updated: Dump network state when VirtualConsumerDemandTest does not receive a forwarded message (#2657)
This is an automated email from the ASF dual-hosted git repository.
jbonofre pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/main by this push:
new d1a3c3db28 Dump network state when VirtualConsumerDemandTest does not receive a forwarded message (#2657)
d1a3c3db28 is described below
commit d1a3c3db28616d3c01f4b42fbaedfcfa38f2af26
Author: JB Onofré <jb...@apache.org>
AuthorDate: Sat Oct 17 20:01:52 2026 +0100
Dump network state when VirtualConsumerDemandTest does not receive a forwarded message (#2657)
VirtualConsumerDemandTest.testToTopicWithDurable and testToTopic occasionally
fail on CI at bridgeConsumer.receive(5000) (see #2238) and cannot be reproduced
locally. On a receive timeout, dump the include.* destinations, their
subscriptions and statistics on both brokers, plus the bridge demand
subscriptions, so the next flaky failure shows where the message was lost.
Diagnostics only, no change to the test behaviour.
---
.../network/VirtualConsumerDemandTest.java | 88 ++++++++++++++++++++--
1 file changed, 80 insertions(+), 8 deletions(-)
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/VirtualConsumerDemandTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/VirtualConsumerDemandTest.java
index 9c8a4d474b..034164f96c 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/VirtualConsumerDemandTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/VirtualConsumerDemandTest.java
@@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.lang.reflect.Array;
import java.lang.reflect.Field;
@@ -39,8 +40,10 @@ import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.broker.BrokerPlugin;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.broker.region.Destination;
import org.apache.activemq.broker.region.DestinationInterceptor;
import org.apache.activemq.broker.region.DestinationStatistics;
+import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.virtual.CompositeQueue;
import org.apache.activemq.broker.region.virtual.CompositeTopic;
import org.apache.activemq.broker.region.virtual.VirtualDestination;
@@ -185,7 +188,7 @@ public class VirtualConsumerDemandTest extends DynamicNetworkTestSupport {
waitForConsumerCount(destinationStatistics, 1);
includedProducer.send(test);
- assertNotNull(bridgeConsumer.receive(5000));
+ receiveOrDumpState(bridgeConsumer, 5000);
waitForDispatchFromLocalBroker(destinationStatistics, 1);
assertLocalBrokerStatistics(destinationStatistics, 1);
@@ -222,7 +225,7 @@ public class VirtualConsumerDemandTest extends DynamicNetworkTestSupport {
waitForConsumerCount(destinationStatistics, 1);
includedProducer.send(test);
- assertNotNull(bridgeConsumer.receive(5000));
+ receiveOrDumpState(bridgeConsumer, 5000);
//assert a message was forwarded
waitForDispatchFromLocalBroker(destinationStatistics, 1);
@@ -825,7 +828,7 @@ public class VirtualConsumerDemandTest extends DynamicNetworkTestSupport {
waitForConsumerCount(destinationStatistics, 1);
includedProducer.send(test);
- assertNotNull(bridgeConsumer.receive(5000));
+ receiveOrDumpState(bridgeConsumer, 5000);
waitForDispatchFromLocalBroker(destinationStatistics, 1);
@@ -887,7 +890,7 @@ public class VirtualConsumerDemandTest extends DynamicNetworkTestSupport {
//make sure the message is still forwarded
waitForDispatchFromLocalBroker(destinationStatistics, 2);
assertLocalBrokerStatistics(destinationStatistics, 2);
- assertNotNull(bridgeConsumer.receive(5000));
+ receiveOrDumpState(bridgeConsumer, 5000);
assertRemoteAdvisoryCount(advisoryConsumer, 4, 3);
if (isUseVirtualDestSubsOnCreation) {
@@ -1027,7 +1030,7 @@ public class VirtualConsumerDemandTest extends DynamicNetworkTestSupport {
waitForConsumerCount(destinationStatistics, 1);
includedProducer.send(test);
- assertNotNull(bridgeConsumer.receive(5000));
+ receiveOrDumpState(bridgeConsumer, 5000);
final DestinationStatistics remoteStats = remoteBroker.getDestination(new ActiveMQQueue(testQueueName)).getDestinationStatistics();
@@ -1179,7 +1182,7 @@ public class VirtualConsumerDemandTest extends DynamicNetworkTestSupport {
DestinationStatistics localDestinationStats = localBroker.getDestination(included).getDestinationStatistics();
waitForConsumerCount(localDestinationStats, 1);
includedProducer.send(test);
- assertNotNull(bridgeConsumer.receive(5000));
+ receiveOrDumpState(bridgeConsumer, 5000);
//with isUseVirtualDestSubsOnCreation is true, there should be 2 advisories
//with !isUseVirtualDestSubsOnCreation, there should be 1 advisory
@@ -1278,7 +1281,7 @@ public class VirtualConsumerDemandTest extends DynamicNetworkTestSupport {
waitForConsumerCount(destinationStatistics, 1);
includedProducer.send(test);
- assertNotNull(bridgeConsumer.receive(5000));
+ receiveOrDumpState(bridgeConsumer, 5000);
assertRemoteAdvisoryCount(advisoryConsumer, 1);
assertAdvisoryBrokerCounts(1,1,0);
@@ -1355,7 +1358,7 @@ public class VirtualConsumerDemandTest extends DynamicNetworkTestSupport {
waitForConsumerCount(destinationStatistics, 1);
includedProducer.send(test);
- assertNotNull(bridgeConsumer.receive(5000));
+ receiveOrDumpState(bridgeConsumer, 5000);
waitForDispatchFromLocalBroker(destinationStatistics, 1);
assertLocalBrokerStatistics(destinationStatistics, 1);
@@ -1566,6 +1569,75 @@ public class VirtualConsumerDemandTest extends DynamicNetworkTestSupport {
new ActiveMQQueue(queue)));
}
+ /**
+ * Receive a message that is expected to have been forwarded across the network bridge.
+ * This test is occasionally flaky on CI (the message never reaches the remote consumer),
+ * so on failure dump the state of both brokers and of the bridge to find where the
+ * message got lost.
+ */
+ protected void receiveOrDumpState(final MessageConsumer consumer, final long timeout) throws JMSException {
+ if (consumer.receive(timeout) == null) {
+ final String state = dumpNetworkState();
+ LOG.error("No message received within {} ms, network state:\n{}", timeout, state);
+ fail("No message received within " + timeout + " ms, network state:\n" + state);
+ }
+ }
+
+ private String dumpNetworkState() {
+ final StringBuilder sb = new StringBuilder();
+ dumpBrokerState(sb, localBroker);
+ dumpBrokerState(sb, remoteBroker);
+ try {
+ for (NetworkBridge bridge : connector.activeBridges()) {
+ sb.append("bridge ").append(bridge.getRemoteBrokerName())
+ .append(" enqueues=").append(bridge.getNetworkBridgeStatistics().getEnqueues().getCount())
+ .append(" dequeues=").append(bridge.getNetworkBridgeStatistics().getDequeues().getCount());
+ if (bridge instanceof DemandForwardingBridgeSupport) {
+ sb.append(" demandSubscriptions=");
+ ((DemandForwardingBridgeSupport) bridge).getLocalSubscriptionMap().values().forEach(sub ->
+ sb.append("\n [local=").append(sub.getLocalInfo().getDestination())
+ .append(" remote=").append(sub.getRemoteInfo().getDestination())
+ .append(" durable=").append(sub.getRemoteInfo().isDurable()).append("]"));
+ }
+ sb.append('\n');
+ }
+ } catch (Exception e) {
+ sb.append("failed to dump bridges: ").append(e).append('\n');
+ }
+ return sb.toString();
+ }
+
+ private void dumpBrokerState(final StringBuilder sb, final BrokerService broker) {
+ try {
+ sb.append("broker ").append(broker.getBrokerName()).append(":\n");
+ // only look at destinations that already exist, do not create any
+ for (Destination dest : broker.getBroker().getDestinationMap().values()) {
+ final String name = dest.getName();
+ if (name.startsWith("ActiveMQ.Advisory") || !name.startsWith("include.")) {
+ continue;
+ }
+ final DestinationStatistics stats = dest.getDestinationStatistics();
+ sb.append(" ").append(dest.getActiveMQDestination())
+ .append(" consumers=").append(stats.getConsumers().getCount())
+ .append(" enqueues=").append(stats.getEnqueues().getCount())
+ .append(" dispatched=").append(stats.getDispatched().getCount())
+ .append(" dequeues=").append(stats.getDequeues().getCount())
+ .append(" forwards=").append(stats.getForwards().getCount())
+ .append(" messages=").append(stats.getMessages().getCount()).append('\n');
+ for (Subscription sub : dest.getConsumers()) {
+ sb.append(" sub ").append(sub.getConsumerInfo().getConsumerId())
+ .append(" ").append(sub.getConsumerInfo().getSubscriptionName())
+ .append(" network=").append(sub.getConsumerInfo().isNetworkSubscription())
+ .append(" active=").append(sub.isSlowConsumer() ? "slow" : "ok")
+ .append(" dispatched=").append(sub.getDispatchedCounter())
+ .append(" pending=").append(sub.getPendingQueueSize()).append('\n');
+ }
+ }
+ } catch (Exception e) {
+ sb.append("failed to dump broker: ").append(e).append('\n');
+ }
+ }
+
protected void assertRemoteAdvisoryCount(final MessageConsumer advisoryConsumer, final int count) throws JMSException {
int available = 0;
ActiveMQMessage message = null;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact