ARTEMIS-5338 AckManager flow control

Say there are too many records in the JournalHashMap. In that case the sender should pace sending more data. The target should limit credits when the configured max pending records is reached.
apache · Mar 7, 2025 · 1a462dd · 1a462dd
1 parent 77199e7
commit 1a462dd
Show file tree

Hide file tree

Showing 8 changed files with 241 additions and 9 deletions.
diff --git a/...src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java b/...src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java
@@ -112,6 +112,8 @@ public static String getMirrorAddress(String connectionName) {
 
    private boolean directDeliver = true;
 
+   private int mirrorMaxPendingAcks = 10_000;
+
    private final AMQPRoutingHandler routingHandler;
 
    /*
@@ -164,6 +166,15 @@ public int getAmqpMinLargeMessageSize() {
       return amqpMinLargeMessageSize;
    }
 
+   public int getMirrorMaxPendingAcks() {
+      return mirrorMaxPendingAcks;
+   }
+
+   public ProtonProtocolManager setMirrorMaxPendingAcks(int maxPendingAcks) {
+      this.mirrorMaxPendingAcks = maxPendingAcks;
+      return this;
+   }
+
    public ProtonProtocolManager setAmqpMinLargeMessageSize(int amqpMinLargeMessageSize) {
       this.amqpMinLargeMessageSize = amqpMinLargeMessageSize;
       return this;

diff --git a/.../org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerTarget.java b/.../org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerTarget.java
@@ -100,6 +100,27 @@ public static MirrorController getControllerInUse() {
       return CONTROLLER_THREAD_LOCAL.get();
    }
 
+
+   @Override
+   public boolean isBusy() {
+      return getAckManager().size() > getMirrorMaxPendingAcks();
+   }
+
+   public int getMirrorMaxPendingAcks() {
+      try {
+         return connection.getProtocolManager().getMirrorMaxPendingAcks();
+      } catch (Throwable e) {
+         // It shouldn't happen, but if it did we just log it
+         logger.warn(e.getMessage(), e);
+         return 0;
+      }
+   }
+
+   public void verifyCredits() {
+      connection.runNow(creditTopUpRunner);
+   }
+
+
    /**
     * Objects of this class can be used by either transaction or by OperationContext. It is important that when you're
     * using the transactions you clear any references to the operation context. Don't use transaction and
@@ -473,14 +494,17 @@ private void performAck(String nodeID,
          logger.trace("performAck (nodeID={}, messageID={}), targetQueue={})", nodeID, messageID, targetQueue.getName());
       }
 
+      getAckManager().ack(nodeID, targetQueue, messageID, reason, true);
+
+      OperationContextImpl.getContext().executeOnCompletion(ackMessageOperation, OperationConsistencyLevel.FULL);
+   }
+
+   private AckManager getAckManager() {
       if (ackManager == null) {
          ackManager = AckManagerProvider.getManager(server);
          ackManager.registerMirror(this);
       }
-
-      ackManager.ack(nodeID, targetQueue, messageID, reason, true);
-
-      OperationContextImpl.getContext().executeOnCompletion(ackMessageOperation, OperationConsistencyLevel.FULL);
+      return ackManager;
    }
 
    /**

diff --git a/...ol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManager.java b/...ol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManager.java
@@ -28,6 +28,7 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.function.LongSupplier;
 
 import io.netty.util.collection.LongObjectHashMap;
@@ -78,6 +79,13 @@ public class AckManager implements ActiveMQComponent {
    volatile MultiStepProgress progress;
    ActiveMQScheduledComponent scheduledComponent;
 
+   private volatile int size;
+   private static final AtomicIntegerFieldUpdater<AckManager> sizeUpdater = AtomicIntegerFieldUpdater.newUpdater(AckManager.class, "size");
+
+   public int size() {
+      return sizeUpdater.get(this);
+   }
+
    public AckManager(ActiveMQServer server) {
       assert server != null && server.getConfiguration() != null;
       this.server = server;
@@ -92,6 +100,7 @@ public AckManager(ActiveMQServer server) {
 
    public void reload(RecordInfo recordInfo) {
       journalHashMapProvider.reload(recordInfo);
+      sizeUpdater.incrementAndGet(this);
    }
 
    @Override
@@ -104,6 +113,13 @@ public synchronized void stop() {
       logger.debug("Stopping ackmanager on server {}", server);
    }
 
+   public synchronized void pause() {
+      if (scheduledComponent != null) {
+         scheduledComponent.stop();
+         scheduledComponent = null;
+      }
+   }
+
    @Override
    public synchronized boolean isStarted() {
       return scheduledComponent != null && scheduledComponent.isStarted();
@@ -188,6 +204,11 @@ private void flushMirrorTargets() {
       targetCopy.forEach(AMQPMirrorControllerTarget::flush);
    }
 
+   private void checkFlowControlMirrorTargets() {
+      List<AMQPMirrorControllerTarget> targetCopy = copyTargets();
+      targetCopy.forEach(AMQPMirrorControllerTarget::verifyCredits);
+   }
+
    private synchronized List<AMQPMirrorControllerTarget> copyTargets() {
       return new ArrayList<>(mirrorControllerTargets);
    }
@@ -267,6 +288,8 @@ public void retryAddress(SimpleString address, LongObjectHashMap<JournalHashMap<
             logger.trace("Page Scan not required for address {}", address);
          }
 
+         checkFlowControlMirrorTargets();
+
       } catch (Throwable e) {
          logger.warn(e.getMessage(), e);
       } finally {
@@ -299,7 +322,9 @@ private void validateExpireSet(SimpleString address, long queueID, JournalHashMa
                if (logger.isDebugEnabled()) {
                   logger.debug("Retried {} {} times, giving up on the entry now. Configured Page Attempts={}", retry, retry.getPageAttempts(), configuration.getMirrorAckManagerPageAttempts());
                }
-               retries.remove(retry);
+               if (retries.remove(retry) != null) {
+                  sizeUpdater.decrementAndGet(AckManager.this);
+               }
             } else {
                if (logger.isDebugEnabled()) {
                   logger.debug("Retry {} attempted {} times on paging, Configuration Page Attempts={}", retry, retry.getPageAttempts(), configuration.getMirrorAckManagerPageAttempts());
@@ -353,7 +378,9 @@ private void retryPage(LongObjectHashMap<JournalHashMap<AckRetry, AckRetry, Queu
                            }
                         }
                      }
-                     retries.remove(ackRetry, transaction.getID());
+                     if (retries.remove(ackRetry, transaction.getID()) != null) {
+                        sizeUpdater.decrementAndGet(AckManager.this);
+                     }
                      transaction.setContainsPersistent();
                      logger.trace("retry performed ok, ackRetry={} for message={} on queue", ackRetry, pagedMessage);
                   }
@@ -389,6 +416,7 @@ private boolean checkRetriesAndPaging(LongObjectHashMap<JournalHashMap<AckRetry,
             if (ack(retry.getNodeID(), queue, retry.getMessageID(), retry.getReason(), false)) {
                logger.trace("Removing retry {} as the retry went ok", retry);
                queueRetries.remove(retry);
+               sizeUpdater.decrementAndGet(this);
             } else {
                int retried = retry.attemptedQueue();
                if (logger.isTraceEnabled()) {
@@ -410,6 +438,7 @@ public synchronized void addRetry(String nodeID, Queue queue, long messageID, Ac
       }
       AckRetry retry = new AckRetry(nodeID, messageID, reason);
       journalHashMapProvider.getMap(queue.getID(), queue).put(retry, retry);
+      sizeUpdater.incrementAndGet(this);
       if (scheduledComponent != null) {
          // we set the retry delay again in case it was changed.
          scheduledComponent.setPeriod(configuration.getMirrorAckManagerRetryDelay());

diff --git a/...ain/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManagerProvider.java b/...ain/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManagerProvider.java
@@ -49,6 +49,9 @@ public static int getSize() {
    }
 
    public static AckManager getManager(ActiveMQServer server) {
+      if (server == null) {
+         throw new NullPointerException("server is null");
+      }
       synchronized (managerHashMap) {
          AckManager ackManager = managerHashMap.get(server);
          if (ackManager != null) {

diff --git a/...rc/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonAbstractReceiver.java b/...rc/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonAbstractReceiver.java
@@ -175,6 +175,10 @@ public boolean isStarted() {
       return state == ReceiverState.STARTED;
    }
 
+   public boolean isBusy() {
+      return false;
+   }
+
    public boolean isStopping() {
       return state == ReceiverState.STOPPING;
    }
@@ -276,7 +280,7 @@ public void run() {
          if (connection.isHandler()) {
             connection.requireInHandler();
 
-            if (context.isStarted()) {
+            if (context.isStarted() && !context.isBusy()) {
                final int pending = context.pendingSettles;
 
                if (isBellowThreshold(receiver.getCredit(), pending, threshold)) {

diff --git a/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/utils/Wait.java b/artemis-unit-test-support/src/main/java/org/apache/activemq/artemis/utils/Wait.java
@@ -172,6 +172,11 @@ public static void assertTrue(String failureMessage, Condition condition, final
       assertTrue(failureMessage, condition, duration, SLEEP_MILLIS);
    }
 
+   public static <T> T assertNotNull(Supplier<T> supplier, final long duration, final long sleep) throws Exception {
+      Assertions.assertTrue(waitFor(() -> supplier.get() != null, duration, sleep));
+      return supplier.get();
+   }
+
    public static void assertTrue(Condition condition, final long duration, final long sleep) {
       assertTrue(DEFAULT_FAILURE_MESSAGE, condition, duration, sleep);
    }