You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2018/07/14 03:12:38 UTC
[GitHub] vongosling closed pull request #223: [ROCKETMQ-365] Add consume
times stat and fix client log
vongosling closed pull request #223: [ROCKETMQ-365] Add consume times stat and fix client log
URL: https://github.com/apache/rocketmq/pull/223
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java b/client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java
index c1524e107..083fca45f 100644
--- a/client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java
+++ b/client/src/main/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStore.java
@@ -125,13 +125,8 @@ public void persistAll(Set<MessageQueue> mqs) {
if (mqs.contains(mq)) {
try {
this.updateConsumeOffsetToBroker(mq, offset.get());
- log.info("[persistAll] Group: {} ClientId: {} updateConsumeOffsetToBroker {} {}",
- this.groupName,
- this.mQClientFactory.getClientId(),
- mq,
- offset.get());
} catch (Exception e) {
- log.error("updateConsumeOffsetToBroker exception, " + mq.toString(), e);
+ log.error("updateConsumeOffsetToBroker exception, " + mq.toString() + " offset:" + offset.get(), e);
}
} else {
unusedMQ.add(mq);
@@ -154,13 +149,8 @@ public void persist(MessageQueue mq) {
if (offset != null) {
try {
this.updateConsumeOffsetToBroker(mq, offset.get());
- log.info("[persist] Group: {} ClientId: {} updateConsumeOffsetToBroker {} {}",
- this.groupName,
- this.mQClientFactory.getClientId(),
- mq,
- offset.get());
} catch (Exception e) {
- log.error("updateConsumeOffsetToBroker exception, " + mq.toString(), e);
+ log.error("updateConsumeOffsetToBroker exception, " + mq.toString() + " offset:" + offset.get(), e);
}
}
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
index 58b985f10..6773a1325 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
@@ -285,6 +285,15 @@ public void processConsumeResult(
break;
}
+ for (int i = 0; i < consumeRequest.getMsgs().size(); i++) {
+ MessageExt messageExt = consumeRequest.getMsgs().get(i);
+ if (i > ackIndex) {
+ this.getConsumerStatsManager().incConsumeTimeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), messageExt.getReconsumeTimes());
+ } else {
+ this.getConsumerStatsManager().incConsumeTimeOkTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), messageExt.getReconsumeTimes());
+ }
+ }
+
switch (this.defaultMQPushConsumer.getMessageModel()) {
case BROADCASTING:
for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
index d4a9953e7..f103811b9 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java
@@ -270,8 +270,14 @@ public boolean processConsumeResult(
case SUCCESS:
commitOffset = consumeRequest.getProcessQueue().commit();
this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
+ for (MessageExt msg : msgs) {
+ this.getConsumerStatsManager().incConsumeTimeOkTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msg.getReconsumeTimes());
+ }
break;
case SUSPEND_CURRENT_QUEUE_A_MOMENT:
+ for (MessageExt msg : msgs) {
+ this.getConsumerStatsManager().incConsumeTimeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msg.getReconsumeTimes());
+ }
this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
if (checkReconsumeTimes(msgs)) {
consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);
@@ -291,6 +297,9 @@ public boolean processConsumeResult(
switch (status) {
case SUCCESS:
this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
+ for (MessageExt msg : msgs) {
+ this.getConsumerStatsManager().incConsumeTimeOkTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msg.getReconsumeTimes());
+ }
break;
case COMMIT:
commitOffset = consumeRequest.getProcessQueue().commit();
@@ -304,6 +313,10 @@ public boolean processConsumeResult(
continueConsume = false;
break;
case SUSPEND_CURRENT_QUEUE_A_MOMENT:
+ for (MessageExt msg : msgs) {
+ this.getConsumerStatsManager().incConsumeTimeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msg.getReconsumeTimes());
+ }
+
this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
if (checkReconsumeTimes(msgs)) {
consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);
diff --git a/client/src/main/java/org/apache/rocketmq/client/stat/ConsumerStatsManager.java b/client/src/main/java/org/apache/rocketmq/client/stat/ConsumerStatsManager.java
index cf347b4dd..d2ffc1d65 100644
--- a/client/src/main/java/org/apache/rocketmq/client/stat/ConsumerStatsManager.java
+++ b/client/src/main/java/org/apache/rocketmq/client/stat/ConsumerStatsManager.java
@@ -28,6 +28,8 @@
public class ConsumerStatsManager {
private static final InternalLogger log = ClientLogger.getLog();
+ private static final String TOPIC_AND_GROUP_CONSUME_TIME_OK_TPS = "CONSUME_TIME_OK_TPS";
+ private static final String TOPIC_AND_GROUP_CONSUME_TIME_FAILED_TPS = "CONSUME_TIME_FAILED_TPS";
private static final String TOPIC_AND_GROUP_CONSUME_OK_TPS = "CONSUME_OK_TPS";
private static final String TOPIC_AND_GROUP_CONSUME_FAILED_TPS = "CONSUME_FAILED_TPS";
private static final String TOPIC_AND_GROUP_CONSUME_RT = "CONSUME_RT";
@@ -40,6 +42,9 @@
private final StatsItemSet topicAndGroupPullTPS;
private final StatsItemSet topicAndGroupPullRT;
+ private final StatsItemSet topicAndGroupConsumeTimeOkTPS;
+ private final StatsItemSet topicAndGroupConsumeTimeFailedTPS;
+
public ConsumerStatsManager(final ScheduledExecutorService scheduledExecutorService) {
this.topicAndGroupConsumeOKTPS =
new StatsItemSet(TOPIC_AND_GROUP_CONSUME_OK_TPS, scheduledExecutorService, log);
@@ -53,6 +58,10 @@ public ConsumerStatsManager(final ScheduledExecutorService scheduledExecutorServ
this.topicAndGroupPullTPS = new StatsItemSet(TOPIC_AND_GROUP_PULL_TPS, scheduledExecutorService, log);
this.topicAndGroupPullRT = new StatsItemSet(TOPIC_AND_GROUP_PULL_RT, scheduledExecutorService, log);
+
+ this.topicAndGroupConsumeTimeOkTPS = new StatsItemSet(TOPIC_AND_GROUP_CONSUME_TIME_OK_TPS, scheduledExecutorService, log);
+
+ this.topicAndGroupConsumeTimeFailedTPS = new StatsItemSet(TOPIC_AND_GROUP_CONSUME_TIME_FAILED_TPS, scheduledExecutorService, log);
}
public void start() {
@@ -69,6 +78,14 @@ public void incPullTPS(final String group, final String topic, final long msgs)
this.topicAndGroupPullTPS.addValue(topic + "@" + group, (int) msgs, 1);
}
+ public void incConsumeTimeOkTPS(final String group, final String topic, final int consumeTimes) {
+ topicAndGroupConsumeTimeOkTPS.addValue(topic + "@" + group + "@" + consumeTimes, 1, 1);
+ }
+
+ public void incConsumeTimeFailedTPS(final String group, final String topic, final int consumeTimes) {
+ topicAndGroupConsumeTimeFailedTPS.addValue(topic + "@" + group + "@" + consumeTimes, 1, 1);
+ }
+
public void incConsumeRT(final String group, final String topic, final long rt) {
this.topicAndGroupConsumeRT.addValue(topic + "@" + group, (int) rt, 1);
}
diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStoreTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStoreTest.java
index 64d64f29a..b91052147 100644
--- a/client/src/test/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStoreTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/consumer/store/RemoteBrokerOffsetStoreTest.java
@@ -18,7 +18,6 @@
import java.util.Collections;
import java.util.HashSet;
-import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.impl.FindBrokerResult;
import org.apache.rocketmq.client.impl.MQClientAPIImpl;
@@ -56,8 +55,8 @@
@Before
public void init() {
System.setProperty("rocketmq.client.localOffsetStoreDir", System.getProperty("java.io.tmpdir") + ".rocketmq_offsets");
- String clientId = new ClientConfig().buildMQClientId() + "#TestNamespace" + System.currentTimeMillis();
- when(mQClientFactory.getClientId()).thenReturn(clientId);
+// String clientId = new ClientConfig().buildMQClientId() + "#TestNamespace" + System.currentTimeMillis();
+// when(mQClientFactory.getClientId()).thenReturn(clientId);
when(mQClientFactory.findBrokerAddressInAdmin(brokerName)).thenReturn(new FindBrokerResult("127.0.0.1", false));
when(mQClientFactory.getMQClientAPIImpl()).thenReturn(mqClientAPI);
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services