Jason918 commented on a change in pull request #3337:
URL: https://github.com/apache/rocketmq/pull/3337#discussion_r718991689
##########
File path:
common/src/main/java/org/apache/rocketmq/common/message/MessageConst.java
##########
@@ -52,6 +52,10 @@
public static final String PROPERTY_PUSH_REPLY_TIME = "PUSH_REPLY_TIME";
public static final String PROPERTY_CLUSTER = "CLUSTER";
public static final String PROPERTY_MESSAGE_TYPE = "MSG_TYPE";
+ public static final String PROPERTY_SPECIFY_DELAY_TIME =
"SPECIFY_DELAY_TIME";
Review comment:
Maybe "CUSTOM_DELAY_TIME" is a better name?
##########
File path: namesrv/src/main/java/org/apache/rocketmq/namesrv/NamesrvStartup.java
##########
@@ -48,6 +50,7 @@
private static CommandLine commandLine = null;
public static void main(String[] args) {
+ System.setProperty(MixAll.ROCKETMQ_HOME_PROPERTY,
"/Users/dragonboy/IdeaProjects/rocketmq/distribution");
Review comment:
Test code?
##########
File path:
store/src/main/java/org/apache/rocketmq/store/schedule/CustomDelayMessageService.java
##########
@@ -0,0 +1,496 @@
+package org.apache.rocketmq.store.schedule;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.ConfigManager;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.TopicFilterType;
+import org.apache.rocketmq.common.constant.LoggerName;
+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.running.RunningStats;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.store.*;
+import org.apache.rocketmq.store.config.StorePathConfigHelper;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.TimerTask;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * @ClassName CutomDelayMessageService
+ * @Version 1.0
+ * @Author dragonboy
+ * @Date 2021/9/6 14:12
+ * @Description
+ **/
+public class CustomDelayMessageService extends ConfigManager {
+ private static final InternalLogger log =
InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
+
+ public static final String SCHEDULE_TOPIC = "DRAGON_TOPIC_XXX";
+
+ private static final Pattern pattern =
Pattern.compile("([0-9]+d)*([0-9]+h)*([0-9]+m)*([0-9]+s)*");
+
+ private static final HashMap<String, Long> timeUnitTable = new
HashMap<String, Long>() {{
+ this.put("s", 1000L);
+ this.put("m", 1000L * 60);
+ this.put("h", 1000L * 60 * 60);
+ this.put("d", 1000L * 60 * 60 * 24);
+ }};
+
+ //最大限度的处理层级
+ public static int MAX_LIMIT_LEVEL = 0;
Review comment:
This should not be constant and not configurable.
##########
File path:
store/src/main/java/org/apache/rocketmq/store/config/StorePathConfigHelper.java
##########
@@ -48,6 +48,10 @@ public static String getDelayOffsetStorePath(final String
rootDir) {
return rootDir + File.separator + "config" + File.separator +
"delayOffset.json";
}
+ public static String getSpcifyDelayOffsetStorePath(final String rootDir) {
+ return rootDir + File.separator + "dragon" + File.separator +
"delayOffset.json";
Review comment:
"dragon" here is a little confusing.
##########
File path:
store/src/main/java/org/apache/rocketmq/store/schedule/CustomDelayMessageService.java
##########
@@ -0,0 +1,496 @@
+package org.apache.rocketmq.store.schedule;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.ConfigManager;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.TopicFilterType;
+import org.apache.rocketmq.common.constant.LoggerName;
+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.running.RunningStats;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.store.*;
+import org.apache.rocketmq.store.config.StorePathConfigHelper;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.TimerTask;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * @ClassName CutomDelayMessageService
+ * @Version 1.0
+ * @Author dragonboy
+ * @Date 2021/9/6 14:12
+ * @Description
+ **/
+public class CustomDelayMessageService extends ConfigManager {
Review comment:
Please add some unit tests.
##########
File path: broker/src/main/java/org/apache/rocketmq/broker/BrokerStartup.java
##########
@@ -64,7 +64,7 @@ public static BrokerController start(BrokerController
controller) {
controller.start();
String tip = "The broker[" +
controller.getBrokerConfig().getBrokerName() + ", "
- + controller.getBrokerAddr() + "] boot success.
serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
+ + controller.getBrokerAddr() + "] boot success.
serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
Review comment:
I think it's better not changing the format of old code without logical
modification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
