diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/ReplyMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/ReplyMessageProcessor.java index b69d635267..3c4a8f8f84 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/ReplyMessageProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/ReplyMessageProcessor.java @@ -45,6 +45,7 @@ import org.apache.rocketmq.store.MessageExtBrokerInner; import org.apache.rocketmq.store.PutMessageResult; import org.apache.rocketmq.store.stats.BrokerStatsManager; +import java.net.InetSocketAddress; import java.util.concurrent.ThreadLocalRandom; public class ReplyMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor { @@ -157,8 +158,10 @@ public class ReplyMessageProcessor extends AbstractSendMessageProcessor implemen final SendMessageRequestHeader requestHeader, final Message msg) { ReplyMessageRequestHeader replyMessageRequestHeader = new ReplyMessageRequestHeader(); - replyMessageRequestHeader.setBornHost(ctx.channel().remoteAddress().toString()); - replyMessageRequestHeader.setStoreHost(this.getStoreHost().toString()); + InetSocketAddress bornAddress = (InetSocketAddress)(ctx.channel().remoteAddress()); + replyMessageRequestHeader.setBornHost(bornAddress.getAddress().getHostAddress() + ":" + bornAddress.getPort()); + InetSocketAddress storeAddress = (InetSocketAddress)(this.getStoreHost()); + replyMessageRequestHeader.setStoreHost(storeAddress.getAddress().getHostAddress() + ":" + storeAddress.getPort()); replyMessageRequestHeader.setStoreTimestamp(System.currentTimeMillis()); replyMessageRequestHeader.setProducerGroup(requestHeader.getProducerGroup()); replyMessageRequestHeader.setTopic(requestHeader.getTopic());