ARTEMIS-5142 support never expiring incoming messages

apache · Dec 13, 2024 · a636a41 · a636a41
1 parent c7ec3c7
commit a636a41
Show file tree

Hide file tree

Showing 7 changed files with 621 additions and 31 deletions.
diff --git a/...server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/...server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -1355,28 +1355,44 @@ private RoutingStatus maybeSendToDLA(final Message message,
       return status;
    }
 
-   // HORNETQ-1029
-   private static void applyExpiryDelay(Message message, AddressSettings settings) {
+   protected static void applyExpiryDelay(Message message, AddressSettings settings) {
       long expirationOverride = settings.getExpiryDelay();
 
-      // A -1 <expiry-delay> means don't do anything
-      if (expirationOverride >= 0) {
-         // only override the expiration on messages where the expiration hasn't been set by the user
+      if (settings.getNeverExpire()) {
+         if (message.getExpiration() != 0) {
+            message.setExpiration(0);
+            message.reencode();
+         }
+      } else if (expirationOverride >= 0) {
+         // A -1 <expiry-delay> means don't do anything
          if (message.getExpiration() == 0) {
-            message.setExpiration(System.currentTimeMillis() + expirationOverride);
+            // only override the expiration on messages where the expiration hasn't been set by the user
+            setExpiration(message, expirationOverride);
          }
       } else {
          long minExpiration = settings.getMinExpiryDelay();
          long maxExpiration = settings.getMaxExpiryDelay();
 
-         if (maxExpiration != AddressSettings.DEFAULT_MAX_EXPIRY_DELAY && (message.getExpiration() == 0 || message.getExpiration() > (System.currentTimeMillis() + maxExpiration))) {
-            message.setExpiration(System.currentTimeMillis() + maxExpiration);
+         if (message.getExpiration() == 0) {
+            // if the incoming message has NO expiration then apply the max if set and if not set then apply the min if set
+            if (maxExpiration != AddressSettings.DEFAULT_MAX_EXPIRY_DELAY) {
+               setExpiration(message, maxExpiration);
+            } else if (minExpiration != AddressSettings.DEFAULT_MIN_EXPIRY_DELAY) {
+               setExpiration(message, minExpiration);
+            }
+         } else if (maxExpiration != AddressSettings.DEFAULT_MAX_EXPIRY_DELAY && message.getExpiration() > (System.currentTimeMillis() + maxExpiration)) {
+            setExpiration(message, maxExpiration);
          } else if (minExpiration != AddressSettings.DEFAULT_MIN_EXPIRY_DELAY && message.getExpiration() < (System.currentTimeMillis() + minExpiration)) {
-            message.setExpiration(System.currentTimeMillis() + minExpiration);
+            setExpiration(message, minExpiration);
          }
       }
    }
 
+   private static void setExpiration(Message m, long expiration) {
+      m.setExpiration(System.currentTimeMillis() + expiration);
+      m.reencode();
+   }
+
    @Override
    public MessageReference reload(final Message message, final Queue queue, final Transaction tx) throws Exception {
 

diff --git a/...-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java b/...-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/AddressSettings.java
@@ -117,6 +117,8 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
 
    public static final long DEFAULT_MAX_EXPIRY_DELAY = -1;
 
+   public static final boolean DEFAULT_NEVER_EXPIRE = false;
+
    public static final boolean DEFAULT_SEND_TO_DLA_ON_NO_ROUTE = false;
 
    public static final long DEFAULT_SLOW_CONSUMER_THRESHOLD = -1;
@@ -266,6 +268,11 @@ public class AddressSettings implements Mergeable<AddressSettings>, Serializable
    }
    private Long maxExpiryDelay = null;
 
+   static {
+      metaBean.add(Boolean.class, "neverExpire", (t, p) -> t.neverExpire = p, t -> t.neverExpire);
+   }
+   private Boolean neverExpire = null;
+
    static {
       metaBean.add(Boolean.class, "defaultLastValueQueue", (t, p) -> t.defaultLastValueQueue = p, t -> t.defaultLastValueQueue);
    }
@@ -1050,6 +1057,15 @@ public AddressSettings setMaxExpiryDelay(final Long maxExpiryDelay) {
       return this;
    }
 
+   public Boolean getNeverExpire() {
+      return neverExpire != null ? neverExpire : AddressSettings.DEFAULT_NEVER_EXPIRE;
+   }
+
+   public AddressSettings setNeverExpire(final Boolean neverExpire) {
+      this.neverExpire = neverExpire;
+      return this;
+   }
+
    public boolean isSendToDLAOnNoRoute() {
       return sendToDLAOnNoRoute != null ? sendToDLAOnNoRoute : AddressSettings.DEFAULT_SEND_TO_DLA_ON_NO_ROUTE;
    }

diff --git a/artemis-server/src/main/resources/schema/artemis-configuration.xsd b/artemis-server/src/main/resources/schema/artemis-configuration.xsd
@@ -3871,6 +3871,14 @@
             </xsd:annotation>
          </xsd:element>
 
+         <xsd:element name="never-expire" type="xsd:boolean" default="false" maxOccurs="1" minOccurs="0">
+            <xsd:annotation>
+               <xsd:documentation>
+                  Overrides the expiration time for all messages so that they never expire.
+               </xsd:documentation>
+            </xsd:annotation>
+         </xsd:element>
+
          <xsd:element name="expiry-delay" type="xsd:long" default="-1" maxOccurs="1" minOccurs="0">
             <xsd:annotation>
                <xsd:documentation>