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/02/24 15:40:24 UTC
(activemq) branch main updated: Relax timeouts and macOS exclusions + small fixes (#1673)
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 3a88c10ed1 Relax timeouts and macOS exclusions + small fixes (#1673)
3a88c10ed1 is described below
commit 3a88c10ed18d55d8c61ebc2fd8f54a7d462bf75b
Author: Jean-Louis Monteiro <je...@gmail.com>
AuthorDate: Tue Feb 24 16:40:18 2026 +0100
Relax timeouts and macOS exclusions + small fixes (#1673)
---
activemq-unit-tests/pom.xml | 24 +-
.../activemq/bugs/DuplicateFromStoreTest.java | 6 +-
.../network/DurableSyncNetworkBridgeTest.java | 86 +++---
.../network/DynamicNetworkTestSupport.java | 36 +--
...callyIncludedDestinationsDuplexNetworkTest.java | 27 +-
.../network/MQTTNetworkOfBrokersFailoverTest.java | 37 ++-
.../apache/activemq/network/SimpleNetworkTest.java | 291 +++++++++++----------
.../RestrictedThreadPoolInactivityTimeoutTest.java | 2 +-
.../QueueZeroPrefetchLazyDispatchPriorityTest.java | 4 +-
9 files changed, 283 insertions(+), 230 deletions(-)
diff --git a/activemq-unit-tests/pom.xml b/activemq-unit-tests/pom.xml
index 68955145d1..7e1d3c73a5 100644
--- a/activemq-unit-tests/pom.xml
+++ b/activemq-unit-tests/pom.xml
@@ -953,7 +953,6 @@
<exclude>org/apache/activemq/transport/SoWriteTimeoutTest.*</exclude>
<exclude>org/apache/activemq/transport/TopicClusterTest.*</exclude>
<exclude>org/apache/activemq/transport/discovery/DiscoveryNetworkReconnectTest.*</exclude>
- <exclude>org/apache/activemq/transport/discovery/DiscoveryTransportBrokerTest.*</exclude>
<exclude>org/apache/activemq/transport/discovery/DiscoveryTransportNoBrokerTest.*</exclude>
<exclude>org/apache/activemq/transport/discovery/DiscoveryUriTest.*</exclude>
<exclude>org/apache/activemq/transport/discovery/MasterSlaveDiscoveryTest.*</exclude>
@@ -1144,6 +1143,29 @@
</plugins>
</build>
</profile>
+ <profile>
+ <id>activemq.tests.mac.excludes</id>
+ <activation>
+ <os>
+ <family>mac</family>
+ </os>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <excludes combine.children="append">
+ <!-- Multicast and UDP based tests are unreliable on macOS CI (no proper multicast interfaces) -->
+ <exclude>org/apache/activemq/transport/peer/PeerTransportTest.*</exclude>
+ <exclude>org/apache/activemq/transport/multicast/MulticastTransportTest.*</exclude>
+ <exclude>org/apache/activemq/network/MulticastNetworkTest.*</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
<profile>
<id>activemq.tests.aix.excludes</id>
<activation>
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/DuplicateFromStoreTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/DuplicateFromStoreTest.java
index 753e41aa2c..04f633893d 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/DuplicateFromStoreTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/DuplicateFromStoreTest.java
@@ -123,7 +123,7 @@ public class DuplicateFromStoreTest {
}
}
- @Test
+ @Test(timeout = 120_000)
public void testDuplicateMessage() throws Exception {
LOG.info("Testing for duplicate messages.");
@@ -134,10 +134,10 @@ public class DuplicateFromStoreTest {
createOpenwireClients(producers, consumers);
LOG.info("All producers and consumers got started. Awaiting their termination");
- producersFinished.await(100, TimeUnit.MINUTES);
+ producersFinished.await(2, TimeUnit.MINUTES);
LOG.info("All producers have terminated. remaining to send: " + totalMessagesToSend.get() + ", sent:" + totalMessagesSent.get());
- consumersFinished.await(100, TimeUnit.MINUTES);
+ consumersFinished.await(2, TimeUnit.MINUTES);
LOG.info("All consumers have terminated.");
producers.shutdownNow();
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeTest.java
index 9727dbc41a..1791ec242f 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DurableSyncNetworkBridgeTest.java
@@ -49,7 +49,6 @@ import org.apache.activemq.plugin.java.JavaRuntimeConfigurationPlugin;
import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
import org.apache.activemq.store.kahadb.disk.journal.Journal.JournalDiskSyncStrategy;
import org.apache.activemq.util.Wait;
-import org.apache.activemq.util.Wait.Condition;
import org.junit.After;
import org.junit.Assume;
import org.junit.Before;
@@ -83,7 +82,7 @@ public class DurableSyncNetworkBridgeTest extends DynamicNetworkTestSupport {
private final FLOW flow;
@Rule
- public Timeout globalTimeout = new Timeout(30, TimeUnit.SECONDS);
+ public Timeout globalTimeout = new Timeout(60, TimeUnit.SECONDS);
@Parameters
public static Collection<Object[]> data() {
@@ -531,7 +530,6 @@ public class DurableSyncNetworkBridgeTest extends DynamicNetworkTestSupport {
session1.createDurableSubscriber(topic, "sub3");
session1.createDurableSubscriber(excludeTopic, "sub-exclude");
- Thread.sleep(1000);
assertNCDurableSubsCount(broker2, topic, 1);
assertNCDurableSubsCount(broker2, excludeTopic, 0);
@@ -570,13 +568,10 @@ public class DurableSyncNetworkBridgeTest extends DynamicNetworkTestSupport {
secondConnector.start();
//Make sure both bridges are connected
- assertTrue(Wait.waitFor(new Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1 &&
- localBroker.getNetworkConnectors().get(1).activeBridges().size() == 1;
- }
- }, 10000, 500));
+ assertTrue(Wait.waitFor(() ->
+ localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1 &&
+ localBroker.getNetworkConnectors().get(1).activeBridges().size() == 1,
+ TimeUnit.SECONDS.toMillis(10), 500));
//Make sure NC durables exist for both bridges
assertNCDurableSubsCount(broker2, topic2, 1);
@@ -637,13 +632,7 @@ public class DurableSyncNetworkBridgeTest extends DynamicNetworkTestSupport {
final DestinationStatistics remoteDestStatistics2 = remoteBroker.getDestination(
new ActiveMQQueue("include.test.bar.bridge")).getDestinationStatistics();
- assertTrue(Wait.waitFor(new Condition() {
-
- @Override
- public boolean isSatisified() throws Exception {
- return remoteDestStatistics2.getMessages().getCount() == 501;
- }
- }));
+ assertTrue(Wait.waitFor(() -> remoteDestStatistics2.getMessages().getCount() == 501));
}
@@ -723,8 +712,36 @@ public class DurableSyncNetworkBridgeTest extends DynamicNetworkTestSupport {
included = new ActiveMQTopic(testTopicName);
doSetUpRemoteBroker(deleteAllMessages, remoteDataDir, 0);
doSetUpLocalBroker(deleteAllMessages, startNetworkConnector, localDataDir);
- //Give time for advisories to propagate
- Thread.sleep(1000);
+ //Wait for the bridge to be fully started (advisory consumers registered).
+ //Note: activeBridges().size() == 1 is NOT sufficient because bridges are added
+ //to the map before start() completes asynchronously. We must wait for the
+ //startedLatch which counts down after advisory consumers are registered.
+ if (startNetworkConnector) {
+ waitForBridgeFullyStarted();
+ }
+ }
+
+ private void waitForBridgeFullyStarted() throws Exception {
+ // Wait for the local bridge to be fully started (advisory consumers registered)
+ assertTrue("Local bridge should be fully started", Wait.waitFor(() -> {
+ if (localBroker.getNetworkConnectors().get(0).activeBridges().isEmpty()) {
+ return false;
+ }
+ final NetworkBridge bridge = localBroker.getNetworkConnectors().get(0).activeBridges().iterator().next();
+ if (bridge instanceof DemandForwardingBridgeSupport) {
+ return ((DemandForwardingBridgeSupport) bridge).startedLatch.getCount() == 0;
+ }
+ return true;
+ }, TimeUnit.SECONDS.toMillis(10), 100));
+
+ // Also wait for the duplex bridge on the remote broker to be fully started.
+ // The duplex connector creates a separate DemandForwardingBridge on the remote side
+ // that also needs its advisory consumers registered before it can process events.
+ assertTrue("Duplex bridge should be fully started", Wait.waitFor(() -> {
+ final DemandForwardingBridge duplexBridge = findDuplexBridge(
+ remoteBroker.getTransportConnectors().get(0));
+ return duplexBridge != null && duplexBridge.startedLatch.getCount() == 0;
+ }, TimeUnit.SECONDS.toMillis(10), 100));
}
protected void restartLocalBroker(boolean startNetworkConnector) throws Exception {
@@ -757,12 +774,12 @@ public class DurableSyncNetworkBridgeTest extends DynamicNetworkTestSupport {
localConnection.start();
if (startNetworkConnector) {
- Wait.waitFor(new Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1;
- }
- }, 5000, 500);
+ // Best-effort wait for the bridge to appear. Do NOT use assertTrue here
+ // because some tests restart localBroker before remoteBroker is running,
+ // relying on the bridge connecting later when remoteBroker restarts.
+ // Tests that need the bridge to be fully started call assertBridgeStarted() explicitly.
+ Wait.waitFor(() -> localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1,
+ TimeUnit.SECONDS.toMillis(10), 500);
}
localSession = localConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
@@ -881,19 +898,16 @@ public class DurableSyncNetworkBridgeTest extends DynamicNetworkTestSupport {
protected void waitForSubscriptionInactive(final BrokerService brokerService,
final ActiveMQTopic topic,
final String subName) throws Exception {
- assertTrue("Subscription should become inactive", Wait.waitFor(new Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- List<org.apache.activemq.broker.region.DurableTopicSubscription> subs = getSubscriptions(brokerService, topic);
- for (org.apache.activemq.broker.region.DurableTopicSubscription sub : subs) {
- if (sub.getSubscriptionKey().getSubscriptionName().equals(subName)) {
- return !sub.isActive();
- }
+ assertTrue("Subscription should become inactive", Wait.waitFor(() -> {
+ final List<org.apache.activemq.broker.region.DurableTopicSubscription> subs = getSubscriptions(brokerService, topic);
+ for (final org.apache.activemq.broker.region.DurableTopicSubscription sub : subs) {
+ if (sub.getSubscriptionKey().getSubscriptionName().equals(subName)) {
+ return !sub.isActive();
}
- // If subscription doesn't exist, it's considered inactive
- return true;
}
- }, 5000, 100));
+ // If subscription doesn't exist, it's considered inactive
+ return true;
+ }, TimeUnit.SECONDS.toMillis(10), 100));
}
}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicNetworkTestSupport.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicNetworkTestSupport.java
index 87ee09040e..a347ffe3b5 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicNetworkTestSupport.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicNetworkTestSupport.java
@@ -24,6 +24,7 @@ import java.io.File;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import jakarta.jms.Connection;
import jakarta.jms.JMSException;
@@ -45,7 +46,6 @@ import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.command.RemoveSubscriptionInfo;
import org.apache.activemq.util.SubscriptionKey;
import org.apache.activemq.util.Wait;
-import org.apache.activemq.util.Wait.Condition;
import org.junit.Rule;
import org.junit.rules.TemporaryFolder;
@@ -95,9 +95,16 @@ public abstract class DynamicNetworkTestSupport {
}
protected void assertBridgeStarted() throws Exception {
- assertTrue(Wait.waitFor(
- () -> localBroker.getNetworkConnectors().get(0).activeBridges().size() == 1,
- 10000, 500));
+ assertTrue("Bridge should be fully started", Wait.waitFor(() -> {
+ if (localBroker.getNetworkConnectors().get(0).activeBridges().size() != 1) {
+ return false;
+ }
+ final NetworkBridge bridge = localBroker.getNetworkConnectors().get(0).activeBridges().iterator().next();
+ if (bridge instanceof DemandForwardingBridgeSupport) {
+ return ((DemandForwardingBridgeSupport) bridge).startedLatch.getCount() == 0;
+ }
+ return true;
+ }, TimeUnit.SECONDS.toMillis(10), 500));
}
protected RemoveSubscriptionInfo getRemoveSubscriptionInfo(final ConnectionContext context,
@@ -135,17 +142,19 @@ public abstract class DynamicNetworkTestSupport {
protected void assertNCDurableSubsCount(final BrokerService brokerService,
final ActiveMQTopic dest, final int count) throws Exception {
- assertTrue(Wait.waitFor(() -> count == getNCDurableSubs(brokerService, dest).size(),
- 10000, 500));
+ assertTrue("Expected " + count + " NC durable subs on " + dest,
+ Wait.waitFor(() -> count == getNCDurableSubs(brokerService, dest).size(),
+ TimeUnit.SECONDS.toMillis(30), 500));
}
protected void assertConsumersCount(final BrokerService brokerService,
final ActiveMQDestination dest, final int count) throws Exception {
assertTrue(Wait.waitFor(() -> count == getConsumers(brokerService, dest).size(),
10000, 500));
- Thread.sleep(1000);
- // Check one more time after a short pause to make sure the count didn't increase past what we wanted
- assertEquals(count, getConsumers(brokerService, dest).size());
+ // Wait a bit longer and verify the count is stable (didn't increase past what we wanted)
+ assertTrue("Consumer count should remain stable at " + count,
+ Wait.waitFor(() -> count == getConsumers(brokerService, dest).size(),
+ TimeUnit.SECONDS.toMillis(5), 500));
}
protected List<Subscription> getConsumers(final BrokerService brokerService,
@@ -208,12 +217,9 @@ public abstract class DynamicNetworkTestSupport {
protected void assertSubscriptionsCount(final BrokerService brokerService,
final ActiveMQTopic dest, final int count) throws Exception {
- assertTrue(Wait.waitFor(new Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return count == getSubscriptions(brokerService, dest).size();
- }
- }, 10000, 500));
+ assertTrue("Expected " + count + " subscriptions on " + dest,
+ Wait.waitFor(() -> count == getSubscriptions(brokerService, dest).size(),
+ TimeUnit.SECONDS.toMillis(30), 500));
}
protected void assertSubscriptionMapCounts(NetworkBridge networkBridge, final int count) {
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicallyIncludedDestinationsDuplexNetworkTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicallyIncludedDestinationsDuplexNetworkTest.java
index d52836e4f6..229c522559 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicallyIncludedDestinationsDuplexNetworkTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DynamicallyIncludedDestinationsDuplexNetworkTest.java
@@ -80,7 +80,7 @@ public class DynamicallyIncludedDestinationsDuplexNetworkTest extends SimpleNetw
assertEquals("Destination not deleted", 0, remoteBroker.getAdminView().getTemporaryQueues().length);
}
- @Test
+ @Test(timeout = 60 * 1000)
public void testDynamicallyIncludedDestinationsForDuplex() throws Exception{
// Once the bridge is set up, we should see the filter used for the duplex end of the bridge
// only subscribe to the specific destinations included in the <dynamicallyIncludedDestinations> list
@@ -88,11 +88,11 @@ public class DynamicallyIncludedDestinationsDuplexNetworkTest extends SimpleNetw
// is correct
// the bridge on the remote broker has the correct filter
- TransportConnection bridgeConnection = getDuplexBridgeConnectionFromRemote();
+ final TransportConnection bridgeConnection = getDuplexBridgeConnectionFromRemote();
assertNotNull(bridgeConnection);
- DemandForwardingBridge duplexBridge = getDuplexBridgeFromConnection(bridgeConnection);
+ final DemandForwardingBridge duplexBridge = waitForDuplexBridge(bridgeConnection);
assertNotNull(duplexBridge);
- NetworkBridgeConfiguration configuration = getConfigurationFromNetworkBridge(duplexBridge);
+ final NetworkBridgeConfiguration configuration = getConfigurationFromNetworkBridge(duplexBridge);
assertNotNull(configuration);
assertFalse("This destinationFilter does not include ONLY the destinations specified in dynamicallyIncludedDestinations",
configuration.getDestinationFilter().equals(AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX + ">"));
@@ -115,17 +115,20 @@ public class DynamicallyIncludedDestinationsDuplexNetworkTest extends SimpleNetw
return bridge;
}
+ private DemandForwardingBridge waitForDuplexBridge(final TransportConnection bridgeConnection) throws Exception {
+ assertTrue("Timed out waiting for duplex bridge to be fully started",
+ Wait.waitFor(() -> {
+ final DemandForwardingBridge bridge = getDuplexBridgeFromConnection(bridgeConnection);
+ return bridge != null && bridge.getRemoteBrokerName() != null;
+ }));
+ return getDuplexBridgeFromConnection(bridgeConnection);
+ }
+
public TransportConnection getDuplexBridgeConnectionFromRemote() throws Exception {
final TransportConnector transportConnector = remoteBroker.getTransportConnectorByScheme("tcp");
assertTrue("Timed out waiting for duplex bridge connection",
- Wait.waitFor(new Wait.Condition() {
- @Override
- public boolean isSatisified() {
- return !transportConnector.getConnections().isEmpty();
- }
- }));
- CopyOnWriteArrayList<TransportConnection> transportConnections = transportConnector.getConnections();
- return transportConnections.get(0);
+ Wait.waitFor(() -> !transportConnector.getConnections().isEmpty()));
+ return transportConnector.getConnections().get(0);
}
@Override
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/MQTTNetworkOfBrokersFailoverTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/MQTTNetworkOfBrokersFailoverTest.java
index ad52fecf70..bc6c985e24 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/MQTTNetworkOfBrokersFailoverTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/MQTTNetworkOfBrokersFailoverTest.java
@@ -23,9 +23,7 @@ import java.util.concurrent.TimeUnit;
import jakarta.jms.Connection;
import jakarta.jms.Destination;
import jakarta.jms.JMSException;
-import jakarta.jms.Message;
import jakarta.jms.MessageConsumer;
-import jakarta.jms.MessageListener;
import jakarta.jms.Session;
import javax.management.ObjectName;
@@ -35,6 +33,7 @@ import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.DurableSubscriptionViewMBean;
import org.apache.activemq.store.PersistenceAdapter;
+import org.apache.activemq.util.Wait;
import org.apache.commons.lang.ArrayUtils;
import org.fusesource.hawtdispatch.Dispatch;
import org.fusesource.mqtt.client.BlockingConnection;
@@ -69,6 +68,10 @@ public class MQTTNetworkOfBrokersFailoverTest extends NetworkTestSupport {
remoteBroker.addNetworkConnector(nc);
nc.start();
+ // Wait for the network bridge to be established before proceeding
+ assertTrue("Network bridge should be established",
+ Wait.waitFor(() -> nc.activeBridges().size() == 1, TimeUnit.SECONDS.toMillis(10), 500));
+
// mqtt port should have been assigned by now
assertFalse(localBrokerMQTTPort == -1);
assertFalse(remoteBrokerMQTTPort == -1);
@@ -104,7 +107,7 @@ public class MQTTNetworkOfBrokersFailoverTest extends NetworkTestSupport {
remoteConn.connect();
remoteConn.subscribe(new Topic[]{new Topic("foo/bar", QoS.AT_LEAST_ONCE)});
- assertTrue("No destination detected!", consumerNetworked.await(1, TimeUnit.SECONDS));
+ assertTrue("No destination detected!", consumerNetworked.await(5, TimeUnit.SECONDS));
assertQueueExistsOn(remoteBroker, "Consumer.foo_AT_LEAST_ONCE.VirtualTopic.foo.bar");
assertQueueExistsOn(broker, "Consumer.foo_AT_LEAST_ONCE.VirtualTopic.foo.bar");
remoteConn.disconnect();
@@ -149,23 +152,17 @@ public class MQTTNetworkOfBrokersFailoverTest extends NetworkTestSupport {
final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination dest = session.createTopic("ActiveMQ.Advisory.Consumer.Queue.Consumer.foo:AT_LEAST_ONCE.VirtualTopic.foo.bar");
MessageConsumer consumer = session.createConsumer(dest);
- consumer.setMessageListener(new MessageListener() {
- @Override
- public void onMessage(Message message) {
- latch.countDown();
- // shutdown this connection
- Dispatch.getGlobalQueue().execute(new Runnable() {
- @Override
- public void run() {
- try {
- session.close();
- connection.close();
- } catch (JMSException e) {
- e.printStackTrace();
- }
- }
- });
- }
+ consumer.setMessageListener(message -> {
+ latch.countDown();
+ // shutdown this connection
+ Dispatch.getGlobalQueue().execute(() -> {
+ try {
+ session.close();
+ connection.close();
+ } catch (JMSException e) {
+ e.printStackTrace();
+ }
+ });
});
return latch;
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
index 6ccf0137b6..08ec477e8d 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
@@ -16,20 +16,15 @@
*/
package org.apache.activemq.network;
-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 java.util.Arrays;
-import java.util.concurrent.ConcurrentMap;
-
import jakarta.jms.DeliveryMode;
import jakarta.jms.Destination;
import jakarta.jms.JMSException;
import jakarta.jms.Message;
import jakarta.jms.MessageConsumer;
-import jakarta.jms.MessageListener;
import jakarta.jms.MessageProducer;
import jakarta.jms.TextMessage;
import jakarta.jms.TopicRequestor;
@@ -41,13 +36,13 @@ import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
-import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.util.Wait;
-import org.apache.activemq.util.Wait.Condition;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.AbstractApplicationContext;
+import java.util.concurrent.TimeUnit;
+
public class SimpleNetworkTest extends BaseNetworkTest {
protected static final int MESSAGE_COUNT = 10;
@@ -70,21 +65,21 @@ public class SimpleNetworkTest extends BaseNetworkTest {
@Test(timeout = 60 * 1000)
public void testMessageCompression() throws Exception {
- ActiveMQConnection localAmqConnection = (ActiveMQConnection) localConnection;
+ final ActiveMQConnection localAmqConnection = (ActiveMQConnection) localConnection;
localAmqConnection.setUseCompression(true);
- MessageConsumer consumer1 = remoteSession.createConsumer(included);
- MessageProducer producer = localSession.createProducer(included);
+ final MessageConsumer consumer1 = remoteSession.createConsumer(included);
+ final MessageProducer producer = localSession.createProducer(included);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
waitForConsumerRegistration(localBroker, 1, included);
for (int i = 0; i < MESSAGE_COUNT; i++) {
- Message test = localSession.createTextMessage("test-" + i);
+ final Message test = localSession.createTextMessage("test-" + i);
producer.send(test);
- Message msg = consumer1.receive(3000);
+ final Message msg = consumer1.receive(3000);
assertNotNull("not null? message: " + i, msg);
- ActiveMQMessage amqMessage = (ActiveMQMessage) msg;
+ final ActiveMQMessage amqMessage = (ActiveMQMessage) msg;
assertTrue(amqMessage.isCompressed());
}
// ensure no more messages received
@@ -96,30 +91,27 @@ public class SimpleNetworkTest extends BaseNetworkTest {
@Test(timeout = 60 * 1000)
public void testRequestReply() throws Exception {
final MessageProducer remoteProducer = remoteSession.createProducer(null);
- MessageConsumer remoteConsumer = remoteSession.createConsumer(included);
- remoteConsumer.setMessageListener(new MessageListener() {
- @Override
- public void onMessage(Message msg) {
- try {
- TextMessage textMsg = (TextMessage)msg;
- String payload = "REPLY: " + textMsg.getText();
- Destination replyTo;
- replyTo = msg.getJMSReplyTo();
- textMsg.clearBody();
- textMsg.setText(payload);
- remoteProducer.send(replyTo, textMsg);
- } catch (JMSException e) {
- e.printStackTrace();
- }
+ final MessageConsumer remoteConsumer = remoteSession.createConsumer(included);
+ remoteConsumer.setMessageListener(msg -> {
+ try {
+ final TextMessage textMsg = (TextMessage) msg;
+ final String payload = "REPLY: " + textMsg.getText();
+ final Destination replyTo = msg.getJMSReplyTo();
+ textMsg.clearBody();
+ textMsg.setText(payload);
+ remoteProducer.send(replyTo, textMsg);
+ } catch (JMSException e) {
+ e.printStackTrace();
}
});
- TopicRequestor requestor = new TopicRequestor((TopicSession)localSession, included);
- // allow for consumer infos to perculate arround
- Thread.sleep(5000);
+ final TopicRequestor requestor = new TopicRequestor((TopicSession) localSession, included);
+ // Wait for consumer demand to propagate across the network bridge
+ waitForConsumerRegistration(localBroker, 1, included);
+
for (int i = 0; i < MESSAGE_COUNT; i++) {
- TextMessage msg = localSession.createTextMessage("test msg: " + i);
- TextMessage result = (TextMessage)requestor.request(msg);
+ final TextMessage msg = localSession.createTextMessage("test msg: " + i);
+ final TextMessage result = (TextMessage) requestor.request(msg);
assertNotNull(result);
LOG.info(result.getText());
}
@@ -129,13 +121,14 @@ public class SimpleNetworkTest extends BaseNetworkTest {
@Test(timeout = 60 * 1000)
public void testFiltering() throws Exception {
- MessageConsumer includedConsumer = remoteSession.createConsumer(included);
- MessageConsumer excludedConsumer = remoteSession.createConsumer(excluded);
- MessageProducer includedProducer = localSession.createProducer(included);
- MessageProducer excludedProducer = localSession.createProducer(excluded);
- // allow for consumer infos to perculate around
- Thread.sleep(2000);
- Message test = localSession.createTextMessage("test");
+ final MessageConsumer includedConsumer = remoteSession.createConsumer(included);
+ final MessageConsumer excludedConsumer = remoteSession.createConsumer(excluded);
+ final MessageProducer includedProducer = localSession.createProducer(included);
+ final MessageProducer excludedProducer = localSession.createProducer(excluded);
+ // Wait for consumer demand to propagate across the network bridge
+ waitForConsumerRegistration(localBroker, 1, included);
+
+ final Message test = localSession.createTextMessage("test");
includedProducer.send(test);
excludedProducer.send(test);
assertNull(excludedConsumer.receive(1000));
@@ -146,15 +139,15 @@ public class SimpleNetworkTest extends BaseNetworkTest {
@Test(timeout = 60 * 1000)
public void testConduitBridge() throws Exception {
- MessageConsumer consumer1 = remoteSession.createConsumer(included);
- MessageConsumer consumer2 = remoteSession.createConsumer(included);
- MessageProducer producer = localSession.createProducer(included);
+ final MessageConsumer consumer1 = remoteSession.createConsumer(included);
+ final MessageConsumer consumer2 = remoteSession.createConsumer(included);
+ final MessageProducer producer = localSession.createProducer(included);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
waitForConsumerRegistration(localBroker, 2, included);
for (int i = 0; i < MESSAGE_COUNT; i++) {
- Message test = localSession.createTextMessage("test-" + i);
+ final Message test = localSession.createTextMessage("test-" + i);
producer.send(test);
assertNotNull(consumer1.receive(1000));
assertNotNull(consumer2.receive(1000));
@@ -170,27 +163,31 @@ public class SimpleNetworkTest extends BaseNetworkTest {
}
private void waitForConsumerRegistration(final BrokerService brokerService, final int min, final ActiveMQDestination destination) throws Exception {
- assertTrue("Internal bridge consumers registered in time", Wait.waitFor(new Wait.Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- Object[] bridges = brokerService.getNetworkConnectors().get(0).bridges.values().toArray();
- if (bridges.length > 0) {
- LOG.info(brokerService + " bridges " + Arrays.toString(bridges));
- DemandForwardingBridgeSupport demandForwardingBridgeSupport = (DemandForwardingBridgeSupport) bridges[0];
- ConcurrentMap<ConsumerId, DemandSubscription> forwardingBridges = demandForwardingBridgeSupport.getLocalSubscriptionMap();
- LOG.info(brokerService + " bridge " + demandForwardingBridgeSupport + ", localSubs: " + forwardingBridges);
- if (!forwardingBridges.isEmpty()) {
- for (DemandSubscription demandSubscription : forwardingBridges.values()) {
- if (demandSubscription.getLocalInfo().getDestination().equals(destination)) {
- LOG.info(brokerService + " DemandSubscription " + demandSubscription + ", size: " + demandSubscription.size());
- return demandSubscription.size() >= min;
- }
+ // Wait for the bridge demand subscriptions to register the expected number of
+ // remote consumers. With conduit subscriptions, multiple remote consumers map
+ // to a single local subscription, so we check DemandSubscription.size().
+ assertTrue("Bridge demand subscription registered for " + destination, Wait.waitFor(() -> {
+ if (brokerService.getNetworkConnectors().isEmpty()) {
+ return false;
+ }
+ for (final NetworkBridge bridge : brokerService.getNetworkConnectors().get(0).activeBridges()) {
+ if (bridge instanceof DemandForwardingBridgeSupport) {
+ final DemandForwardingBridgeSupport dfBridge = (DemandForwardingBridgeSupport) bridge;
+ for (final DemandSubscription ds : dfBridge.getLocalSubscriptionMap().values()) {
+ if (ds.getLocalInfo().getDestination().equals(destination)) {
+ return ds.size() >= min;
}
}
}
- return false;
}
- }));
+ return false;
+ }, TimeUnit.SECONDS.toMillis(30), 100));
+
+ // Also verify the consumer is actually dispatching on the broker's destination.
+ // The DemandSubscription may exist before the local consumer is fully registered.
+ assertTrue("Consumer dispatching on " + destination, Wait.waitFor(
+ () -> brokerService.getDestination(destination).getDestinationStatistics().getConsumers().getCount() >= 1,
+ TimeUnit.SECONDS.toMillis(10), 100));
}
//Added for AMQ-6465 to make sure memory usage decreased back to 0 after messages are forwarded
@@ -198,27 +195,28 @@ public class SimpleNetworkTest extends BaseNetworkTest {
@Test(timeout = 60 * 1000)
public void testDurableTopicSubForwardMemoryUsage() throws Exception {
// create a remote durable consumer to create demand
- MessageConsumer remoteConsumer = remoteSession.createDurableSubscriber(included, consumerName);
- Thread.sleep(1000);
+ final MessageConsumer remoteConsumer = remoteSession.createDurableSubscriber(included, consumerName);
+ // Wait for consumer demand to propagate across the network bridge
+ waitForConsumerRegistration(localBroker, 1, included);
- MessageProducer producer = localSession.createProducer(included);
+ final MessageProducer producer = localSession.createProducer(included);
for (int i = 0; i < MESSAGE_COUNT; i++) {
- Message test = localSession.createTextMessage("test-" + i);
+ final Message test = localSession.createTextMessage("test-" + i);
producer.send(test);
}
- Thread.sleep(1000);
-
- //Make sure stats are set
- assertEquals(MESSAGE_COUNT,
- localBroker.getDestination(included).getDestinationStatistics().getForwards().getCount());
-
- assertTrue(Wait.waitFor(new Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return localBroker.getSystemUsage().getMemoryUsage().getUsage() == 0;
- }
- }, 10000, 500));
+ //Make sure stats are set - wait for forwards to complete
+ assertTrue("Forwards count did not reach expected value", Wait.waitFor(
+ () -> {
+ final long count = localBroker.getDestination(included).getDestinationStatistics().getForwards().getCount();
+ LOG.info("testDurableTopicSubForwardMemoryUsage: forwards count = " + count + " (expected " + MESSAGE_COUNT + ")");
+ return count == MESSAGE_COUNT;
+ },
+ TimeUnit.SECONDS.toMillis(30), 100));
+
+ assertTrue("Memory usage did not return to 0", Wait.waitFor(
+ () -> localBroker.getSystemUsage().getMemoryUsage().getUsage() == 0,
+ TimeUnit.SECONDS.toMillis(10), 500));
remoteConsumer.close();
}
@@ -226,28 +224,25 @@ public class SimpleNetworkTest extends BaseNetworkTest {
//to the other broker
@Test(timeout = 60 * 1000)
public void testTopicSubForwardMemoryUsage() throws Exception {
- // create a remote durable consumer to create demand
- MessageConsumer remoteConsumer = remoteSession.createConsumer(included);
- Thread.sleep(1000);
+ // create a remote consumer to create demand
+ final MessageConsumer remoteConsumer = remoteSession.createConsumer(included);
+ // Wait for consumer demand to propagate across the network bridge
+ waitForConsumerRegistration(localBroker, 1, included);
- MessageProducer producer = localSession.createProducer(included);
+ final MessageProducer producer = localSession.createProducer(included);
for (int i = 0; i < MESSAGE_COUNT; i++) {
- Message test = localSession.createTextMessage("test-" + i);
+ final Message test = localSession.createTextMessage("test-" + i);
producer.send(test);
}
- Thread.sleep(1000);
-
- //Make sure stats are set
- assertEquals(MESSAGE_COUNT,
- localBroker.getDestination(included).getDestinationStatistics().getForwards().getCount());
- assertTrue(Wait.waitFor(new Condition() {
+ //Make sure stats are set - wait for forwards to complete
+ assertTrue("Forwards count did not reach expected value", Wait.waitFor(
+ () -> localBroker.getDestination(included).getDestinationStatistics().getForwards().getCount() == MESSAGE_COUNT,
+ TimeUnit.SECONDS.toMillis(30), 100));
- @Override
- public boolean isSatisified() throws Exception {
- return localBroker.getSystemUsage().getMemoryUsage().getUsage() == 0;
- }
- }, 10000, 500));
+ assertTrue("Memory usage did not return to 0", Wait.waitFor(
+ () -> localBroker.getSystemUsage().getMemoryUsage().getUsage() == 0,
+ TimeUnit.SECONDS.toMillis(10), 500));
for (int i = 0; i < MESSAGE_COUNT; i++) {
assertNotNull("message count: " + i, remoteConsumer.receive(2500));
@@ -259,28 +254,25 @@ public class SimpleNetworkTest extends BaseNetworkTest {
//to the other broker
@Test(timeout = 60 * 1000)
public void testQueueSubForwardMemoryUsage() throws Exception {
- ActiveMQQueue queue = new ActiveMQQueue("include.test.foo");
- MessageConsumer remoteConsumer = remoteSession.createConsumer(queue);
- Thread.sleep(1000);
+ final ActiveMQQueue queue = new ActiveMQQueue("include.test.foo");
+ final MessageConsumer remoteConsumer = remoteSession.createConsumer(queue);
+ // Wait for consumer demand to propagate across the network bridge
+ waitForConsumerRegistration(localBroker, 1, queue);
- MessageProducer producer = localSession.createProducer(queue);
+ final MessageProducer producer = localSession.createProducer(queue);
for (int i = 0; i < MESSAGE_COUNT; i++) {
- Message test = localSession.createTextMessage("test-" + i);
+ final Message test = localSession.createTextMessage("test-" + i);
producer.send(test);
}
- Thread.sleep(1000);
-
- //Make sure stats are set
- assertEquals(MESSAGE_COUNT,
- localBroker.getDestination(queue).getDestinationStatistics().getForwards().getCount());
- assertTrue(Wait.waitFor(new Condition() {
+ //Make sure stats are set - wait for forwards to complete
+ assertTrue("Forwards count did not reach expected value", Wait.waitFor(
+ () -> localBroker.getDestination(queue).getDestinationStatistics().getForwards().getCount() == MESSAGE_COUNT,
+ TimeUnit.SECONDS.toMillis(30), 100));
- @Override
- public boolean isSatisified() throws Exception {
- return localBroker.getSystemUsage().getMemoryUsage().getUsage() == 0;
- }
- }, 10000, 500));
+ assertTrue("Memory usage did not return to 0", Wait.waitFor(
+ () -> localBroker.getSystemUsage().getMemoryUsage().getUsage() == 0,
+ TimeUnit.SECONDS.toMillis(10), 500));
for (int i = 0; i < MESSAGE_COUNT; i++) {
assertNotNull("message count: " + i, remoteConsumer.receive(2500));
@@ -291,28 +283,30 @@ public class SimpleNetworkTest extends BaseNetworkTest {
@Test(timeout = 60 * 1000)
public void testDurableStoreAndForward() throws Exception {
// create a remote durable consumer
- MessageConsumer remoteConsumer = remoteSession.createDurableSubscriber(included, consumerName);
- Thread.sleep(1000);
+ final MessageConsumer remoteConsumer = remoteSession.createDurableSubscriber(included, consumerName);
+ // Wait for consumer demand to propagate across the network bridge
+ waitForConsumerRegistration(localBroker, 1, included);
+
// now close everything down and restart
doTearDown();
doSetUp(false);
- MessageProducer producer = localSession.createProducer(included);
+ final MessageProducer producer = localSession.createProducer(included);
for (int i = 0; i < MESSAGE_COUNT; i++) {
- Message test = localSession.createTextMessage("test-" + i);
+ final Message test = localSession.createTextMessage("test-" + i);
producer.send(test);
}
- Thread.sleep(1000);
- //Make sure stats are set
- assertEquals(MESSAGE_COUNT,
- localBroker.getDestination(included).getDestinationStatistics().getForwards().getCount());
+ //Make sure stats are set - wait for forwards to complete
+ assertTrue("Forwards count did not reach expected value", Wait.waitFor(
+ () -> localBroker.getDestination(included).getDestinationStatistics().getForwards().getCount() == MESSAGE_COUNT,
+ TimeUnit.SECONDS.toMillis(30), 100));
// close everything down and restart
doTearDown();
doSetUp(false);
- remoteConsumer = remoteSession.createDurableSubscriber(included, consumerName);
+ final MessageConsumer remoteConsumer2 = remoteSession.createDurableSubscriber(included, consumerName);
for (int i = 0; i < MESSAGE_COUNT; i++) {
- assertNotNull("message count: " + i, remoteConsumer.receive(2500));
+ assertNotNull("message count: " + i, remoteConsumer2.receive(2500));
}
}
@@ -320,35 +314,55 @@ public class SimpleNetworkTest extends BaseNetworkTest {
"it requires a connection per durable to match that connectionId")
public void testDurableStoreAndForwardReconnect() throws Exception {
// create a local durable consumer
- MessageConsumer localConsumer = localSession.createDurableSubscriber(included, consumerName);
- Thread.sleep(5000);
+ final MessageConsumer localConsumer = localSession.createDurableSubscriber(included, consumerName);
+ // Wait for consumer demand to propagate across the network bridge
+ waitForConsumerRegistration(localBroker, 1, included);
+
// now close everything down and restart
doTearDown();
doSetUp(false);
// send messages
- MessageProducer producer = localSession.createProducer(included);
+ final MessageProducer producer = localSession.createProducer(included);
for (int i = 0; i < MESSAGE_COUNT; i++) {
- Message test = localSession.createTextMessage("test-" + i);
+ final Message test = localSession.createTextMessage("test-" + i);
producer.send(test);
}
- Thread.sleep(5000);
+
+ //Make sure stats are set - wait for forwards to complete
+ assertTrue("Forwards count did not reach expected value", Wait.waitFor(
+ () -> localBroker.getDestination(included).getDestinationStatistics().getForwards().getCount() == MESSAGE_COUNT,
+ TimeUnit.SECONDS.toMillis(30), 100));
+
// consume some messages locally
- localConsumer = localSession.createDurableSubscriber(included, consumerName);
- LOG.info("Consume from local consumer: " + localConsumer);
+ final MessageConsumer localConsumer2 = localSession.createDurableSubscriber(included, consumerName);
+ LOG.info("Consume from local consumer: " + localConsumer2);
for (int i = 0; i < MESSAGE_COUNT / 2; i++) {
- assertNotNull("message count: " + i, localConsumer.receive(2500));
+ assertNotNull("message count: " + i, localConsumer2.receive(2500));
}
- Thread.sleep(5000);
+
+ // Wait for local messages to be consumed
+ assertTrue("Local messages consumed", Wait.waitFor(
+ () -> localBroker.getDestination(included).getDestinationStatistics().getDequeues().getCount() >= MESSAGE_COUNT / 2,
+ TimeUnit.SECONDS.toMillis(10), 100));
+
// close everything down and restart
doTearDown();
doSetUp(false);
- Thread.sleep(5000);
+
+ // Wait for network bridge to re-form
+ assertTrue("Network bridge re-formed", Wait.waitFor(
+ () -> !localBroker.getNetworkConnectors().isEmpty()
+ && !localBroker.getNetworkConnectors().get(0).activeBridges().isEmpty(),
+ TimeUnit.SECONDS.toMillis(10), 100));
LOG.info("Consume from remote");
// consume the rest remotely
- MessageConsumer remoteConsumer = remoteSession.createDurableSubscriber(included, consumerName);
+ final MessageConsumer remoteConsumer = remoteSession.createDurableSubscriber(included, consumerName);
LOG.info("Remote consumer: " + remoteConsumer);
- Thread.sleep(5000);
+
+ // Wait for consumer demand to propagate
+ waitForConsumerRegistration(localBroker, 1, included);
+
for (int i = 0; i < MESSAGE_COUNT / 2; i++) {
assertNotNull("message count: " + i, remoteConsumer.receive(10000));
}
@@ -359,14 +373,11 @@ public class SimpleNetworkTest extends BaseNetworkTest {
final NetworkBridge localBridge = localBroker.getNetworkConnectors().get(0).activeBridges().iterator().next();
final NetworkBridge remoteBridge = remoteBroker.getNetworkConnectors().get(0).activeBridges().iterator().next();
- assertTrue(Wait.waitFor(new Wait.Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return expectedLocalSent == localBridge.getNetworkBridgeStatistics().getDequeues().getCount() &&
- 0 == localBridge.getNetworkBridgeStatistics().getReceivedCount().getCount() &&
- expectedRemoteSent == remoteBridge.getNetworkBridgeStatistics().getDequeues().getCount() &&
- 0 == remoteBridge.getNetworkBridgeStatistics().getReceivedCount().getCount();
- }
- }));
+ assertTrue(Wait.waitFor(() ->
+ expectedLocalSent == localBridge.getNetworkBridgeStatistics().getDequeues().getCount() &&
+ 0 == localBridge.getNetworkBridgeStatistics().getReceivedCount().getCount() &&
+ expectedRemoteSent == remoteBridge.getNetworkBridgeStatistics().getDequeues().getCount() &&
+ 0 == remoteBridge.getNetworkBridgeStatistics().getReceivedCount().getCount()
+ , TimeUnit.SECONDS.toMillis(30), 100));
}
}
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/RestrictedThreadPoolInactivityTimeoutTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/RestrictedThreadPoolInactivityTimeoutTest.java
index d5e2a1a842..7eaefd23bd 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/RestrictedThreadPoolInactivityTimeoutTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/RestrictedThreadPoolInactivityTimeoutTest.java
@@ -124,7 +124,7 @@ public class RestrictedThreadPoolInactivityTimeoutTest extends JmsTestSupport {
assertTrue("Should be at most inactivity monitor pool size * 2. Diff = " + diff, diff <= 2*poolSize);
- assertTrue("all work complete", doneConsumers.await(10, TimeUnit.SECONDS));
+ assertTrue("all work complete", doneConsumers.await(30, TimeUnit.SECONDS));
}
@Override
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueZeroPrefetchLazyDispatchPriorityTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueZeroPrefetchLazyDispatchPriorityTest.java
index dd4dd8e356..4c5b193741 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueZeroPrefetchLazyDispatchPriorityTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueZeroPrefetchLazyDispatchPriorityTest.java
@@ -251,7 +251,7 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest {
return queue != null
&& queue.getDestinationStatistics().getMessages().getCount() == numToSend
&& queue.getDestinationStatistics().getInflight().getCount() == 0;
- }, 5000, 100));
+ }, TimeUnit.SECONDS.toMillis(20), 100));
// Use the retry-based consume to handle zero-prefetch dispatch timing:
// with prioritized messages + lazy dispatch + redelivered messages in the
@@ -297,7 +297,7 @@ public class QueueZeroPrefetchLazyDispatchPriorityTest {
boolean finished = false;
while (!finished) {
- Message message = consumer.receive(returnedMessages.isEmpty() ? 5000 : 1000);
+ Message message = consumer.receive(returnedMessages.isEmpty() ? TimeUnit.SECONDS.toMillis(20) : 1000);
if (message == null) {
finished = true;
}
---------------------------------------------------------------------
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