Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[ISSUE #7381] Fix the problem of inaccurate timer message metric #7382

Merged
merged 4 commits into from
Sep 23, 2023
Merged

[ISSUE #7381] Fix the problem of inaccurate timer message metric #7382

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ee9f66e
correct the timerMetrics' result.
GenerousMan Sep 21, 2023
8e5f8cf
for further extension.
GenerousMan Sep 22, 2023
a2fee37
checkstyle.
GenerousMan Sep 22, 2023
2ae68d7
use toLong.
GenerousMan Sep 22, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.TopicFilterType;
Expand Down Expand Up @@ -599,7 +600,12 @@ public void addMetric(MessageExt msg, int value) {
if (null == msg || null == msg.getProperty(MessageConst.PROPERTY_REAL_TOPIC)) {
return;
}
timerMetrics.addAndGet(msg.getProperty(MessageConst.PROPERTY_REAL_TOPIC), value);
if (msg.getProperty(TIMER_ENQUEUE_MS) != null
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

NumberUtils.toLong() might be more clear

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

no need of msg.getProperty(TIMER_ENQUEUE_MS) != null if using NumberUtils.toLong, you can check this out

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Under normal situation, this indicator is still null when exiting the queue, so this judgment is necessary.
When leaving the queue, the 'enququ_time' attribute in the message has not been retrieved from the timerRequest yet. When leaving the queue, this attribute will not be null unless it is determined that it has never entered the time wheel.

&& NumberUtils.toLong(msg.getProperty(TIMER_ENQUEUE_MS)) == Long.MAX_VALUE) {
return;
}
// pass msg into addAndGet, for further more judgement extension.
timerMetrics.addAndGet(msg, value);
} catch (Throwable t) {
if (frequency.incrementAndGet() % 1000 == 0) {
LOGGER.error("error in adding metric", t);
Expand Down Expand Up @@ -1323,6 +1329,7 @@ protected void putMessageToTimerWheel(TimerRequest req) {
perfCounterTicks.startTick(ENQUEUE_PUT);
DefaultStoreMetricsManager.incTimerEnqueueCount(getRealTopic(req.getMsg()));
if (shouldRunningDequeue && req.getDelayTime() < currWriteTimeMs) {
req.setEnqueueTime(Long.MAX_VALUE);
dequeuePutQueue.put(req);
} else {
boolean doEnqueueRes = doEnqueue(
Expand Down Expand Up @@ -1452,9 +1459,14 @@ public void run() {
}
try {
perfCounterTicks.startTick(DEQUEUE_PUT);
DefaultStoreMetricsManager.incTimerDequeueCount(getRealTopic(tr.getMsg()));
addMetric(tr.getMsg(), -1);
MessageExtBrokerInner msg = convert(tr.getMsg(), tr.getEnqueueTime(), needRoll(tr.getMagic()));
MessageExt msgExt = tr.getMsg();
DefaultStoreMetricsManager.incTimerDequeueCount(getRealTopic(msgExt));
if (tr.getEnqueueTime() == Long.MAX_VALUE) {
// never enqueue, mark it.
MessageAccessor.putProperty(msgExt, TIMER_ENQUEUE_MS, String.valueOf(Long.MAX_VALUE));
}
addMetric(msgExt, -1);
MessageExtBrokerInner msg = convert(msgExt, tr.getEnqueueTime(), needRoll(tr.getMagic()));
doRes = PUT_NEED_RETRY != doPut(msg, needRoll(tr.getMagic()));
while (!doRes && !isStopped()) {
if (!isRunningDequeue()) {
Expand Down
5 changes: 4 additions & 1 deletion store/src/main/java/org/apache/rocketmq/store/timer/TimerMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@
import java.util.concurrent.locks.ReentrantLock;
import org.apache.rocketmq.common.ConfigManager;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -78,7 +80,8 @@ public long updateDistPair(int period, int value) {
return distPair.getCount().addAndGet(value);
}

public long addAndGet(String topic, int value) {
public long addAndGet(MessageExt msg, int value) {
String topic = msg.getProperty(MessageConst.PROPERTY_REAL_TOPIC);
Metric pair = getTopicPair(topic);
getDataVersion().nextVersion();
pair.setTimeStamp(System.currentTimeMillis());
Expand Down
7 changes: 5 additions & 2 deletions store/src/main/java/org/apache/rocketmq/store/timer/TimerRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,9 @@ public class TimerRequest {
private final int sizePy;
private final long delayTime;

private final long enqueueTime;
private final int magic;

private long enqueueTime;
private MessageExt msg;


Expand Down Expand Up @@ -94,7 +95,9 @@ public void setDeleteList(Set<String> deleteList) {
public void setLatch(CountDownLatch latch) {
this.latch = latch;
}

public void setEnqueueTime(long enqueueTime) {
this.enqueueTime = enqueueTime;
}
public void idempotentRelease() {
idempotentRelease(true);
}
Expand Down
10 changes: 8 additions & 2 deletions store/src/test/java/org/apache/rocketmq/store/timer/TimerMetricsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,9 @@
*/
package org.apache.rocketmq.store.timer;

import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Assert;
import org.junit.Test;

Expand All @@ -31,8 +34,11 @@ public void testTimingCount() {

TimerMetrics first = new TimerMetrics(baseDir);
Assert.assertTrue(first.load());
first.addAndGet("AAA", 1000);
first.addAndGet("BBB", 2000);
MessageExt msg = new MessageExt();
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, "AAA");
first.addAndGet(msg, 1000);
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, "BBB");
first.addAndGet(msg, 2000);
Assert.assertEquals(1000, first.getTimingCount("AAA"));
Assert.assertEquals(2000, first.getTimingCount("BBB"));
long curr = System.currentTimeMillis();
Expand Down
Loading