[ISSUE #1152] [RIP-80] [Java] Implementation of Priority Message (#1153)

imzs · web-flow · commit 8903b2fd5a04 · 2025-12-29T14:34:51.000+08:00
Change-Id: Ib4d1d67573948a5c66ec1e0c83082d2bfb44d9e6
diff --git a/README-CN.md b/README-CN.md
@@ -28,6 +28,7 @@
 | Push consumer with concurrent message listener |   ✅   |   ✅   |   ✅   |   🚧    |   ✅   |   🚧    |    🚧    |   🚧   |
 | Push consumer with FIFO message listener       |   ✅   |   ✅   |   ✅   |   🚧    |   ✅   |   🚧    |    🚧    |   🚧   |
 | Push consumer with FIFO consume accelerator    |   ✅   |   ✅   |   🚧   |   🚧    |   🚧   |   🚧    |    🚧    |   🚧   |
+| Priority Message                               |   ✅   |   🚧   |   🚧   |   🚧    |   🚧   |   🚧    |    🚧    |   🚧   |
 
 ## 先决条件和构建
 
diff --git a/README.md b/README.md
@@ -28,6 +28,7 @@ Provide cloud-native and robust solutions for Java, C++, C#, Golang, Rust and al
 | Push consumer with concurrent message listener |   ✅   |   ✅   |   ✅   |   🚧    |   ✅   |   🚧    |    🚧    |   🚧   |
 | Push consumer with FIFO message listener       |   ✅   |   ✅   |   ✅   |   🚧    |   ✅   |   🚧    |    🚧    |   🚧   |
 | Push consumer with FIFO consume accelerator    |   ✅   |   ✅   |   🚧   |   🚧    |   🚧   |   🚧    |    🚧    |   🚧   |
+| Priority Message                               |   ✅   |   🚧   |   🚧   |   🚧    |   🚧   |   🚧    |    🚧    |   🚧   |
 
 ## Prerequisite and Build
 
diff --git a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/Message.java b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/Message.java
@@ -86,4 +86,11 @@ public interface Message {
      * timestamp is not specified.
      */
     Optional<Long> getDeliveryTimestamp();
+
+    /**
+     * Get the priority of the message, which makes sense only when topic type is priority.
+     *
+     * @return message priority, which is optional, {@link Optional#empty()} means priority is not specified.
+     */
+    Optional<Integer> getPriority();
 }
diff --git a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageBuilder.java b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageBuilder.java
@@ -105,6 +105,13 @@ public interface MessageBuilder {
      */
     MessageBuilder setDeliveryTimestamp(long deliveryTimestamp);
 
+    /**
+     * Set the priority for the message, which is optional.
+     * @param priority non-negative number in the range [0, N], regarded as highest priority if exceeds N
+     * @return the message builder instance.
+     */
+    MessageBuilder setPriority(int priority);
+
     /**
      * Add user property for the message.
      *
diff --git a/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java b/java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageView.java
@@ -94,6 +94,13 @@ public interface MessageView {
      */
     Optional<Long> getDeliveryTimestamp();
 
+    /**
+     * Get the priority of the message, which makes sense only when topic type is priority.
+     *
+     * @return message priority, which is optional, {@link Optional#empty()} means priority is not specified.
+     */
+    Optional<Integer> getPriority();
+
     /**
      * Get the born host of the message.
      *
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/example/ProducerPriorityMessageExample.java b/java/client/src/main/java/org/apache/rocketmq/client/java/example/ProducerPriorityMessageExample.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.client.java.example;
+
+import java.nio.charset.StandardCharsets;
+import org.apache.rocketmq.client.apis.ClientException;
+import org.apache.rocketmq.client.apis.ClientServiceProvider;
+import org.apache.rocketmq.client.apis.message.Message;
+import org.apache.rocketmq.client.apis.producer.Producer;
+import org.apache.rocketmq.client.apis.producer.SendReceipt;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ProducerPriorityMessageExample {
+    private static final Logger log = LoggerFactory.getLogger(ProducerPriorityMessageExample.class);
+
+    private ProducerPriorityMessageExample() {
+    }
+
+    public static void main(String[] args) throws ClientException {
+        final ClientServiceProvider provider = ClientServiceProvider.loadService();
+
+        String topic = "yourPriorityTopic";
+        final Producer producer = ProducerSingleton.getInstance(topic);
+        // Define your message body.
+        byte[] body = "This is a delay message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
+        String tag = "yourMessageTagA";
+        final Message message = provider.newMessageBuilder()
+            // Set topic for the current message.
+            .setTopic(topic)
+            // Message secondary classifier of message besides topic.
+            .setTag(tag)
+            // Key(s) of the message, another way to mark message besides message id.
+            .setKeys("yourMessageKey")
+            // Set priority of message.
+            .setPriority(1)
+            .setBody(body)
+            .build();
+        try {
+            final SendReceipt sendReceipt = producer.send(message);
+            log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
+        } catch (Throwable t) {
+            log.error("Failed to send message", t);
+        }
+        // Close the producer when you don't need it anymore.
+        // You could close it manually or add this into the JVM shutdown hook.
+        // producer.close();
+    }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessage.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessage.java
@@ -96,6 +96,13 @@ public interface GeneralMessage {
      */
     Optional<Long> getDeliveryTimestamp();
 
+    /**
+     * Get the priority of the message, which makes sense only when topic type is priority.
+     *
+     * @return message priority, which is optional, {@link Optional#empty()} means priority is not specified.
+     */
+    Optional<Integer> getPriority();
+
     /**
      * Get the born host of the message.
      *
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessageImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/GeneralMessageImpl.java
@@ -37,6 +37,7 @@ public class GeneralMessageImpl implements GeneralMessage {
     private final String messageGroup;
     private final String liteTopic;
     private final Long deliveryTimestamp;
+    private final Integer priority;
     private final String bornHost;
     private final Long bornTimestamp;
     private final Integer deliveryAttempt;
@@ -61,6 +62,7 @@ public GeneralMessageImpl(Message message) {
         this.messageGroup = message.getMessageGroup().orElse(null);
         this.liteTopic = message.getLiteTopic().orElse(null);
         this.deliveryTimestamp = message.getDeliveryTimestamp().orElse(null);
+        this.priority = message.getPriority().orElse(null);
         this.bornHost = null;
         this.bornTimestamp = null;
         this.deliveryAttempt = null;
@@ -96,6 +98,7 @@ public GeneralMessageImpl(MessageView message) {
         this.messageGroup = message.getMessageGroup().orElse(null);
         this.liteTopic = message.getLiteTopic().orElse(null);
         this.deliveryTimestamp = message.getDeliveryTimestamp().orElse(null);
+        this.priority = message.getPriority().orElse(null);
         this.bornHost = message.getBornHost();
         this.bornTimestamp = message.getBornTimestamp();
         this.deliveryAttempt = message.getDeliveryAttempt();
@@ -149,6 +152,11 @@ public Optional<Long> getDeliveryTimestamp() {
         return Optional.ofNullable(deliveryTimestamp);
     }
 
+    @Override
+    public Optional<Integer> getPriority() {
+        return Optional.ofNullable(priority);
+    }
+
     @Override
     public Optional<String> getBornHost() {
         return Optional.ofNullable(bornHost);
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageBuilderImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageBuilderImpl.java
@@ -41,6 +41,7 @@ public class MessageBuilderImpl implements MessageBuilder {
     protected String messageGroup = null;
     protected String liteTopic = null;
     protected Long deliveryTimestamp = null;
+    protected Integer priority = null;
     protected Collection<String> keys = new HashSet<>();
     protected final Map<String, String> properties = new HashMap<>();
 
@@ -100,6 +101,7 @@ public MessageBuilder setKeys(String... keys) {
     public MessageBuilder setMessageGroup(String messageGroup) {
         checkArgument(null == deliveryTimestamp, "messageGroup and deliveryTimestamp should not be set at same time");
         checkArgument(null == liteTopic, "messageGroup and liteTopic should not be set at same time");
+        checkArgument(null == priority, "messageGroup and priority should not be set at same time");
         checkArgument(StringUtils.isNotBlank(messageGroup), "messageGroup should not be blank");
         this.messageGroup = messageGroup;
         return this;
@@ -109,6 +111,7 @@ public MessageBuilder setMessageGroup(String messageGroup) {
     public MessageBuilder setLiteTopic(String liteTopic) {
         checkArgument(null == deliveryTimestamp, "liteTopic and deliveryTimestamp should not be set at same time");
         checkArgument(null == messageGroup, "liteTopic and messageGroup should not be set at same time");
+        checkArgument(null == priority, "liteTopic and priority should not be set at same time");
         checkArgument(StringUtils.isNotBlank(liteTopic), "liteTopic should not be blank");
         this.liteTopic = liteTopic;
         return this;
@@ -121,10 +124,24 @@ public MessageBuilder setLiteTopic(String liteTopic) {
     public MessageBuilder setDeliveryTimestamp(long deliveryTimestamp) {
         checkArgument(null == messageGroup, "deliveryTimestamp and messageGroup should not be set at same time");
         checkArgument(null == liteTopic, "deliveryTimestamp and liteTopic should not be set at same time");
+        checkArgument(null == priority, "deliveryTimestamp and priority should not be set at same time");
         this.deliveryTimestamp = deliveryTimestamp;
         return this;
     }
 
+    /**
+     * See {@link MessageBuilder#setPriority(int)}
+     */
+    @Override
+    public MessageBuilder setPriority(int priority) {
+        checkArgument(null == deliveryTimestamp, "priority and deliveryTimestamp should not be set at same time");
+        checkArgument(null == messageGroup, "priority and messageGroup should not be set at same time");
+        checkArgument(null == liteTopic, "priority and liteTopic should not be set at same time");
+        checkArgument(priority >= 0, "priority must be greater than or equal to 0");
+        this.priority = priority;
+        return this;
+    }
+
     /**
      * See {@link MessageBuilder#addProperty(String, String)}
      */
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageImpl.java
@@ -49,6 +49,8 @@ public class MessageImpl implements Message {
     private final String liteTopic;
     @Nullable
     private final Long deliveryTimestamp;
+    @Nullable
+    private final Integer priority;
 
     /**
      * The caller is supposed to have validated the arguments and handled throwing exception or
@@ -61,6 +63,7 @@ public class MessageImpl implements Message {
         this.messageGroup = builder.messageGroup;
         this.liteTopic = builder.liteTopic;
         this.deliveryTimestamp = builder.deliveryTimestamp;
+        this.priority = builder.priority;
         this.keys = builder.keys;
         this.properties = builder.properties;
     }
@@ -81,6 +84,7 @@ public class MessageImpl implements Message {
         this.messageGroup = message.getMessageGroup().orElse(null);
         this.deliveryTimestamp = message.getDeliveryTimestamp().orElse(null);
         this.liteTopic = message.getLiteTopic().orElse(null);
+        this.priority = message.getPriority().orElse(null);
         this.keys = message.getKeys();
         this.properties = message.getProperties();
     }
@@ -133,6 +137,14 @@ public Optional<Long> getDeliveryTimestamp() {
         return Optional.ofNullable(deliveryTimestamp);
     }
 
+    /**
+     * @see Message#getPriority()
+     */
+    @Override
+    public Optional<Integer> getPriority() {
+        return Optional.ofNullable(priority);
+    }
+
     /**
      * @see Message#getMessageGroup()
      */
@@ -152,7 +164,9 @@ public String toString() {
             .add("topic", topic)
             .add("tag", tag)
             .add("messageGroup", messageGroup)
+            .add("liteTopic", liteTopic)
             .add("deliveryTimestamp", deliveryTimestamp)
+            .add("priority", priority)
             .add("keys", keys)
             .add("properties", properties)
             .toString();
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageType.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageType.java
@@ -22,6 +22,7 @@ public enum MessageType {
     FIFO,
     LITE,
     DELAY,
+    PRIORITY,
     TRANSACTION;
 
     public static MessageType fromProtobuf(apache.rocketmq.v2.MessageType messageType) {
@@ -36,6 +37,8 @@ public static MessageType fromProtobuf(apache.rocketmq.v2.MessageType messageTyp
                 return MessageType.DELAY;
             case TRANSACTION:
                 return MessageType.TRANSACTION;
+            case PRIORITY:
+                return MessageType.PRIORITY;
             case MESSAGE_TYPE_UNSPECIFIED:
             default:
                 throw new IllegalArgumentException("Message type is not specified");
@@ -54,6 +57,8 @@ public static apache.rocketmq.v2.MessageType toProtobuf(MessageType messageType)
                 return apache.rocketmq.v2.MessageType.DELAY;
             case TRANSACTION:
                 return apache.rocketmq.v2.MessageType.TRANSACTION;
+            case PRIORITY:
+                return apache.rocketmq.v2.MessageType.PRIORITY;
             default:
                 return apache.rocketmq.v2.MessageType.MESSAGE_TYPE_UNSPECIFIED;
         }
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageViewImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/MessageViewImpl.java
@@ -53,6 +53,7 @@ public class MessageViewImpl implements MessageView {
     private final String messageGroup;
     private final String liteTopic;
     private final Long deliveryTimestamp;
+    private final Integer priority;
     private final Collection<String> keys;
     private final Map<String, String> properties;
     private final String bornHost;
@@ -68,7 +69,7 @@ public class MessageViewImpl implements MessageView {
 
     public MessageViewImpl(MessageId messageId, String topic, byte[] body, String tag,
         String messageGroup, String liteTopic,
-        Long deliveryTimestamp, Collection<String> keys, Map<String, String> properties,
+        Long deliveryTimestamp, Integer priority, Collection<String> keys, Map<String, String> properties,
         String bornHost, long bornTimestamp, int deliveryAttempt, MessageQueueImpl messageQueue,
         String receiptHandle, long offset, boolean corrupted,
         Long transportDeliveryTimestamp) {
@@ -79,6 +80,7 @@ public MessageViewImpl(MessageId messageId, String topic, byte[] body, String ta
         this.messageGroup = messageGroup;
         this.liteTopic = liteTopic;
         this.deliveryTimestamp = deliveryTimestamp;
+        this.priority = priority;
         this.keys = checkNotNull(keys, "keys should not be null");
         this.properties = checkNotNull(properties, "properties should not be null");
         this.bornHost = checkNotNull(bornHost, "bornHost should not be null");
@@ -165,6 +167,14 @@ public Optional<Long> getDeliveryTimestamp() {
         return Optional.ofNullable(deliveryTimestamp);
     }
 
+    /**
+     * @see MessageView#getPriority()
+     */
+    @Override
+    public Optional<Integer> getPriority() {
+        return Optional.ofNullable(priority);
+    }
+
     /**
      * @see MessageView#getBornHost()
      */
@@ -303,14 +313,15 @@ public static MessageViewImpl fromProtobuf(Message message, MessageQueueImpl mq,
         String liteTopic = systemProperties.hasLiteTopic() ? systemProperties.getLiteTopic() : null;
         Long deliveryTimestamp = systemProperties.hasDeliveryTimestamp() ?
             Timestamps.toMillis(systemProperties.getDeliveryTimestamp()) : null;
+        Integer priority = systemProperties.hasPriority() ? systemProperties.getPriority() : null;
         final ProtocolStringList keys = systemProperties.getKeysList();
         final String bornHost = systemProperties.getBornHost();
         final long bornTimestamp = Timestamps.toMillis(systemProperties.getBornTimestamp());
         final int deliveryAttempt = systemProperties.getDeliveryAttempt();
         final long offset = systemProperties.getQueueOffset();
         final Map<String, String> properties = message.getUserPropertiesMap();
         final String receiptHandle = systemProperties.getReceiptHandle();
-        return new MessageViewImpl(messageId, topic, body, tag, messageGroup, liteTopic, deliveryTimestamp,
+        return new MessageViewImpl(messageId, topic, body, tag, messageGroup, liteTopic, deliveryTimestamp, priority,
             keys, properties, bornHost, bornTimestamp, deliveryAttempt,
             mq, receiptHandle, offset, corrupted, transportDeliveryTimestamp);
     }
@@ -329,6 +340,7 @@ public String toString() {
             .add("messageGroup", messageGroup)
             .add("liteTopic", liteTopic)
             .add("deliveryTimestamp", deliveryTimestamp)
+            .add("priority", priority)
             .add("properties", properties)
             .toString();
     }
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/message/PublishingMessageImpl.java
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/message/GeneralMessageImplTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/message/GeneralMessageImplTest.java
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/message/MessageImplTest.java b/java/client/src/test/java/org/apache/rocketmq/client/java/message/MessageImplTest.java
diff --git a/java/client/src/test/java/org/apache/rocketmq/client/java/tool/TestBase.java b/java/client/src/test/java/org/apache/rocketmq/client/java/tool/TestBase.java
diff --git a/java/pom.xml b/java/pom.xml
