You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2022/08/08 08:00:32 UTC
[rocketmq] branch develop updated: [ISSUE #4725] Optimize the CK/ACK 's invisibleTime in POP consumption mode. (#4753)
This is an automated email from the ASF dual-hosted git repository.
jinrongtong pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 705e717a0 [ISSUE #4725] Optimize the CK/ACK 's invisibleTime in POP consumption mode. (#4753)
705e717a0 is described below
commit 705e717a092239919e819d08df9844a19b89ceb0
Author: Ji Juntao <31...@zju.edu.cn>
AuthorDate: Mon Aug 8 16:00:25 2022 +0800
[ISSUE #4725] Optimize the CK/ACK 's invisibleTime in POP consumption mode. (#4753)
* Replace scheduled message in Pop with timer message.
* add unit test.
* refactor
* WIP
* improve unit test
* improve unit test
* improve unit test
* fix unnecessary stubbing problem.
* fix misspelling 'replase' in doc.
* refactor and rebuild
* fix the bug of #4702
Co-authored-by: 傅冲 <yu...@alibaba-inc.com>
---
.../apache/rocketmq/broker/BrokerController.java | 4 +
.../broker/processor/AckMessageProcessor.java | 3 +-
.../processor/ChangeInvisibleTimeProcessor.java | 5 +-
.../broker/processor/PopBufferMergeService.java | 7 +-
.../broker/processor/PopMessageProcessor.java | 3 +-
.../broker/processor/PopReviveService.java | 13 +-
.../org/apache/rocketmq/broker/util/MsgUtil.java | 34 ---
.../broker/processor/AckMessageProcessorTest.java | 4 -
.../ChangeInvisibleTimeProcessorTest.java | 4 -
.../broker/processor/PopMessageProcessorTest.java | 5 -
.../broker/processor/PopReviveServiceTest.java | 227 +++++++++++++++++++++
.../apache/rocketmq/common/message/Message.java | 4 +
docs/cn/controller/quick_start.md | 2 +-
.../rocketmq/example/filter/TagFilterConsumer.java | 1 -
14 files changed, 249 insertions(+), 67 deletions(-)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
index 6d4fe3d80..d5e93b388 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
@@ -1186,6 +1186,10 @@ public class BrokerController {
return timerMessageStore;
}
+ public void setTimerMessageStore(TimerMessageStore timerMessageStore) {
+ this.timerMessageStore = timerMessageStore;
+ }
+
public AckMessageProcessor getAckMessageProcessor() {
return ackMessageProcessor;
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
index c3c94b8f1..8bd3c613c 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
@@ -20,7 +20,6 @@ import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.broker.BrokerController;
-import org.apache.rocketmq.broker.util.MsgUtil;
import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.PopAckConstants;
import org.apache.rocketmq.common.TopicConfig;
@@ -199,7 +198,7 @@ public class AckMessageProcessor implements NettyRequestProcessor {
msgInner.setBornTimestamp(System.currentTimeMillis());
msgInner.setBornHost(this.brokerController.getStoreHost());
msgInner.setStoreHost(this.brokerController.getStoreHost());
- MsgUtil.setMessageDeliverTime(this.brokerController, msgInner, ExtraInfoUtil.getPopTime(extraInfo) + ExtraInfoUtil.getInvisibleTime(extraInfo));
+ msgInner.setDeliverTimeMs(ExtraInfoUtil.getPopTime(extraInfo) + ExtraInfoUtil.getInvisibleTime(extraInfo));
msgInner.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, PopMessageProcessor.genAckUniqueId(ackMsg));
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
PutMessageResult putMessageResult = this.brokerController.getEscapeBridge().putMessageToSpecificQueue(msgInner);
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java
index 5ab1647e4..bb68badc1 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java
@@ -20,7 +20,6 @@ import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.broker.BrokerController;
-import org.apache.rocketmq.broker.util.MsgUtil;
import org.apache.rocketmq.common.PopAckConstants;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.LoggerName;
@@ -152,7 +151,7 @@ public class ChangeInvisibleTimeProcessor implements NettyRequestProcessor {
msgInner.setBornTimestamp(System.currentTimeMillis());
msgInner.setBornHost(this.brokerController.getStoreHost());
msgInner.setStoreHost(this.brokerController.getStoreHost());
- MsgUtil.setMessageDeliverTime(this.brokerController, msgInner, ExtraInfoUtil.getPopTime(extraInfo) + ExtraInfoUtil.getInvisibleTime(extraInfo));
+ msgInner.setDeliverTimeMs(ExtraInfoUtil.getPopTime(extraInfo) + ExtraInfoUtil.getInvisibleTime(extraInfo));
msgInner.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, PopMessageProcessor.genAckUniqueId(ackMsg));
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
PutMessageResult putMessageResult = this.brokerController.getEscapeBridge().putMessageToSpecificQueue(msgInner);
@@ -187,7 +186,7 @@ public class ChangeInvisibleTimeProcessor implements NettyRequestProcessor {
msgInner.setBornTimestamp(System.currentTimeMillis());
msgInner.setBornHost(this.brokerController.getStoreHost());
msgInner.setStoreHost(this.brokerController.getStoreHost());
- MsgUtil.setMessageDeliverTime(this.brokerController, msgInner, ck.getReviveTime() - PopAckConstants.ackTimeInterval);
+ msgInner.setDeliverTimeMs(ck.getReviveTime() - PopAckConstants.ackTimeInterval);
msgInner.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, PopMessageProcessor.genCkUniqueId(ck));
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
index c21dec6b1..4b2df0875 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
@@ -23,7 +23,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.rocketmq.broker.BrokerController;
-import org.apache.rocketmq.broker.util.MsgUtil;
import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.PopAckConstants;
import org.apache.rocketmq.common.ServiceThread;
@@ -572,8 +571,7 @@ public class PopBufferMergeService extends ServiceThread {
msgInner.setBornTimestamp(System.currentTimeMillis());
msgInner.setBornHost(brokerController.getStoreHost());
msgInner.setStoreHost(brokerController.getStoreHost());
-
- MsgUtil.setMessageDeliverTime(this.brokerController, msgInner, point.getReviveTime());
+ msgInner.setDeliverTimeMs(point.getReviveTime());
msgInner.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, PopMessageProcessor.genAckUniqueId(ackMsg));
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
@@ -607,8 +605,7 @@ public class PopBufferMergeService extends ServiceThread {
msgInner.setBornHost(brokerController.getStoreHost());
msgInner.setStoreHost(brokerController.getStoreHost());
- MsgUtil.setMessageDeliverTime(brokerController, msgInner, point.getReviveTime() - PopAckConstants.ackTimeInterval);
-
+ msgInner.setDeliverTimeMs(point.getReviveTime() - PopAckConstants.ackTimeInterval);
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
PutMessageResult putMessageResult = brokerController.getEscapeBridge().putMessageToSpecificQueue(msgInner);
if (putMessageResult.getPutMessageStatus() != PutMessageStatus.PUT_OK
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
index 8ac4a6a54..842d0d98b 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
@@ -38,7 +38,6 @@ import org.apache.rocketmq.broker.filter.ConsumerFilterManager;
import org.apache.rocketmq.broker.filter.ExpressionMessageFilter;
import org.apache.rocketmq.broker.longpolling.PopRequest;
import org.apache.rocketmq.broker.pagecache.ManyMessageTransfer;
-import org.apache.rocketmq.broker.util.MsgUtil;
import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.PopAckConstants;
import org.apache.rocketmq.common.ServiceThread;
@@ -658,7 +657,7 @@ public class PopMessageProcessor implements NettyRequestProcessor {
msgInner.setBornTimestamp(System.currentTimeMillis());
msgInner.setBornHost(this.brokerController.getStoreHost());
msgInner.setStoreHost(this.brokerController.getStoreHost());
- MsgUtil.setMessageDeliverTime(this.brokerController, msgInner, ck.getReviveTime() - PopAckConstants.ackTimeInterval);
+ msgInner.setDeliverTimeMs(ck.getReviveTime() - PopAckConstants.ackTimeInterval);
msgInner.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, genCkUniqueId(ck));
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java
index cbe38a3b8..1b95cce4e 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java
@@ -25,7 +25,6 @@ import java.util.HashMap;
import java.util.List;
import org.apache.rocketmq.broker.BrokerController;
-import org.apache.rocketmq.broker.util.MsgUtil;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.common.KeyBuilder;
@@ -154,7 +153,7 @@ public class PopReviveService extends ServiceThread {
}
}
- private List<MessageExt> getReviveMessage(long offset, int queueId) {
+ protected List<MessageExt> getReviveMessage(long offset, int queueId) {
PullResult pullResult = getMessage(PopAckConstants.REVIVE_GROUP, reviveTopic, queueId, offset, 32);
if (pullResult == null) {
return null;
@@ -259,7 +258,7 @@ public class PopReviveService extends ServiceThread {
return foundList;
}
- private void consumeReviveMessage(ConsumeReviveObj consumeReviveObj) {
+ protected void consumeReviveMessage(ConsumeReviveObj consumeReviveObj) {
HashMap<String, PopCheckPoint> map = consumeReviveObj.map;
long startScanTime = System.currentTimeMillis();
long endTime = 0;
@@ -278,8 +277,10 @@ public class PopReviveService extends ServiceThread {
List<MessageExt> messageExts = getReviveMessage(offset, queueId);
if (messageExts == null || messageExts.isEmpty()) {
long old = endTime;
+ long timerDelay = brokerController.getMessageStore().getTimerMessageStore().getReadBehind();
+ long commitLogDelay = brokerController.getMessageStore().getTimerMessageStore().getEnqueueBehind();
// move endTime
- if (endTime != 0 && ((System.currentTimeMillis() - endTime) > (3 * PopAckConstants.SECOND))) {
+ if (endTime != 0 && System.currentTimeMillis() - endTime > 3 * PopAckConstants.SECOND && timerDelay <= 0 && commitLogDelay <= 0) {
endTime = System.currentTimeMillis();
}
POP_LOGGER.info("reviveQueueId={}, offset is {}, can not get new msg, old endTime {}, new endTime {}",
@@ -336,7 +337,7 @@ public class PopReviveService extends ServiceThread {
POP_LOGGER.error("invalid ack index, {}, {}", ackMsg, point);
}
}
- long deliverTime = MsgUtil.getMessageDeliverTime(this.brokerController, messageExt);
+ long deliverTime = messageExt.getDeliverTimeMs();
if (deliverTime > endTime) {
endTime = deliverTime;
}
@@ -346,7 +347,7 @@ public class PopReviveService extends ServiceThread {
consumeReviveObj.endTime = endTime;
}
- private void mergeAndRevive(ConsumeReviveObj consumeReviveObj) throws Throwable {
+ protected void mergeAndRevive(ConsumeReviveObj consumeReviveObj) throws Throwable {
ArrayList<PopCheckPoint> sortList = consumeReviveObj.genSortList();
POP_LOGGER.info("reviveQueueId={},ck listSize={}", queueId, sortList.size());
if (sortList.size() != 0) {
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/util/MsgUtil.java b/broker/src/main/java/org/apache/rocketmq/broker/util/MsgUtil.java
deleted file mode 100644
index 938df4377..000000000
--- a/broker/src/main/java/org/apache/rocketmq/broker/util/MsgUtil.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.rocketmq.broker.util;
-
-import org.apache.rocketmq.broker.BrokerController;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.message.MessageExt;
-
-public final class MsgUtil {
- private MsgUtil() {
- }
-
- public static void setMessageDeliverTime(BrokerController brokerController, Message msgInner, long timeMillis) {
- msgInner.setDelayTimeLevel(brokerController.getScheduleMessageService().computeDelayLevel(timeMillis));
- }
-
- public static long getMessageDeliverTime(BrokerController brokerController, MessageExt msgInner) {
- return brokerController.getScheduleMessageService().computeDeliverTimestamp(msgInner.getDelayTimeLevel(), msgInner.getStoreTimestamp());
- }
-}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/AckMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/AckMessageProcessorTest.java
index 000058b0a..a59fdde16 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/processor/AckMessageProcessorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/AckMessageProcessorTest.java
@@ -23,7 +23,6 @@ import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.net.Broker2Client;
import org.apache.rocketmq.broker.failover.EscapeBridge;
-import org.apache.rocketmq.broker.schedule.ScheduleMessageService;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.MessageConst;
@@ -85,9 +84,6 @@ public class AckMessageProcessorTest {
Field field = BrokerController.class.getDeclaredField("broker2Client");
field.setAccessible(true);
field.set(brokerController, broker2Client);
- ScheduleMessageService scheduleMessageService = new ScheduleMessageService(brokerController);
- scheduleMessageService.parseDelayLevel();
- when(brokerController.getScheduleMessageService()).thenReturn(scheduleMessageService);
EscapeBridge escapeBridge = new EscapeBridge(brokerController);
Mockito.when(brokerController.getEscapeBridge()).thenReturn(escapeBridge);
Channel mockChannel = mock(Channel.class);
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessorTest.java
index ff792b358..2a009e951 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessorTest.java
@@ -22,7 +22,6 @@ import java.lang.reflect.Field;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.net.Broker2Client;
-import org.apache.rocketmq.broker.schedule.ScheduleMessageService;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.MessageConst;
@@ -82,9 +81,6 @@ public class ChangeInvisibleTimeProcessorTest {
Field field = BrokerController.class.getDeclaredField("broker2Client");
field.setAccessible(true);
field.set(brokerController, broker2Client);
- ScheduleMessageService scheduleMessageService = new ScheduleMessageService(brokerController);
- scheduleMessageService.parseDelayLevel();
- when(brokerController.getScheduleMessageService()).thenReturn(scheduleMessageService);
Channel mockChannel = mock(Channel.class);
when(handlerContext.channel()).thenReturn(mockChannel);
brokerController.getTopicConfigManager().getTopicConfigTable().put(topic, new TopicConfig());
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java
index 85582fcff..af6b4bb55 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java
@@ -22,7 +22,6 @@ import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
-import org.apache.rocketmq.broker.schedule.ScheduleMessageService;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.ConsumeInitMode;
@@ -71,7 +70,6 @@ public class PopMessageProcessorTest {
private ChannelHandlerContext handlerContext;
@Mock
private DefaultMessageStore messageStore;
- private ScheduleMessageService scheduleMessageService;
private ClientChannelInfo clientChannelInfo;
private String group = "FooBarGroup";
private String topic = "FooBar";
@@ -80,9 +78,6 @@ public class PopMessageProcessorTest {
public void init() {
brokerController.setMessageStore(messageStore);
popMessageProcessor = new PopMessageProcessor(brokerController);
- scheduleMessageService = new ScheduleMessageService(brokerController);
- scheduleMessageService.parseDelayLevel();
- when(brokerController.getScheduleMessageService()).thenReturn(scheduleMessageService);
when(messageStore.putMessage(any())).thenReturn(new PutMessageResult(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK)));
Channel mockChannel = mock(Channel.class);
when(mockChannel.remoteAddress()).thenReturn(new InetSocketAddress(1024));
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java
new file mode 100644
index 000000000..e9af449cc
--- /dev/null
+++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java
@@ -0,0 +1,227 @@
+/*
+ * 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.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.broker.processor;
+
+import com.alibaba.fastjson.JSON;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
+import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
+import org.apache.rocketmq.broker.topic.TopicConfigManager;
+import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.PopAckConstants;
+import org.apache.rocketmq.common.TopicConfig;
+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.MessageExtBrokerInner;
+import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.common.utils.DataConverter;
+import org.apache.rocketmq.remoting.common.RemotingUtil;
+import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.config.BrokerRole;
+import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.apache.rocketmq.store.pop.AckMsg;
+import org.apache.rocketmq.store.pop.PopCheckPoint;
+import org.apache.rocketmq.store.timer.TimerMessageStore;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.mockito.stubbing.Answer;
+
+import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.Silent.class)
+public class PopReviveServiceTest {
+
+ private static final String REVIVE_TOPIC = PopAckConstants.REVIVE_TOPIC + "test";
+ private static final int REVIVE_QUEUE_ID = 0;
+ private static final String GROUP = "group";
+ private static final String TOPIC = "topic";
+ private static final SocketAddress STORE_HOST = RemotingUtil.string2SocketAddress("127.0.0.1:8080");
+
+ @Mock
+ private MessageStore messageStore;
+ @Mock
+ private ConsumerOffsetManager consumerOffsetManager;
+ @Mock
+ private MessageStoreConfig messageStoreConfig;
+ @Mock
+ private TopicConfigManager topicConfigManager;
+ @Mock
+ private TimerMessageStore timerMessageStore;
+ @Mock
+ private SubscriptionGroupManager subscriptionGroupManager;
+ @Mock
+ private BrokerController brokerController;
+
+ private BrokerConfig brokerConfig;
+ private PopReviveService popReviveService;
+
+ @Before
+ public void before() {
+ brokerConfig = new BrokerConfig();
+
+ when(brokerController.getBrokerConfig()).thenReturn(brokerConfig);
+ when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
+ when(brokerController.getMessageStore()).thenReturn(messageStore);
+ when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager);
+ when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager);
+ when(messageStore.getTimerMessageStore()).thenReturn(timerMessageStore);
+ when(timerMessageStore.getReadBehind()).thenReturn(0L);
+ when(timerMessageStore.getEnqueueBehind()).thenReturn(0L);
+
+ when(topicConfigManager.selectTopicConfig(anyString())).thenReturn(new TopicConfig());
+ when(subscriptionGroupManager.findSubscriptionGroupConfig(anyString())).thenReturn(new SubscriptionGroupConfig());
+
+ popReviveService = spy(new PopReviveService(brokerController, REVIVE_TOPIC, REVIVE_QUEUE_ID));
+ popReviveService.setShouldRunPopRevive(true);
+ }
+ @Test
+ public void testWhenAckMoreThanCk() throws Throwable {
+ long maxReviveOffset = 4;
+
+ when(consumerOffsetManager.queryOffset(PopAckConstants.REVIVE_GROUP, REVIVE_TOPIC, REVIVE_QUEUE_ID))
+ .thenReturn(0L);
+ List<MessageExt> reviveMessageExtList = new ArrayList<>();
+ long basePopTime = System.currentTimeMillis();
+ {
+ // put a pair of ck and ack
+ PopCheckPoint ck = buildPopCheckPoint(1, basePopTime, 1);
+ reviveMessageExtList.add(buildCkMsg(ck));
+ reviveMessageExtList.add(buildAckMsg(buildAckMsg(1, basePopTime), ck.getReviveTime(), 1, basePopTime));
+ }
+ {
+ for (int i = 2; i <= maxReviveOffset; i++) {
+ long popTime = basePopTime + i;
+ PopCheckPoint ck = buildPopCheckPoint(i, popTime, i);
+ reviveMessageExtList.add(buildAckMsg(buildAckMsg(i, popTime), ck.getReviveTime(), i, popTime));
+ }
+ }
+ AtomicBoolean firstCall = new AtomicBoolean(true);
+ doAnswer((Answer<List<MessageExt>>) mock -> {
+ if (firstCall.get()) {
+ firstCall.set(false);
+ return reviveMessageExtList;
+ }
+ return null;
+ }).when(popReviveService).getReviveMessage(anyLong(), anyInt());
+
+ PopReviveService.ConsumeReviveObj consumeReviveObj = new PopReviveService.ConsumeReviveObj();
+ popReviveService.consumeReviveMessage(consumeReviveObj);
+
+ assertEquals(1, consumeReviveObj.map.size());
+
+ AtomicLong committedOffset = new AtomicLong(-1);
+ doAnswer(mock -> {
+ committedOffset.set(mock.getArgument(4));
+ return null;
+ }).when(consumerOffsetManager).commitOffset(anyString(), anyString(), anyString(), anyInt(), anyLong());
+ popReviveService.mergeAndRevive(consumeReviveObj);
+
+ assertEquals(1, committedOffset.get());
+ }
+
+ public static PopCheckPoint buildPopCheckPoint(long startOffset, long popTime, long reviveOffset) {
+ PopCheckPoint ck = new PopCheckPoint();
+ ck.setStartOffset(startOffset);
+ ck.setPopTime(popTime);
+ ck.setQueueId((byte) 0);
+ ck.setCId(GROUP);
+ ck.setTopic(TOPIC);
+ ck.setNum((byte) 1);
+ ck.setBitMap(0);
+ ck.setReviveOffset(reviveOffset);
+ ck.setInvisibleTime(1000);
+ return ck;
+ }
+
+ public static AckMsg buildAckMsg(long offset, long popTime) {
+ AckMsg ackMsg = new AckMsg();
+ ackMsg.setAckOffset(offset);
+ ackMsg.setStartOffset(offset);
+ ackMsg.setConsumerGroup(GROUP);
+ ackMsg.setTopic(TOPIC);
+ ackMsg.setQueueId(0);
+ ackMsg.setPopTime(popTime);
+
+ return ackMsg;
+ }
+
+ public static MessageExtBrokerInner buildCkMsg(PopCheckPoint ck) {
+ MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
+
+ msgInner.setTopic(REVIVE_TOPIC);
+ msgInner.setBody(JSON.toJSONString(ck).getBytes(DataConverter.charset));
+ msgInner.setQueueId(REVIVE_QUEUE_ID);
+ msgInner.setTags(PopAckConstants.CK_TAG);
+ msgInner.setBornTimestamp(System.currentTimeMillis());
+ msgInner.setBornHost(STORE_HOST);
+ msgInner.setStoreHost(STORE_HOST);
+ msgInner.setDeliverTimeMs(ck.getReviveTime() - PopAckConstants.ackTimeInterval);
+ msgInner.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, PopMessageProcessor.genCkUniqueId(ck));
+ msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
+
+ msgInner.setQueueOffset(ck.getReviveOffset());
+
+ return msgInner;
+ }
+
+ public static MessageExtBrokerInner buildAckMsg(AckMsg ackMsg, long deliverMs, long reviveOffset, long deliverTime) {
+ MessageExtBrokerInner messageExtBrokerInner = buildAckInnerMessage(
+ REVIVE_TOPIC,
+ ackMsg,
+ REVIVE_QUEUE_ID,
+ STORE_HOST,
+ deliverMs,
+ PopMessageProcessor.genAckUniqueId(ackMsg)
+ );
+ messageExtBrokerInner.setQueueOffset(reviveOffset);
+ messageExtBrokerInner.setDeliverTimeMs(deliverMs);
+ messageExtBrokerInner.setStoreTimestamp(deliverTime);
+ return messageExtBrokerInner;
+ }
+
+ public static MessageExtBrokerInner buildAckInnerMessage(String reviveTopic, AckMsg ackMsg, int reviveQid, SocketAddress host, long deliverMs, String ackUniqueId) {
+ MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
+ msgInner.setTopic(reviveTopic);
+ msgInner.setBody(JSON.toJSONString(ackMsg).getBytes(DataConverter.charset));
+ msgInner.setQueueId(reviveQid);
+ msgInner.setTags(PopAckConstants.ACK_TAG);
+ msgInner.setBornTimestamp(System.currentTimeMillis());
+ msgInner.setBornHost(host);
+ msgInner.setStoreHost(host);
+ msgInner.setDeliverTimeMs(deliverMs);
+ msgInner.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, ackUniqueId);
+ msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
+
+ return msgInner;
+ }
+}
\ No newline at end of file
diff --git a/common/src/main/java/org/apache/rocketmq/common/message/Message.java b/common/src/main/java/org/apache/rocketmq/common/message/Message.java
index eddb4c288..6e8cad0c3 100644
--- a/common/src/main/java/org/apache/rocketmq/common/message/Message.java
+++ b/common/src/main/java/org/apache/rocketmq/common/message/Message.java
@@ -224,4 +224,8 @@ public class Message implements Serializable {
public void setDeliverTimeMs(long timeMs) {
this.putProperty(MessageConst.PROPERTY_TIMER_DELIVER_MS, String.valueOf(timeMs));
}
+
+ public long getDeliverTimeMs() {
+ return Long.parseLong(this.getUserProperty(MessageConst.PROPERTY_TIMER_DELIVER_MS));
+ }
}
diff --git a/docs/cn/controller/quick_start.md b/docs/cn/controller/quick_start.md
index db36d241b..0bc20eb48 100644
--- a/docs/cn/controller/quick_start.md
+++ b/docs/cn/controller/quick_start.md
@@ -25,7 +25,7 @@ $ mvn -Prelease-all -DskipTests clean install -U
在构建成功后
```shell
-#{rocketmq-version} replase replace with rocketmq actual version. example: 5.0.0-SNAPSHOT
+#{rocketmq-version} replace with rocketmq actual version. example: 5.0.0-SNAPSHOT
$ cd distribution/target/rocketmq-{rocketmq-version}/rocketmq-{rocketmq-version}/
$ sh bin/controller/fast-try.sh start
diff --git a/example/src/main/java/org/apache/rocketmq/example/filter/TagFilterConsumer.java b/example/src/main/java/org/apache/rocketmq/example/filter/TagFilterConsumer.java
index d5639ab98..74fc4a9ee 100644
--- a/example/src/main/java/org/apache/rocketmq/example/filter/TagFilterConsumer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/filter/TagFilterConsumer.java
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.example.filter;
-import java.io.IOException;
import java.util.List;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;