From e91d90af260fca6741066d3bd13946680cbaaffb Mon Sep 17 00:00:00 2001 From: Matt Pavlovich Date: Thu, 6 Aug 2026 14:58:34 -0500 Subject: [PATCH 1/3] [#2238] Disable periodic expiry task in ActiveDurableSubscriptionBrowseExpireTest The 30s expiry task raced the JMX browse and could double-count the expired statistic. With the task disabled, browse is the only expiry trigger and the count is exact. --- ...tiveDurableSubscriptionBrowseExpireTest.java | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/ActiveDurableSubscriptionBrowseExpireTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/ActiveDurableSubscriptionBrowseExpireTest.java index 53c1f0d8f36..1a1a598243f 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/ActiveDurableSubscriptionBrowseExpireTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/ActiveDurableSubscriptionBrowseExpireTest.java @@ -18,7 +18,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; import java.io.IOException; import java.util.Arrays; @@ -36,12 +35,13 @@ import org.apache.activemq.TestSupport.PersistenceAdapterChoice; import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.jmx.DurableSubscriptionViewMBean; +import org.apache.activemq.broker.region.policy.PolicyEntry; +import org.apache.activemq.broker.region.policy.PolicyMap; import org.apache.activemq.store.MessageRecoveryListener; import org.apache.activemq.store.PersistenceAdapter; import org.apache.activemq.store.TopicMessageStore; import org.apache.activemq.broker.region.Destination; import org.apache.activemq.command.MessageId; -import org.apache.activemq.util.Wait; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; @@ -71,6 +71,17 @@ public PersistenceAdapter setDefaultPersistenceAdapter(BrokerService broker) thr return super.setPersistenceAdapter(broker, PersistenceAdapterChoice.MEM); } + @Override + public void configurePlugins(BrokerService broker) throws Exception { + // Disable the periodic expiry task so the JMX browse is the only expiry + // trigger and the expired count is deterministic + PolicyEntry policy = new PolicyEntry(); + policy.setExpireMessagesPeriod(0); + PolicyMap policyMap = new PolicyMap(); + policyMap.setDefaultEntry(policy); + broker.setDestinationPolicy(policyMap); + } + @Override protected ActiveMQConnectionFactory createConnectionFactory() throws Exception { ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://" + getName(true)); @@ -148,7 +159,7 @@ public boolean isDuplicate(MessageId id) { assertNotNull(data); if (enableExpiration) { - assertTrue(Wait.waitFor(() -> dest.getDestinationStatistics().getExpired().getCount() == messagesToExpire.size(), 5_000, 100)); + assertEquals(messagesToExpire.size(), dest.getDestinationStatistics().getExpired().getCount()); } else { assertEquals(0L, dest.getDestinationStatistics().getExpired().getCount()); } From f4e1cd1f879be36bbb824435537b9aff0d7cd152 Mon Sep 17 00:00:00 2001 From: Matt Pavlovich Date: Thu, 6 Aug 2026 14:58:37 -0500 Subject: [PATCH 2/3] [#2238] Wait for async consumer add/remove in TwoBrokerVirtualTopicSelectorAwareForwardingTest Consumer close sends RemoveInfo async, so assert the region consumer count with Wait.waitFor instead of reading it immediately. --- .../TwoBrokerVirtualTopicSelectorAwareForwardingTest.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TwoBrokerVirtualTopicSelectorAwareForwardingTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TwoBrokerVirtualTopicSelectorAwareForwardingTest.java index 7f31655f9cc..940ca92ddbc 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TwoBrokerVirtualTopicSelectorAwareForwardingTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TwoBrokerVirtualTopicSelectorAwareForwardingTest.java @@ -478,14 +478,17 @@ public void testSelectorAwareForwarding() throws Exception { brokerA.waitUntilStopped(); deleteSelectorCacheFile("BrokerA"); - assertEquals(0, destination.getConsumers().size()); + // consumer close is async - wait for the broker to remove the subscription + assertTrue("consumer should be removed from BrokerB", + Wait.waitFor(() -> destination.getConsumers().isEmpty(), 5000, 10)); remoteConsumer = createConsumer("BrokerB", createDestination("Consumer.B.VirtualTopic.tempTopic", false), "ceposta = 'redhat'"); - assertEquals(1, destination.getConsumers().size()); + assertTrue("new consumer should be registered on BrokerB", + Wait.waitFor(() -> destination.getConsumers().size() == 1, 5000, 10)); // now let's start broker A back up From 50331f34c5664ff2b2fad4641ae07c7392a37e6f Mon Sep 17 00:00:00 2001 From: Matt Pavlovich Date: Thu, 6 Aug 2026 14:58:37 -0500 Subject: [PATCH 3/3] [#2238] Use randomize=false for FailoverClusterTest clients The failover transport shuffles the broker-supplied round-robin order, so each client kept its current broker ~50% of the time and the rebalance advisory only fires once. randomize=false makes the client spread deterministic. --- .../activemq/transport/failover/FailoverClusterTest.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverClusterTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverClusterTest.java index d60cd20de89..3445a22d790 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverClusterTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverClusterTest.java @@ -103,7 +103,9 @@ public void testClusterConnectedBeforeClients() throws Exception { protected void setUp() throws Exception { if (brokerA == null) { brokerA = createBrokerA(getBindAddress() + "?transport.closeAsync=false"); - clientUrl = "failover://(" + brokerA.getTransportConnectors().get(0).getPublishableConnectString() + ")"; + // randomize=false so clients honor the broker-provided round-robin order on + // rebalance, making the client spread across brokers deterministic + clientUrl = "failover://(" + brokerA.getTransportConnectors().get(0).getPublishableConnectString() + ")?randomize=false"; } } @@ -182,6 +184,6 @@ private static boolean waitForClientRebalance(final List con } } return set.size() >= minBrokerCount; - }, TimeUnit.SECONDS.toMillis(15), TimeUnit.MILLISECONDS.toMillis(500)); + }, TimeUnit.SECONDS.toMillis(15), 10l); } }