From 42b911c2437d8fcd625371c3759286a3f55bc98e Mon Sep 17 00:00:00 2001 From: Matt Pavlovich Date: Tue, 4 Aug 2026 10:18:40 -0500 Subject: [PATCH 1/3] Add test proving advisory dispatch counter races under concurrency demandConsumerDispatched is a plain int incremented and reset from multiple transport/executor threads in ackAdvisory. The unsynchronized read-modify-write and unguarded reset let concurrent threads claim overlapping counter ranges in their advisory acks (observed: 800k dispatches acked as 1.16M) or lose counted dispatches outright, corrupting the remote advisory prefetch window. Widens ackAdvisory to package-private for the test. --- .../DemandForwardingBridgeSupport.java | 2 +- .../network/AckAdvisoryConcurrencyTest.java | 196 ++++++++++++++++++ 2 files changed, 197 insertions(+), 1 deletion(-) create mode 100644 activemq-broker/src/test/java/org/apache/activemq/network/AckAdvisoryConcurrencyTest.java 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 c8bb1386c5f..775590a3cf4 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 @@ -940,7 +940,7 @@ public void onCompletion(FutureResponse resp) { } } - private void ackAdvisory(Message message) throws IOException { + void ackAdvisory(Message message) throws IOException { demandConsumerDispatched++; if (demandConsumerDispatched > (demandConsumerInfo.getPrefetchSize() * (configuration.getAdvisoryAckPercentage() / 100f))) { diff --git a/activemq-broker/src/test/java/org/apache/activemq/network/AckAdvisoryConcurrencyTest.java b/activemq-broker/src/test/java/org/apache/activemq/network/AckAdvisoryConcurrencyTest.java new file mode 100644 index 00000000000..d4385095bc8 --- /dev/null +++ b/activemq-broker/src/test/java/org/apache/activemq/network/AckAdvisoryConcurrencyTest.java @@ -0,0 +1,196 @@ +/** + * 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.assertEquals; + +import java.security.cert.X509Certificate; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ActiveMQMessage; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.command.ConnectionId; +import org.apache.activemq.command.ConsumerId; +import org.apache.activemq.command.ConsumerInfo; +import org.apache.activemq.command.MessageAck; +import org.apache.activemq.command.MessageId; +import org.apache.activemq.command.SessionId; +import org.apache.activemq.transport.TransportSupport; +import org.apache.activemq.util.ServiceStopper; +import org.apache.activemq.util.Wait; +import org.apache.activemq.wireformat.WireFormat; +import org.junit.After; +import org.junit.Test; + +/** + * Verifies that the advisory dispatch counter in + * {@link DemandForwardingBridgeSupport#ackAdvisory} neither loses nor + * double-claims dispatches under concurrent advisory delivery. + * + * The counter is incremented from remote-transport and executor threads; + * an unsynchronized read-modify-write (and an unguarded reset to zero) + * silently drops counted dispatches, so the periodic advisory ack + * under-acknowledges and the remote advisory prefetch window leaks until + * demand-subscription creation silently stops. + * + * Invariant asserted: sum of all acked message counts + counter residue + * == total dispatches observed. + */ +public class AckAdvisoryConcurrencyTest { + + private DemandForwardingBridge bridge; + private BrokerService brokerService; + + @After + public void tearDown() throws Exception { + if (bridge != null) { + shutdownExecutor(bridge, "serialExecutor"); + shutdownExecutor(bridge, "syncExecutor"); + } + if (brokerService != null) { + brokerService.getTaskRunnerFactory().shutdown(); + } + } + + @Test(timeout = 60000) + public void testConcurrentAckAdvisoryDoesNotLoseDispatches() throws Exception { + final var configuration = new NetworkBridgeConfiguration(); + final var localTransport = new RecordingTransport(); + final var remoteTransport = new RecordingTransport(); + + bridge = new DemandForwardingBridge(configuration, localTransport, remoteTransport); + brokerService = new BrokerService(); + // setBrokerService() dereferences the region broker of a started broker; + // ackAdvisory only needs the task runner factory, so set the field directly + bridge.brokerService = brokerService; + + var consumerInfo = new ConsumerInfo(); + consumerInfo.setConsumerId(new ConsumerId(new SessionId(new ConnectionId("advisory-storm"), 1), 1)); + consumerInfo.setPrefetchSize(1000); // threshold = 1000 * 75% = 750 + bridge.demandConsumerInfo = consumerInfo; + + final var advisory = new ActiveMQMessage(); + advisory.setMessageId(new MessageId("ID:advisory-storm-1:1:1:1")); + advisory.setDestination(new ActiveMQQueue("ActiveMQ.Advisory.Consumer.Queue.TEST")); + + final var threadCount = 8; + final var perThread = 100_000; + final var barrier = new CyclicBarrier(threadCount); + + var pool = Executors.newFixedThreadPool(threadCount); + try { + var futures = new ArrayList>(); + for (var t = 0; t < threadCount; t++) { + futures.add(pool.submit(() -> { + barrier.await(); + for (var i = 0; i < perThread; i++) { + bridge.ackAdvisory(advisory); + } + return null; + })); + } + // get() propagates worker failures that a raw thread would swallow + for (var future : futures) { + future.get(30, TimeUnit.SECONDS); + } + } finally { + pool.shutdownNow(); + } + + final var expected = (long) threadCount * perThread; + + // acks are sent async on the task runner; wait for the claimed total to + // converge (post-fix it reaches exactly `expected`; pre-fix it stalls short) + Wait.waitFor(() -> ackedTotal(remoteTransport) + counterResidue(bridge) == expected, 10_000, 100); + + assertEquals("acked dispatch counts + counter residue must equal total dispatches" + + " (lost updates in the advisory dispatch counter)", + expected, ackedTotal(remoteTransport) + counterResidue(bridge)); + } + + private static long ackedTotal(RecordingTransport remote) { + var total = 0L; + synchronized (remote.oneways) { + for (var command : remote.oneways) { + total += ((MessageAck) command).getMessageCount(); + } + } + return total; + } + + /** Reads the counter for both field shapes: plain int and AtomicInteger (a Number). */ + private static int counterResidue(DemandForwardingBridgeSupport bridge) throws Exception { + var f = DemandForwardingBridgeSupport.class.getDeclaredField("demandConsumerDispatched"); + f.setAccessible(true); + return ((Number) f.get(bridge)).intValue(); + } + + private static void shutdownExecutor(Object target, String fieldName) throws Exception { + var f = DemandForwardingBridgeSupport.class.getDeclaredField(fieldName); + f.setAccessible(true); + ((ExecutorService) f.get(target)).shutdownNow(); + } + + private static class RecordingTransport extends TransportSupport { + final List oneways = Collections.synchronizedList(new ArrayList<>()); + + @Override + public void oneway(Object command) { + oneways.add(command); + } + + @Override + public String getRemoteAddress() { + return "stub://recording"; + } + + @Override + public int getReceiveCounter() { + return 0; + } + + @Override + public X509Certificate[] getPeerCertificates() { + return null; + } + + @Override + public void setPeerCertificates(X509Certificate[] certificates) { + } + + @Override + public WireFormat getWireFormat() { + return null; + } + + @Override + protected void doStart() throws Exception { + } + + @Override + protected void doStop(ServiceStopper stopper) throws Exception { + } + } +} From 6280d964f24b84f08b59a4e235d7ab82fe98b87c Mon Sep 17 00:00:00 2001 From: Matt Pavlovich Date: Tue, 4 Aug 2026 10:19:27 -0500 Subject: [PATCH 2/3] Fix advisory dispatch counter races with AtomicInteger claim semantics Replace the plain-int demandConsumerDispatched with an AtomicInteger. The threshold check now claims the observed count via compareAndSet before acking, so concurrent advisory deliveries can neither lose increments nor ack overlapping counter ranges; a losing thread's increment stays counted for a later advisory to claim (ack lag bounded by one message). --- .../network/DemandForwardingBridgeSupport.java | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) 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 775590a3cf4..4c27a4d182e 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 @@ -37,6 +37,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Pattern; @@ -126,7 +127,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br protected String remoteBrokerName = "Unknown"; protected String localClientId; protected ConsumerInfo demandConsumerInfo; - protected int demandConsumerDispatched; + protected final AtomicInteger demandConsumerDispatched = new AtomicInteger(); protected final AtomicBoolean localBridgeStarted = new AtomicBoolean(false); protected final AtomicBoolean remoteBridgeStarted = new AtomicBoolean(false); protected final AtomicBoolean bridgeFailed = new AtomicBoolean(); @@ -941,10 +942,13 @@ public void onCompletion(FutureResponse resp) { } void ackAdvisory(Message message) throws IOException { - demandConsumerDispatched++; - if (demandConsumerDispatched > (demandConsumerInfo.getPrefetchSize() * - (configuration.getAdvisoryAckPercentage() / 100f))) { - final MessageAck ack = new MessageAck(message, MessageAck.STANDARD_ACK_TYPE, demandConsumerDispatched); + final int dispatched = demandConsumerDispatched.incrementAndGet(); + if (dispatched > (demandConsumerInfo.getPrefetchSize() * + (configuration.getAdvisoryAckPercentage() / 100f)) + // the CAS claims the observed count for this ack; a losing thread's + // increment stays in the counter for a later advisory to claim + && demandConsumerDispatched.compareAndSet(dispatched, 0)) { + final MessageAck ack = new MessageAck(message, MessageAck.STANDARD_ACK_TYPE, dispatched); ack.setConsumerId(demandConsumerInfo.getConsumerId()); brokerService.getTaskRunnerFactory().execute(new Runnable() { @Override @@ -956,7 +960,6 @@ public void run() { } } }); - demandConsumerDispatched = 0; } } From 822fd3aaaf1b1e8b18d5650d6aa02224666da3d9 Mon Sep 17 00:00:00 2001 From: Matt Pavlovich Date: Wed, 5 Aug 2026 15:32:59 -0500 Subject: [PATCH 3/3] [#2404] Flush advisory acks on an interval so quiet bridges are not aborted as slow The bridge's demand advisory consumer batched acks purely by advisoryAckPercentage, so a bridge holding a partial batch below the threshold never acked during quiet periods. New bridge config advisoryAckInterval (default 15000 ms, <= 0 disables) acknowledges pending advisory dispatches once the interval elapses: the --- .../DemandForwardingBridgeSupport.java | 115 +++++++-- .../network/NetworkBridgeConfiguration.java | 19 ++ .../network/AckAdvisoryTimeBasedAckTest.java | 230 ++++++++++++++++++ .../network/NetworkConnectorDefaultsTest.java | 1 + 4 files changed, 345 insertions(+), 20 deletions(-) create mode 100644 activemq-broker/src/test/java/org/apache/activemq/network/AckAdvisoryTimeBasedAckTest.java 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 4c27a4d182e..5d8760cf2c0 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 @@ -37,7 +37,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Pattern; @@ -127,7 +126,14 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br protected String remoteBrokerName = "Unknown"; protected String localClientId; protected ConsumerInfo demandConsumerInfo; - protected final AtomicInteger demandConsumerDispatched = new AtomicInteger(); + // Pending advisory dispatch accounting; guarded by advisoryAckLock. The count and + // the most recent counted dispatch are claimed together when building an ack so + // the STANDARD_ACK's lastMessageId always pairs with the batch it acknowledges. + private final Object advisoryAckLock = new Object(); + private int demandConsumerDispatched; + private Message lastAdvisoryDispatch; + private long lastAdvisoryAckTime = System.currentTimeMillis(); + private volatile Runnable advisoryAckFlushTask; protected final AtomicBoolean localBridgeStarted = new AtomicBoolean(false); protected final AtomicBoolean remoteBridgeStarted = new AtomicBoolean(false); protected final AtomicBoolean bridgeFailed = new AtomicBoolean(); @@ -285,6 +291,13 @@ public void onException(IOException error) { @Override public void stop() throws Exception { + // cancel outside the started branch: the flush task is scheduled during + // startRemoteBridge and must not outlive a bridge torn down mid-start + Runnable flushTask = advisoryAckFlushTask; + if (flushTask != null) { + advisoryAckFlushTask = null; + brokerService.getScheduler().cancel(flushTask); + } if (started.compareAndSet(true, false)) { if (disposed.compareAndSet(false, true)) { LOG.debug(" stopping {} bridge to {}", configuration.getBrokerName(), remoteBrokerName); @@ -657,6 +670,23 @@ protected void startRemoteBridge() throws Exception { demandConsumerInfo.setDestination(new ActiveMQTopic(advisoryTopic)); configureConsumerPrefetch(demandConsumerInfo); remoteBroker.oneway(demandConsumerInfo); + + final long advisoryAckInterval = configuration.getAdvisoryAckInterval(); + if (advisoryAckInterval > 0) { + synchronized (advisoryAckLock) { + lastAdvisoryAckTime = System.currentTimeMillis(); + } + advisoryAckFlushTask = new Runnable() { + @Override + public void run() { + flushPendingAdvisoryAcks(); + } + }; + // check at half the interval so pending acks age at most ~1.5x + // the interval before flushing + brokerService.getScheduler().executePeriodically(advisoryAckFlushTask, + Math.max(advisoryAckInterval / 2, 500)); + } } startedLatch.countDown(); } @@ -942,27 +972,72 @@ public void onCompletion(FutureResponse resp) { } void ackAdvisory(Message message) throws IOException { - final int dispatched = demandConsumerDispatched.incrementAndGet(); - if (dispatched > (demandConsumerInfo.getPrefetchSize() * - (configuration.getAdvisoryAckPercentage() / 100f)) - // the CAS claims the observed count for this ack; a losing thread's - // increment stays in the counter for a later advisory to claim - && demandConsumerDispatched.compareAndSet(dispatched, 0)) { - final MessageAck ack = new MessageAck(message, MessageAck.STANDARD_ACK_TYPE, dispatched); - ack.setConsumerId(demandConsumerInfo.getConsumerId()); - brokerService.getTaskRunnerFactory().execute(new Runnable() { - @Override - public void run() { - try { - remoteBroker.oneway(ack); - } catch (IOException e) { - LOG.warn("Failed to send advisory ack {}", ack, e); - } - } - }); + final MessageAck ack; + synchronized (advisoryAckLock) { + demandConsumerDispatched++; + lastAdvisoryDispatch = message; + ack = claimPendingAdvisoryAcks(false); + } + if (ack != null) { + sendAdvisoryAck(ack); + } + } + + /** + * Acknowledges any pending advisory dispatches once the configured + * advisoryAckInterval has elapsed. Runs periodically so a quiet bridge + * holding a partial batch below the advisoryAckPercentage threshold does + * not sit unacked indefinitely and get aborted as a slow consumer. + */ + void flushPendingAdvisoryAcks() { + final MessageAck ack; + synchronized (advisoryAckLock) { + ack = claimPendingAdvisoryAcks(true); + } + if (ack != null) { + sendAdvisoryAck(ack); } } + /** + * Claims the pending advisory dispatch count as a single ack when a + * trigger applies; callers must hold advisoryAckLock. The percentage + * threshold is the primary trigger on the dispatch path; the + * advisoryAckInterval is the trigger for the periodic flush and a + * secondary trigger on the dispatch path so a slow trickle of advisories + * cannot hold acks back indefinitely. + */ + private MessageAck claimPendingAdvisoryAcks(boolean timerTriggered) { + if (demandConsumerDispatched == 0 || demandConsumerInfo == null) { + return null; + } + final long interval = configuration.getAdvisoryAckInterval(); + final boolean intervalElapsed = interval > 0 + && System.currentTimeMillis() - lastAdvisoryAckTime >= interval; + if (!intervalElapsed && (timerTriggered || demandConsumerDispatched <= (demandConsumerInfo.getPrefetchSize() * + (configuration.getAdvisoryAckPercentage() / 100f)))) { + return null; + } + final MessageAck ack = new MessageAck(lastAdvisoryDispatch, MessageAck.STANDARD_ACK_TYPE, demandConsumerDispatched); + ack.setConsumerId(demandConsumerInfo.getConsumerId()); + demandConsumerDispatched = 0; + lastAdvisoryAckTime = System.currentTimeMillis(); + return ack; + } + + private void sendAdvisoryAck(final MessageAck ack) { + brokerService.getTaskRunnerFactory().execute(new Runnable() { + @Override + public void run() { + try { + remoteBroker.oneway(ack); + } catch (IOException e) { + LOG.warn("Failed to send advisory ack {}", ack, e); + } + } + }); + } + private void serviceRemoteConsumerAdvisory(DataStructure data) throws IOException { final int networkTTL = configuration.getConsumerTTL(); if (data.getClass() == ConsumerInfo.class) { diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java index e08ddd7fec8..a184ee53cb6 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java +++ b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java @@ -51,6 +51,7 @@ public class NetworkBridgeConfiguration { */ private int advisoryPrefetchSize = 0; private int advisoryAckPercentage = 75; + private long advisoryAckInterval = 15000; private int networkTTL = 1; private int consumerTTL = networkTTL; private int messageTTL = networkTTL; @@ -290,6 +291,24 @@ public void setAdvisoryAckPercentage(int advisoryAckPercentage) { this.advisoryAckPercentage = advisoryAckPercentage; } + public long getAdvisoryAckInterval() { + return advisoryAckInterval; + } + + /** + * @param advisoryAckInterval the maximum time in milliseconds pending + * advisory dispatches may wait before being acknowledged even though the + * advisoryAckPercentage threshold has not been reached. Keeps a quiet + * bridge from being aborted as a slow consumer by an + * abortSlowAckConsumerStrategy configured to not ignore network consumers + * (its default maxTimeSinceLastAck is 30000 - keep this value below it). + * Less than or equal to zero disables time-based advisory acks. + * Defaults to 15000. + */ + public void setAdvisoryAckInterval(long advisoryAckInterval) { + this.advisoryAckInterval = advisoryAckInterval; + } + /** * @return the userName */ diff --git a/activemq-broker/src/test/java/org/apache/activemq/network/AckAdvisoryTimeBasedAckTest.java b/activemq-broker/src/test/java/org/apache/activemq/network/AckAdvisoryTimeBasedAckTest.java new file mode 100644 index 00000000000..3dd93369355 --- /dev/null +++ b/activemq-broker/src/test/java/org/apache/activemq/network/AckAdvisoryTimeBasedAckTest.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.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.security.cert.X509Certificate; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ExecutorService; + +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ActiveMQMessage; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.command.ConnectionId; +import org.apache.activemq.command.ConsumerId; +import org.apache.activemq.command.ConsumerInfo; +import org.apache.activemq.command.MessageAck; +import org.apache.activemq.command.MessageId; +import org.apache.activemq.command.SessionId; +import org.apache.activemq.transport.TransportSupport; +import org.apache.activemq.util.ServiceStopper; +import org.apache.activemq.util.Wait; +import org.apache.activemq.wireformat.WireFormat; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** + * Verifies the time-based advisory ack flush in + * {@link DemandForwardingBridgeSupport#ackAdvisory}: pending advisory + * dispatches below the advisoryAckPercentage threshold must still be + * acknowledged once advisoryAckInterval elapses, so a quiet bridge is not + * flagged and aborted as a slow consumer by an abortSlowAckConsumerStrategy + * that does not ignore network consumers. + */ +public class AckAdvisoryTimeBasedAckTest { + + private NetworkBridgeConfiguration configuration; + private RecordingTransport remoteTransport; + private DemandForwardingBridge bridge; + private BrokerService brokerService; + private ActiveMQMessage advisory; + + @Before + public void setUp() throws Exception { + configuration = new NetworkBridgeConfiguration(); + remoteTransport = new RecordingTransport(); + + bridge = new DemandForwardingBridge(configuration, new RecordingTransport(), remoteTransport); + brokerService = new BrokerService(); + // setBrokerService() dereferences the region broker of a started broker; + // ackAdvisory only needs the task runner factory, so set the field directly + bridge.brokerService = brokerService; + + var consumerInfo = new ConsumerInfo(); + consumerInfo.setConsumerId(new ConsumerId(new SessionId(new ConnectionId("advisory-time"), 1), 1)); + consumerInfo.setPrefetchSize(1000); // threshold = 1000 * 75% = 750 + bridge.demandConsumerInfo = consumerInfo; + + advisory = new ActiveMQMessage(); + advisory.setMessageId(new MessageId("ID:advisory-time-1:1:1:1")); + advisory.setDestination(new ActiveMQQueue("ActiveMQ.Advisory.Consumer.Queue.TEST")); + } + + @After + public void tearDown() throws Exception { + if (bridge != null) { + shutdownExecutor(bridge, "serialExecutor"); + shutdownExecutor(bridge, "syncExecutor"); + } + if (brokerService != null) { + brokerService.getTaskRunnerFactory().shutdown(); + } + } + + @Test(timeout = 30000) + public void testFlushAcksPendingAfterInterval() throws Exception { + // dispatch below the percentage threshold while time-based acks cannot fire + configuration.setAdvisoryAckInterval(60_000); + for (var i = 0; i < 5; i++) { + bridge.ackAdvisory(advisory); + } + assertEquals(0, ackCount()); + + // let the (shortened) interval elapse, then flush as the timer would + configuration.setAdvisoryAckInterval(50); + Thread.sleep(80); + bridge.flushPendingAdvisoryAcks(); + + assertTrue("pending advisories should be acked by the interval flush", + Wait.waitFor(() -> ackedTotal() == 5, 5_000, 50)); + assertEquals(1, ackCount()); + } + + @Test(timeout = 30000) + public void testFlushIsNoOpBeforeInterval() throws Exception { + configuration.setAdvisoryAckInterval(60_000); + for (var i = 0; i < 5; i++) { + bridge.ackAdvisory(advisory); + } + + bridge.flushPendingAdvisoryAcks(); + + Thread.sleep(100); // acks are sent async; allow a wrong ack to surface + assertEquals("no ack may be sent before the interval elapses", 0, ackCount()); + } + + @Test(timeout = 30000) + public void testFlushDisabledWhenIntervalNotPositive() throws Exception { + configuration.setAdvisoryAckInterval(0); + for (var i = 0; i < 5; i++) { + bridge.ackAdvisory(advisory); + } + + Thread.sleep(80); + bridge.flushPendingAdvisoryAcks(); + + Thread.sleep(100); + assertEquals("time-based acks are disabled at interval <= 0", 0, ackCount()); + } + + @Test(timeout = 30000) + public void testDispatchPathAcksWhenIntervalElapsed() throws Exception { + // a slow trickle must flush via the dispatch path itself, without the timer + configuration.setAdvisoryAckInterval(60_000); + bridge.ackAdvisory(advisory); + assertEquals(0, ackCount()); + + configuration.setAdvisoryAckInterval(50); + Thread.sleep(80); + bridge.ackAdvisory(advisory); + + assertTrue("a dispatch after the interval must ack the pending batch", + Wait.waitFor(() -> ackedTotal() == 2, 5_000, 50)); + } + + @Test(timeout = 30000) + public void testPercentageThresholdStillAcksImmediately() throws Exception { + configuration.setAdvisoryAckInterval(60_000); + bridge.demandConsumerInfo.setPrefetchSize(4); // threshold = 3 + + for (var i = 0; i < 4; i++) { + bridge.ackAdvisory(advisory); + } + + assertTrue("crossing the percentage threshold must ack without waiting", + Wait.waitFor(() -> ackedTotal() == 4, 5_000, 50)); + assertEquals(1, ackCount()); + } + + private int ackCount() { + synchronized (remoteTransport.oneways) { + return remoteTransport.oneways.size(); + } + } + + private long ackedTotal() { + var total = 0L; + synchronized (remoteTransport.oneways) { + for (var command : remoteTransport.oneways) { + total += ((MessageAck) command).getMessageCount(); + } + } + return total; + } + + private static void shutdownExecutor(Object target, String fieldName) throws Exception { + var f = DemandForwardingBridgeSupport.class.getDeclaredField(fieldName); + f.setAccessible(true); + ((ExecutorService) f.get(target)).shutdownNow(); + } + + private static class RecordingTransport extends TransportSupport { + final List oneways = Collections.synchronizedList(new ArrayList<>()); + + @Override + public void oneway(Object command) { + oneways.add(command); + } + + @Override + public String getRemoteAddress() { + return "stub://recording"; + } + + @Override + public int getReceiveCounter() { + return 0; + } + + @Override + public X509Certificate[] getPeerCertificates() { + return null; + } + + @Override + public void setPeerCertificates(X509Certificate[] certificates) { + } + + @Override + public WireFormat getWireFormat() { + return null; + } + + @Override + protected void doStart() throws Exception { + } + + @Override + protected void doStop(ServiceStopper stopper) throws Exception { + } + } +} diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkConnectorDefaultsTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkConnectorDefaultsTest.java index b6f27f7a41e..3b71f6394a1 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkConnectorDefaultsTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkConnectorDefaultsTest.java @@ -42,6 +42,7 @@ public void testDefaultValues() throws Exception { nc.setName("NC1"); // Check values before calling .start() + assertEquals(Long.valueOf(15000L), Long.valueOf(nc.getAdvisoryAckInterval())); assertEquals(Integer.valueOf(75), Integer.valueOf(nc.getAdvisoryAckPercentage())); assertEquals(Integer.valueOf(0), Integer.valueOf(nc.getAdvisoryPrefetchSize())); assertEquals(Integer.valueOf(ConsumerInfo.NETWORK_CONSUMER_PRIORITY), Integer.valueOf(nc.getConsumerPriorityBase()));