#2020 Add enableExpiry option to PolicyEntry to skip O(n) expiry scan

activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java

@@ -25,7 +25,6 @@
 import org.apache.activemq.broker.region.policy.MessageEvictionStrategy;
 import org.apache.activemq.broker.region.policy.OldestMessageEvictionStrategy;
 import org.apache.activemq.command.*;
-import org.apache.activemq.management.MessageFlowStats;
 import org.apache.activemq.thread.Scheduler;
 import org.apache.activemq.transaction.Synchronization;
 import org.apache.activemq.transport.TransmitCallback;
@@ -166,7 +165,7 @@ public void add(MessageReference node) throws Exception {
                         if (maximumPendingMessages > 0 && maximumPendingMessages < max) {
                             max = maximumPendingMessages;
                         }
-                        if (!matched.isEmpty() && matched.size() > max) {
+                        if (messageEvictionStrategy.isExpiryScanEnabled() && !matched.isEmpty() && matched.size() > max) {
                             removeExpiredMessages();
                         }
                         // lets discard old messages as we are a slow consumer
@@ -530,6 +529,7 @@ public void setMaximumPendingMessages(int maximumPendingMessages) {
         this.maximumPendingMessages = maximumPendingMessages;
     }
 
+
     public MessageEvictionStrategy getMessageEvictionStrategy() {
         return messageEvictionStrategy;
     }

activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java

@@ -42,4 +42,18 @@ public interface MessageEvictionStrategy {
      */
     int getEvictExpiredMessagesHighWatermark();
 
+    /**
+     * Returns whether the eager expired-message scan is enabled.
+     * <p>
+     * When {@code false}, the O(n) scan inside
+     * {@link org.apache.activemq.broker.region.TopicSubscription#add} is skipped entirely.
+     * Set to {@code false} when messages carry no TTL, or when the scan cost outweighs
+     * the benefit of eagerly evicting expired messages from slow-consumer buffers.
+     * <p>
+     * See {@link MessageEvictionStrategySupport} for the default implementation that returns {@code true}.
+     *
+     * @return {@code true} if the expiry scan is enabled (default), {@code false} if skipped
+     */
+    boolean isExpiryScanEnabled();
+
 }

activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategySupport.java

@@ -24,6 +24,7 @@
 public abstract class MessageEvictionStrategySupport implements MessageEvictionStrategy {
 
     private int evictExpiredMessagesHighWatermark = 1000;
+    private boolean expiryScanEnabled = true;
 
     public int getEvictExpiredMessagesHighWatermark() {
         return evictExpiredMessagesHighWatermark;
@@ -35,6 +36,27 @@ public int getEvictExpiredMessagesHighWatermark() {
     public void setEvictExpiredMessagesHighWatermark(int evictExpiredMessagesHighWaterMark) {
         this.evictExpiredMessagesHighWatermark = evictExpiredMessagesHighWaterMark;
     }
-
-
+
+    @Override
+    public boolean isExpiryScanEnabled() {
+        return expiryScanEnabled;
+    }
+
+    /**
+     * Controls whether the broker performs an eager expired-message scan when a
+     * non-durable topic subscription's pending slow-consumer buffer exceeds
+     * {@link #getEvictExpiredMessagesHighWatermark()}.
+     * <p>
+     * Set to {@code false} when messages carry no TTL, or when the O(n) scan cost
+     * outweighs the benefit of eagerly evicting expired messages from slow-consumer
+     * buffers. When messages have no TTL, every scan iterates the full buffer without
+     * removing anything, adding latency to every enqueue once the buffer exceeds the
+     * high-water mark.
+     *
+     * @param expiryScanEnabled {@code false} to skip the scan; {@code true} to enable it (default)
+     */
+    public void setExpiryScanEnabled(boolean expiryScanEnabled) {
+        this.expiryScanEnabled = expiryScanEnabled;
+    }
+
 }

activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java

@@ -119,6 +119,7 @@ public class PolicyEntry extends DestinationMapEntry {
 
     private MessageInterceptorStrategy messageInterceptorStrategy = null;
 
+
     public void configure(Broker broker,Queue queue) {
         baseConfiguration(broker,queue);
         if (dispatchPolicy != null) {
@@ -704,6 +705,7 @@ public void setEnableAudit(boolean enableAudit) {
         this.enableAudit = enableAudit;
     }
 
+
     public int getMaxQueueAuditDepth() {
         return maxQueueAuditDepth;
     }