rename REPLY_TO to REPLY_TO_CLIENT

broker/src/main/java/org/apache/rocketmq/broker/processor/ReplyMessageProcessor.java

@@ -173,7 +173,7 @@ public class ReplyMessageProcessor extends AbstractSendMessageProcessor implemen
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PUSH_REPLY_MESSAGE_TO_CLIENT, replyMessageRequestHeader);
         request.setBody(msg.getBody());
 
-        String senderId = msg.getProperties().get(MessageConst.PROPERTY_MESSAGE_REPLY_TO);
+        String senderId = msg.getProperties().get(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT);
         PushReplyResult pushReplyResult = new PushReplyResult(false);
 
         if (senderId != null) {
@@ -207,9 +207,9 @@ public class ReplyMessageProcessor extends AbstractSendMessageProcessor implemen
                 log.warn(pushReplyResult.getRemark());
             }
         } else {
-            log.warn("REPLY_TO is null, can not reply message");
+            log.warn(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT + " is null, can not reply message");
             pushReplyResult.setPushOk(false);
-            pushReplyResult.setRemark("reply message properties[" + MessageConst.PROPERTY_MESSAGE_REPLY_TO + "] is null");
+            pushReplyResult.setRemark("reply message properties[" + MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT + "] is null");
         }
         return pushReplyResult;
     }

broker/src/test/java/org/apache/rocketmq/broker/processor/ReplyMessageProcessorTest.java

@@ -30,7 +30,6 @@ import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageDecoder;
-import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.protocol.RequestCode;
 import org.apache.rocketmq.common.protocol.ResponseCode;
 import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
@@ -58,7 +57,6 @@ import org.mockito.junit.MockitoJUnitRunner;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -126,7 +124,7 @@ public class ReplyMessageProcessorTest {
         requestHeader.setFlag(124);
         requestHeader.setReconsumeTimes(0);
         Map<String, String> map = new HashMap<String, String>();
-        map.put(MessageConst.PROPERTY_MESSAGE_REPLY_TO, "127.0.0.1");
+        map.put(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT, "127.0.0.1");
         requestHeader.setProperties(MessageDecoder.messageProperties2String(map));
         return requestHeader;
     }

client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java

@@ -1548,7 +1548,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
         String correlationId = CorrelationIdUtil.createCorrelationId();
         String requestClientId = this.getmQClientFactory().getClientId();
         MessageAccessor.putProperty(msg, MessageConst.PROPERTY_CORRELATION_ID, correlationId);
-        MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MESSAGE_REPLY_TO, requestClientId);
+        MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT, requestClientId);
         MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MESSAGE_TTL, String.valueOf(timeout));
 
         boolean hasRouteData = this.getmQClientFactory().getTopicRouteTable().containsKey(msg.getTopic());

client/src/main/java/org/apache/rocketmq/client/utils/MessageUtil.java

@@ -29,7 +29,7 @@ public class MessageUtil {
         if (requestMessage != null) {
             Message replyMessage = new Message();
             String cluster = requestMessage.getProperty(MessageConst.PROPERTY_CLUSTER);
-            String replyTo = requestMessage.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO);
+            String replyTo = requestMessage.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT);
             String correlationId = requestMessage.getProperty(MessageConst.PROPERTY_CORRELATION_ID);
             String ttl = requestMessage.getProperty(MessageConst.PROPERTY_MESSAGE_TTL);
             replyMessage.setBody(body);
@@ -38,7 +38,7 @@ public class MessageUtil {
                 replyMessage.setTopic(replyTopic);
                 MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_TYPE, MixAll.REPLY_MESSAGE_FLAG);
                 MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_CORRELATION_ID, correlationId);
-                MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_REPLY_TO, replyTo);
+                MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT, replyTo);
                 MessageAccessor.putProperty(replyMessage, MessageConst.PROPERTY_MESSAGE_TTL, ttl);
 
                 return replyMessage;
@@ -48,4 +48,8 @@ public class MessageUtil {
         }
         throw new MQClientException(ClientErrorCode.CREATE_REPLY_MESSAGE_EXCEPTION, "create reply message fail, requestMessage cannot be null.");
     }
+
+    public static String getReplyToClient(final Message msg) {
+        return msg.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT);
+    }
 }

