Clean up

datastax · May 6, 2024 · aee5b24 · aee5b24
1 parent c088bbe
commit aee5b24
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 18 deletions.
diff --git a/.../src/main/java/com/datastax/oss/pulsar/jms/MessagePriorityGrowableArrayBlockingQueue.java b/.../src/main/java/com/datastax/oss/pulsar/jms/MessagePriorityGrowableArrayBlockingQueue.java
@@ -28,11 +28,6 @@
 
 @Slf4j
 public class MessagePriorityGrowableArrayBlockingQueue extends GrowableArrayBlockingQueue<Message> {
-
-  static int getPriority(Message m) {
-    return PulsarMessage.readJMSPriority(m, PulsarMessage.DEFAULT_PRIORITY);
-  }
-
   private final PriorityBlockingQueue<Pair<Integer, Message>> queue;
   private final AtomicBoolean terminated = new AtomicBoolean(false);
 
@@ -41,11 +36,12 @@ static int getPriority(Message m) {
 
   private static final Comparator<Pair<Integer, Message>> comparator =
       (o1, o2) -> {
+        // ORDER BY priority DESC, messageId ASC
         int priority1 = o1.getLeft();
         int priority2 = o2.getLeft();
         if (priority1 == priority2) {
           // if priorities are equal, we want to sort by messageId
-          return o2.getRight().getMessageId().compareTo(o1.getRight().getMessageId());
+          return o1.getRight().getMessageId().compareTo(o2.getRight().getMessageId());
         }
         return Integer.compare(priority2, priority1);
       };
@@ -93,7 +89,7 @@ public Message peek() {
       return null;
     }
     Message result = pair.getRight();
-    log.info("peeking message: {} prio {}", result.getMessageId(), getPriority(result));
+    log.info("peeking message: {} prio {}", result.getMessageId(), PulsarMessage.readJMSPriority(result));
     return result;
   }
 
@@ -102,12 +98,12 @@ public boolean offer(Message e) {
 
     boolean result;
     if (!this.terminated.get()) {
-      int prio = getPriority(e);
+      int prio = PulsarMessage.readJMSPriority(e);
       numberMessagesByPrority[prio].incrementAndGet();
       result = queue.offer(Pair.of(prio, e));
       if (log.isDebugEnabled()) {
         log.debug("offered message: {} prio {} stats {}",
-                e.getMessageId(), getPriority(e), Arrays.toString(numberMessagesByPrority));
+                e.getMessageId(), prio, Arrays.toString(numberMessagesByPrority));
       }
     } else {
       log.info("queue is terminated, not offering message: {}", e.getMessageId());

diff --git a/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java b/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java
@@ -1091,8 +1091,7 @@ Producer<byte[]> getProducerForDestination(Destination defaultDestination, boole
                             public int choosePartition(Message<?> msg, TopicMetadata metadata) {
 
                               int key =
-                                  PulsarMessage.readJMSPriority(
-                                      msg, PulsarMessage.DEFAULT_PRIORITY);
+                                  PulsarMessage.readJMSPriority(msg);
                               return Utils.mapPriorityToPartition(
                                   key,
                                   metadata.numPartitions(),
@@ -1348,8 +1347,8 @@ private static void replaceIncomingMessageList(Consumer c) {
                 new Comparator<Message>() {
                   @Override
                   public int compare(Message o1, Message o2) {
-                    int priority1 = MessagePriorityGrowableArrayBlockingQueue.getPriority(o1);
-                    int priority2 = MessagePriorityGrowableArrayBlockingQueue.getPriority(o2);
+                    int priority1 = PulsarMessage.readJMSPriority(o1);
+                    int priority2 = PulsarMessage.readJMSPriority(o2);
                     return Integer.compare(priority2, priority1);
                   }
                 });

diff --git a/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarMessage.java b/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarMessage.java
@@ -1378,7 +1378,7 @@ protected PulsarMessage applyMessage(
       this.correlationId = Base64.getDecoder().decode(msg.getProperty("JMSCorrelationID"));
     }
     if (msg.hasProperty("JMSPriority")) {
-      this.jmsPriority = readJMSPriority(msg, Message.DEFAULT_PRIORITY);
+      this.jmsPriority = readJMSPriority(msg);
     }
     if (msg.hasProperty("JMSDeliveryMode")) {
       try {
@@ -1478,14 +1478,17 @@ public org.apache.pulsar.client.api.Message<?> getReceivedPulsarMessage() {
     return receivedPulsarMessage;
   }
 
-  public static int readJMSPriority(org.apache.pulsar.client.api.Message<?> msg, int defaultValue) {
+  public static int readJMSPriority(org.apache.pulsar.client.api.Message<?> msg) {
     if (msg.hasProperty("JMSPriority")) {
       try {
-        return Integer.parseInt(msg.getProperty("JMSPriority"));
+        int value = Integer.parseInt(msg.getProperty("JMSPriority"));
+        if (value < 0 || value >= 10) {
+          return PulsarMessage.DEFAULT_PRIORITY;
+        }
       } catch (NumberFormatException err) {
         // cannot decode priority, not a big deal as it is not supported in Pulsar
       }
     }
-    return defaultValue;
+    return PulsarMessage.DEFAULT_PRIORITY;
   }
 }
diff --git a/.../test/java/com/datastax/oss/pulsar/jms/MessagePriorityGrowableArrayBlockingQueueTest.java b/.../test/java/com/datastax/oss/pulsar/jms/MessagePriorityGrowableArrayBlockingQueueTest.java
@@ -66,7 +66,7 @@ private static void test(int... priorities) {
   }
 
   private static int getPriority(Message m) {
-    return PulsarMessage.readJMSPriority(m, PulsarMessage.DEFAULT_PRIORITY);
+    return PulsarMessage.readJMSPriority(m);
   }
 
   private static Message messageWithPriority(int priority) {