JMS Selectors: reduce impact of JMSExpirations checks (#127)

datastax · Apr 12, 2024 · a01348a · a01348a
1 parent bb30452
commit a01348a
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 9 deletions.
diff --git a/pulsar-jms-filters/src/main/java/com/datastax/oss/pulsar/jms/selectors/JMSFilter.java b/pulsar-jms-filters/src/main/java/com/datastax/oss/pulsar/jms/selectors/JMSFilter.java
@@ -51,6 +51,10 @@ public class JMSFilter implements EntryFilter {
 
   @Override
   public FilterResult filterEntry(Entry entry, FilterContext context) {
+    return filterEntry(entry, context, false);
+  }
+
+  public FilterResult filterEntry(Entry entry, FilterContext context, boolean onMessagePublish) {
     Consumer consumer = context.getConsumer();
     Map<String, String> consumerMetadata =
         consumer != null ? consumer.getMetadata() : Collections.emptyMap();
@@ -246,14 +250,6 @@ public FilterResult filterEntry(Entry entry, FilterContext context) {
                 null,
                 metadata);
 
-        // timetoLive filter
-        long jmsExpiration = getJMSExpiration(typedProperties);
-        if (jmsExpiration > 0 && System.currentTimeMillis() > jmsExpiration) {
-          // message expired, this does not depend on the Consumer
-          // we can to REJECT it immediately
-          return FilterResult.REJECT;
-        }
-
         if (!jmsSelectorOnSubscription.isEmpty()) {
           boolean matchesSubscriptionFilter = matches(typedProperties, selectorOnSubscription);
           // the subscription filter always deletes the messages
@@ -262,6 +258,18 @@ public FilterResult filterEntry(Entry entry, FilterContext context) {
           }
         }
 
+        // it is expensive to extract the message properties
+        // and it is very unlikely that onPublish we are accepting a message that is already expired
+        if (!onMessagePublish) {
+          // timetoLive filter
+          long jmsExpiration = getJMSExpiration(typedProperties);
+          if (jmsExpiration > 0 && System.currentTimeMillis() > jmsExpiration) {
+            // message expired, this does not depend on the Consumer
+            // we can to REJECT it immediately
+            return FilterResult.REJECT;
+          }
+        }
+
         boolean matches = true;
         if (selector != null) {
           matches = matches(typedProperties, selector);

diff --git a/...ar-jms-filters/src/main/java/com/datastax/oss/pulsar/jms/selectors/JMSPublishFilters.java b/...ar-jms-filters/src/main/java/com/datastax/oss/pulsar/jms/selectors/JMSPublishFilters.java
@@ -150,7 +150,7 @@ public void messageProduced(
             filterContext.setMsgMetadata(messageMetadata);
             filterContext.setConsumer(null);
             Entry entry = null; // we would need the Entry only in case of batch messages
-            EntryFilter.FilterResult filterResult = filter.filterEntry(entry, filterContext);
+            EntryFilter.FilterResult filterResult = filter.filterEntry(entry, filterContext, true);
             if (filterResult == EntryFilter.FilterResult.REJECT) {
               if (log.isDebugEnabled()) {
                 log.debug(