client/src/test/java/org/apache/rocketmq/client/utils/MessageUtilsTest.java

@@ -35,7 +35,7 @@ public class MessageUtilsTest {
     public void testCreateReplyMessage() throws MQClientException {
         Message msg = MessageUtil.createReplyMessage(createReplyMessage("clusterName"), new byte[] {'a'});
         assertThat(msg.getTopic()).isEqualTo("clusterName" + "_" + MixAll.REPLY_TOPIC_POSTFIX);
-        assertThat(msg.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO)).isEqualTo("127.0.0.1");
+        assertThat(msg.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT)).isEqualTo("127.0.0.1");
         assertThat(msg.getProperty(MessageConst.PROPERTY_MESSAGE_TTL)).isEqualTo("3000");
     }
 
@@ -59,10 +59,18 @@ public class MessageUtilsTest {
         }
     }
 
+    @Test
+    public void testGetReplyToClient() throws MQClientException {
+        Message msg = createReplyMessage("clusterName");
+        String replyToClient = MessageUtil.getReplyToClient(msg);
+        assertThat(replyToClient).isNotNull();
+        assertThat(replyToClient).isEqualTo("127.0.0.1");
+    }
+
     private Message createReplyMessage(String clusterName) {
         Message requestMessage = new Message();
         Map map = new HashMap<String, String>();
-        map.put(MessageConst.PROPERTY_MESSAGE_REPLY_TO, "127.0.0.1");
+        map.put(MessageConst.PROPERTY_MESSAGE_REPLY_TO_CLIENT, "127.0.0.1");
         map.put(MessageConst.PROPERTY_CLUSTER, clusterName);
         map.put(MessageConst.PROPERTY_MESSAGE_TTL, "3000");
         MessageAccessor.setProperties(requestMessage, map);

common/src/main/java/org/apache/rocketmq/common/message/MessageConst.java

@@ -46,7 +46,7 @@ public class MessageConst {
     public static final String PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS = "CHECK_IMMUNITY_TIME_IN_SECONDS";
     public static final String PROPERTY_INSTANCE_ID = "INSTANCE_ID";
     public static final String PROPERTY_CORRELATION_ID = "CORRELATION_ID";
-    public static final String PROPERTY_MESSAGE_REPLY_TO = "REPLY_TO";
+    public static final String PROPERTY_MESSAGE_REPLY_TO_CLIENT = "REPLY_TO_CLIENT";
     public static final String PROPERTY_MESSAGE_TTL = "TTL";
     public static final String PROPERTY_REPLY_MESSAGE_ARRIVE_TIME = "ARRIVE_TIME";
     public static final String PROPERTY_PUSH_REPLY_TIME = "PUSH_REPLY_TIME";
@@ -82,7 +82,7 @@ public class MessageConst {
         STRING_HASH_SET.add(PROPERTY_CONSUME_START_TIMESTAMP);
         STRING_HASH_SET.add(PROPERTY_INSTANCE_ID);
         STRING_HASH_SET.add(PROPERTY_CORRELATION_ID);
-        STRING_HASH_SET.add(PROPERTY_MESSAGE_REPLY_TO);
+        STRING_HASH_SET.add(PROPERTY_MESSAGE_REPLY_TO_CLIENT);
         STRING_HASH_SET.add(PROPERTY_MESSAGE_TTL);
         STRING_HASH_SET.add(PROPERTY_REPLY_MESSAGE_ARRIVE_TIME);
         STRING_HASH_SET.add(PROPERTY_PUSH_REPLY_TIME);

example/src/main/java/org/apache/rocketmq/example/rpc/ResponseConsumer.java

@@ -57,7 +57,7 @@ public class ResponseConsumer {
                 for (MessageExt msg : msgs) {
                     try {
                         System.out.printf("handle message: %s", msg.toString());
-                        String replyTo = msg.getProperty(MessageConst.PROPERTY_MESSAGE_REPLY_TO);
+                        String replyTo = MessageUtil.getReplyToClient(msg);
                         byte[] replyContent = "reply message contents.".getBytes();
                         // create reply message with given util, do not create reply message by yourself
                         Message replyMessage = MessageUtil.createReplyMessage(msg, replyContent);