You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by du...@apache.org on 2019/02/23 08:28:09 UTC
[rocketmq] branch snode updated: Fix the issue of storesize and
bornhost inaccurate in push message
This is an automated email from the ASF dual-hosted git repository.
duhengforever pushed a commit to branch snode
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/snode by this push:
new ff6b687 Fix the issue of storesize and bornhost inaccurate in push message
new 41d5adc Merge branch 'snode' of github.com:apache/rocketmq into snode
ff6b687 is described below
commit ff6b687151f69f7791861bc2f3f9262bffd0a8f2
Author: duhenglucky <du...@gmail.com>
AuthorDate: Sat Feb 23 16:26:37 2019 +0800
Fix the issue of storesize and bornhost inaccurate in push message
---
.../rocketmq/broker/processor/SendMessageProcessor.java | 13 +++----------
.../rocketmq/common/protocol/body/SnodeClusterInfo.java | 4 ++--
.../common/protocol/header/SendMessageRequestHeader.java | 13 ++++++-------
.../common/protocol/header/SendMessageRequestHeaderV2.java | 13 ++++++-------
.../org/apache/rocketmq/remoting/common/RemotingUtil.java | 13 -------------
.../apache/rocketmq/remoting/protocol/RemotingCommand.java | 1 -
.../main/java/org/apache/rocketmq/snode/SnodeStartup.java | 5 ++++-
.../apache/rocketmq/snode/offset/ConsumerOffsetManager.java | 13 -------------
.../rocketmq/snode/processor/SendMessageProcessor.java | 5 +++--
.../apache/rocketmq/snode/service/impl/PushServiceImpl.java | 5 +++--
.../src/main/java/org/apache/rocketmq/store/CommitLog.java | 4 ++--
11 files changed, 29 insertions(+), 60 deletions(-)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
index 1408cce..e5bd8e7 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
@@ -78,13 +78,11 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
if (requestHeader == null) {
return null;
}
-
mqtraceContext = buildMsgContext(ctx, requestHeader);
this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
RemotingCommand response;
-// String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
- SocketAddress bornHost = ctx.channel().remoteAddress();
+ SocketAddress bornHost = RemotingHelper.string2SocketAddress(requestHeader.getBornHost());
if (requestHeader.isBatch()) {
response = this.sendBatchMessage(bornHost, request, mqtraceContext, requestHeader);
@@ -350,12 +348,10 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
MessageAccessor.setProperties(msgInner, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
msgInner.setPropertiesString(requestHeader.getProperties());
msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
- msgInner.setBornHost(requestHeader.getBornHost());
+ msgInner.setBornHost(remoteAddress);
msgInner.setStoreHost(this.getStoreHost());
msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
msgInner.setBornHost(remoteAddress);
-// ByteBuffer hostHolder = ByteBuffer.allocate(8);
-// String bornHost = msgInner.getStoreHostBytes(hostHolder).toString();
PutMessageResult putMessageResult = null;
Map<String, String> oriProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
@@ -548,14 +544,11 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
MessageAccessor.setProperties(messageExtBatch, MessageDecoder.string2messageProperties(requestHeader.getProperties()));
messageExtBatch.setBody(request.getBody());
messageExtBatch.setBornTimestamp(requestHeader.getBornTimestamp());
- messageExtBatch.setBornHost(requestHeader.getBornHost());
+ messageExtBatch.setBornHost(remoteAddress);
messageExtBatch.setStoreHost(this.getStoreHost());
messageExtBatch.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
-// ByteBuffer hostHolder = ByteBuffer.allocate(8);
-// String storeHost = messageExtBatch.getStoreHostBytes(hostHolder).toString();
-
PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessages(messageExtBatch);
return handlePutMessageResult(putMessageResult, response, request, messageExtBatch, responseHeader, sendMessageContext, queueIdInt, storeHost);
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/body/SnodeClusterInfo.java b/common/src/main/java/org/apache/rocketmq/common/protocol/body/SnodeClusterInfo.java
index 2d42bdf..55999af 100644
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/body/SnodeClusterInfo.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/body/SnodeClusterInfo.java
@@ -1,4 +1,4 @@
-package org.apache.rocketmq.common.protocol.body;/*
+/*
* 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.
@@ -14,7 +14,7 @@ package org.apache.rocketmq.common.protocol.body;/*
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
+package org.apache.rocketmq.common.protocol.body;
import java.util.HashMap;
import java.util.Set;
import org.apache.rocketmq.common.protocol.route.SnodeData;
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/SendMessageRequestHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/SendMessageRequestHeader.java
index 84ab10d..2577bf5 100644
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/header/SendMessageRequestHeader.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/SendMessageRequestHeader.java
@@ -20,7 +20,6 @@
*/
package org.apache.rocketmq.common.protocol.header;
-import java.net.SocketAddress;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.annotation.CFNotNull;
import org.apache.rocketmq.remoting.annotation.CFNullable;
@@ -56,9 +55,9 @@ public class SendMessageRequestHeader implements CommandCustomHeader {
private String enodeName;
- private SocketAddress bornHost;
+ private String bornHost;
- private SocketAddress snodeHost;
+ private String snodeHost;
@Override
public void checkFields() throws RemotingCommandException {
@@ -176,19 +175,19 @@ public class SendMessageRequestHeader implements CommandCustomHeader {
this.enodeName = enodeName;
}
- public SocketAddress getBornHost() {
+ public String getBornHost() {
return bornHost;
}
- public void setBornHost(SocketAddress bornHost) {
+ public void setBornHost(String bornHost) {
this.bornHost = bornHost;
}
- public SocketAddress getSnodeHost() {
+ public String getSnodeHost() {
return snodeHost;
}
- public void setSnodeHost(SocketAddress snodeHost) {
+ public void setSnodeHost(String snodeHost) {
this.snodeHost = snodeHost;
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/SendMessageRequestHeaderV2.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/SendMessageRequestHeaderV2.java
index 3260c25..048d201 100644
--- a/common/src/main/java/org/apache/rocketmq/common/protocol/header/SendMessageRequestHeaderV2.java
+++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/SendMessageRequestHeaderV2.java
@@ -17,7 +17,6 @@
package org.apache.rocketmq.common.protocol.header;
-import java.net.SocketAddress;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.annotation.CFNotNull;
import org.apache.rocketmq.remoting.annotation.CFNullable;
@@ -57,9 +56,9 @@ public class SendMessageRequestHeaderV2 implements CommandCustomHeader {
private String n; //enode name
- private SocketAddress o; //born host
+ private String o; //born host
- private SocketAddress p; //snode host
+ private String p; //snode host
public static SendMessageRequestHeader createSendMessageRequestHeaderV1(final SendMessageRequestHeaderV2 v2) {
SendMessageRequestHeader v1 = new SendMessageRequestHeader();
@@ -219,19 +218,19 @@ public class SendMessageRequestHeaderV2 implements CommandCustomHeader {
this.n = n;
}
- public SocketAddress getO() {
+ public String getO() {
return o;
}
- public void setO(SocketAddress o) {
+ public void setO(String o) {
this.o = o;
}
- public SocketAddress getP() {
+ public String getP() {
return p;
}
- public void setP(SocketAddress p) {
+ public void setP(String p) {
this.p = p;
}
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java
index a118c40..4e592cb 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java
@@ -153,19 +153,6 @@ public class RemotingUtil {
return isa;
}
- public static SocketAddress string2SocketAddressWithIp(final String address) {
- String[] s = address.split(":");
- try {
- String ip = s[0].substring(1);
- InetAddress inetAddress = InetAddress.getByName(ip);
- InetSocketAddress isa = new InetSocketAddress(inetAddress, Integer.parseInt(s[1]));
- return isa;
- } catch (Exception e) {
- log.error("Failed to obtain address", e);
- }
- return null;
- }
-
public static String socketAddress2String(final SocketAddress addr) {
StringBuilder sb = new StringBuilder();
InetSocketAddress inetSocketAddress = (InetSocketAddress) addr;
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RemotingCommand.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RemotingCommand.java
index d392e79..e823a69 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RemotingCommand.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RemotingCommand.java
@@ -49,7 +49,6 @@ public class RemotingCommand {
private static final int RPC_ONEWAY = 1;
-
private static volatile int configVersion = -1;
private static AtomicInteger requestId = new AtomicInteger(0);
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java b/snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java
index 23ea68a..7079771 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java
@@ -29,6 +29,7 @@ import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.PosixParser;
+import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerStartup;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.SnodeConfig;
@@ -54,7 +55,9 @@ public class SnodeStartup {
public static void main(String[] args) throws IOException, JoranException {
SnodeConfig snodeConfig = loadConfig(args);
if (snodeConfig.isEmbeddedModeEnable()) {
- BrokerStartup.start(BrokerStartup.createBrokerController(args));
+ BrokerController brokerController = BrokerStartup.createBrokerController(args);
+ BrokerStartup.start(brokerController);
+ snodeConfig.setSnodeName(brokerController.getBrokerConfig().getBrokerName());
}
SnodeController snodeController = createSnodeController(snodeConfig);
startup(snodeController);
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/offset/ConsumerOffsetManager.java b/snode/src/main/java/org/apache/rocketmq/snode/offset/ConsumerOffsetManager.java
index ba2a800..2cb6e69 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/offset/ConsumerOffsetManager.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/offset/ConsumerOffsetManager.java
@@ -75,19 +75,6 @@ public class ConsumerOffsetManager {
}
}
-// private long parserOffset(final RemotingChannel remotingChannel, final String enodeName, final String group,
-// final String topic, final int queueId) {
-// try {
-// RemotingCommand remotingCommand = queryOffset(remotingChannel, enodeName, group, topic, queueId);
-// QueryConsumerOffsetResponseHeader responseHeader =
-// (QueryConsumerOffsetResponseHeader) remotingCommand.decodeCommandCustomHeader(QueryConsumerOffsetResponseHeader.class);
-// return responseHeader.getOffset();
-// } catch (Exception ex) {
-// log.error("Load offset from broker error", ex);
-// }
-// return -1;
-// }
-
public long queryCacheOffset(final String enodeName, final String group, final String topic, final int queueId) {
String key = buildKey(enodeName, topic, group);
ConcurrentMap<Integer, CacheOffset> map = this.offsetTable.get(key);
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/processor/SendMessageProcessor.java b/snode/src/main/java/org/apache/rocketmq/snode/processor/SendMessageProcessor.java
index 019794e..dda5774 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/processor/SendMessageProcessor.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/processor/SendMessageProcessor.java
@@ -28,6 +28,7 @@ import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.RequestProcessor;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.interceptor.ExceptionContext;
import org.apache.rocketmq.remoting.interceptor.RequestContext;
@@ -74,8 +75,9 @@ public class SendMessageProcessor implements RequestProcessor {
request.getCode() == RequestCode.SEND_BATCH_MESSAGE) {
sendMessageRequestHeaderV2 = (SendMessageRequestHeaderV2) request.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
enodeName = sendMessageRequestHeaderV2.getN();
- sendMessageRequestHeaderV2.setP(remotingChannel.localAddress());
stringBuffer.append(sendMessageRequestHeaderV2.getB());
+ request.getExtFields().putIfAbsent("o", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
+ request.getExtFields().putIfAbsent("p", RemotingHelper.parseChannelRemoteAddr(remotingChannel.localAddress()));
} else {
isSendBack = true;
consumerSendMsgBackRequestHeader = (ConsumerSendMsgBackRequestHeader) request.decodeCommandCustomHeader(ConsumerSendMsgBackRequestHeader.class);
@@ -85,7 +87,6 @@ public class SendMessageProcessor implements RequestProcessor {
CompletableFuture<RemotingCommand> responseFuture = snodeController.getEnodeService().sendMessage(remotingChannel, enodeName, request);
- sendMessageRequestHeaderV2.setO(remotingChannel.remoteAddress());
final byte[] message = request.getBody();
final boolean needPush = !isSendBack;
final SendMessageRequestHeader sendMessageRequestHeader =
diff --git a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
index 9f4b016..24a5a9c 100644
--- a/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
+++ b/snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
@@ -87,10 +87,11 @@ public class PushServiceImpl implements PushService {
messageExt.setReconsumeTimes(sendMessageRequestHeader.getReconsumeTimes());
messageExt.setCommitLogOffset(sendMessageResponseHeader.getCommitLogOffset());
messageExt.setBornTimestamp(sendMessageRequestHeader.getBornTimestamp());
- messageExt.setBornHost(sendMessageRequestHeader.getBornHost());
- messageExt.setStoreHost(RemotingUtil.string2SocketAddressWithIp(sendMessageResponseHeader.getStoreHost()));
+ messageExt.setBornHost(RemotingUtil.string2SocketAddress(sendMessageRequestHeader.getBornHost()));
+ messageExt.setStoreHost(RemotingUtil.string2SocketAddress(sendMessageResponseHeader.getStoreHost()));
messageExt.setStoreTimestamp(sendMessageResponseHeader.getStoreTimestamp());
messageExt.setWaitStoreMsgOK(false);
+ messageExt.setStoreSize(sendMessageResponseHeader.getStoreSize());
messageExt.setSysFlag(sendMessageRequestHeader.getSysFlag());
messageExt.setFlag(sendMessageRequestHeader.getFlag());
messageExt.setBody(message);
diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
index cbcc1a7..b02d36e 100644
--- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -26,14 +26,14 @@ import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.message.MessageAccessor;
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.message.MessageExtBatch;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.ha.HAService;