From 7e018520ef707a841c66c55d621f6560d03b631b Mon Sep 17 00:00:00 2001
From: Zhouxiang Zhan <zhouxzhan@apache.org>
Date: Fri, 25 Aug 2023 09:49:22 +0800
Subject: [PATCH 1/6] Add expireAfterAccess for cache (#7247)
Add expireAfterAccess for cache
.../rocketmq/proxy/config/ProxyConfig.java | 59 ++++++++++++++-----
.../metadata/ClusterMetadataService.java | 6 +-
.../service/route/TopicRouteService.java | 14 +++--
3 files changed, 56 insertions(+), 23 deletions(-)
@@ -155,14 +155,17 @@ public class ProxyConfig implements ConfigFile {
private int consumerProcessorThreadPoolQueueCapacity = 10000;
private boolean useEndpointPortFromRequest = false;
- private int topicRouteServiceCacheExpiredInSeconds = 20;
+
+ private int topicRouteServiceCacheExpiredSeconds = 300;
+ private int topicRouteServiceCacheRefreshSeconds = 20;
private int topicRouteServiceCacheMaxNum = 20000;
private int topicRouteServiceThreadPoolNums = PROCESSOR_NUMBER;
private int topicRouteServiceThreadPoolQueueCapacity = 5000;
-
- private int topicConfigCacheExpiredInSeconds = 20;
+ private int topicConfigCacheExpiredSeconds = 300;
+ private int topicConfigCacheRefreshSeconds = 20;
private int topicConfigCacheMaxNum = 20000;
- private int subscriptionGroupConfigCacheExpiredInSeconds = 20;
+ private int subscriptionGroupConfigCacheExpiredSeconds = 300;
+ private int subscriptionGroupConfigCacheRefreshSeconds = 20;
private int subscriptionGroupConfigCacheMaxNum = 20000;
private int metadataThreadPoolNums = 3;
private int metadataThreadPoolQueueCapacity = 100000;
@@ -794,12 +797,20 @@ public class ProxyConfig implements ConfigFile {
this.consumerProcessorThreadPoolQueueCapacity = consumerProcessorThreadPoolQueueCapacity;
}
- public int getTopicRouteServiceCacheExpiredInSeconds() {
- return topicRouteServiceCacheExpiredInSeconds;
+ public int getTopicRouteServiceCacheExpiredSeconds() {
+ return topicRouteServiceCacheExpiredSeconds;
+ }
+
+ public void setTopicRouteServiceCacheExpiredSeconds(int topicRouteServiceCacheExpiredSeconds) {
+ this.topicRouteServiceCacheExpiredSeconds = topicRouteServiceCacheExpiredSeconds;
}
- public void setTopicRouteServiceCacheExpiredInSeconds(int topicRouteServiceCacheExpiredInSeconds) {
- this.topicRouteServiceCacheExpiredInSeconds = topicRouteServiceCacheExpiredInSeconds;
+ public int getTopicRouteServiceCacheRefreshSeconds() {
+ return topicRouteServiceCacheRefreshSeconds;
+ }
+
+ public void setTopicRouteServiceCacheRefreshSeconds(int topicRouteServiceCacheRefreshSeconds) {
+ this.topicRouteServiceCacheRefreshSeconds = topicRouteServiceCacheRefreshSeconds;
}
public int getTopicRouteServiceCacheMaxNum() {
@@ -826,12 +837,20 @@ public class ProxyConfig implements ConfigFile {
this.topicRouteServiceThreadPoolQueueCapacity = topicRouteServiceThreadPoolQueueCapacity;
}
- public int getTopicConfigCacheExpiredInSeconds() {
- return topicConfigCacheExpiredInSeconds;
+ public int getTopicConfigCacheRefreshSeconds() {
+ return topicConfigCacheRefreshSeconds;
+ }
+
+ public void setTopicConfigCacheRefreshSeconds(int topicConfigCacheRefreshSeconds) {
+ this.topicConfigCacheRefreshSeconds = topicConfigCacheRefreshSeconds;
+ }
+
+ public int getTopicConfigCacheExpiredSeconds() {
+ return topicConfigCacheExpiredSeconds;
}
- public void setTopicConfigCacheExpiredInSeconds(int topicConfigCacheExpiredInSeconds) {
- this.topicConfigCacheExpiredInSeconds = topicConfigCacheExpiredInSeconds;
+ public void setTopicConfigCacheExpiredSeconds(int topicConfigCacheExpiredSeconds) {
+ this.topicConfigCacheExpiredSeconds = topicConfigCacheExpiredSeconds;
}
public int getTopicConfigCacheMaxNum() {
@@ -842,12 +861,20 @@ public class ProxyConfig implements ConfigFile {
this.topicConfigCacheMaxNum = topicConfigCacheMaxNum;
}
- public int getSubscriptionGroupConfigCacheExpiredInSeconds() {
- return subscriptionGroupConfigCacheExpiredInSeconds;
+ public int getSubscriptionGroupConfigCacheRefreshSeconds() {
+ return subscriptionGroupConfigCacheRefreshSeconds;
+ }
+
+ public void setSubscriptionGroupConfigCacheRefreshSeconds(int subscriptionGroupConfigCacheRefreshSeconds) {
+ this.subscriptionGroupConfigCacheRefreshSeconds = subscriptionGroupConfigCacheRefreshSeconds;
+ }
+
+ public int getSubscriptionGroupConfigCacheExpiredSeconds() {
+ return subscriptionGroupConfigCacheExpiredSeconds;
}
- public void setSubscriptionGroupConfigCacheExpiredInSeconds(int subscriptionGroupConfigCacheExpiredInSeconds) {
- this.subscriptionGroupConfigCacheExpiredInSeconds = subscriptionGroupConfigCacheExpiredInSeconds;
+ public void setSubscriptionGroupConfigCacheExpiredSeconds(int subscriptionGroupConfigCacheExpiredSeconds) {
+ this.subscriptionGroupConfigCacheExpiredSeconds = subscriptionGroupConfigCacheExpiredSeconds;
}
public int getSubscriptionGroupConfigCacheMaxNum() {
@@ -69,11 +69,13 @@ public class ClusterMetadataService extends AbstractStartAndShutdown implements
);
this.topicConfigCache = CacheBuilder.newBuilder()
.maximumSize(config.getTopicConfigCacheMaxNum())
- .refreshAfterWrite(config.getTopicConfigCacheExpiredInSeconds(), TimeUnit.SECONDS)
+ .expireAfterAccess(config.getTopicConfigCacheExpiredSeconds(), TimeUnit.SECONDS)
+ .refreshAfterWrite(config.getTopicConfigCacheRefreshSeconds(), TimeUnit.SECONDS)
.build(new ClusterTopicConfigCacheLoader());
this.subscriptionGroupConfigCache = CacheBuilder.newBuilder()
.maximumSize(config.getSubscriptionGroupConfigCacheMaxNum())
- .refreshAfterWrite(config.getSubscriptionGroupConfigCacheExpiredInSeconds(), TimeUnit.SECONDS)
+ .expireAfterAccess(config.getSubscriptionGroupConfigCacheExpiredSeconds(), TimeUnit.SECONDS)
+ .refreshAfterWrite(config.getSubscriptionGroupConfigCacheRefreshSeconds(), TimeUnit.SECONDS)
.build(new ClusterSubscriptionGroupConfigCacheLoader());
this.init();
@@ -68,10 +68,13 @@ public abstract class TopicRouteService extends AbstractStartAndShutdown {
);
this.mqClientAPIFactory = mqClientAPIFactory;
- this.topicCache = Caffeine.newBuilder().maximumSize(config.getTopicRouteServiceCacheMaxNum()).
- refreshAfterWrite(config.getTopicRouteServiceCacheExpiredInSeconds(), TimeUnit.SECONDS).
- executor(cacheRefreshExecutor).build(new CacheLoader<String, MessageQueueView>() {
- @Override public @Nullable MessageQueueView load(String topic) throws Exception {
+ this.topicCache = Caffeine.newBuilder().maximumSize(config.getTopicRouteServiceCacheMaxNum())
+ .expireAfterAccess(config.getTopicRouteServiceCacheExpiredSeconds(), TimeUnit.SECONDS)
+ .refreshAfterWrite(config.getTopicRouteServiceCacheRefreshSeconds(), TimeUnit.SECONDS)
+ .executor(cacheRefreshExecutor)
+ .build(new CacheLoader<String, MessageQueueView>() {
+ @Override
+ public @Nullable MessageQueueView load(String topic) throws Exception {
try {
TopicRouteData topicRouteData = mqClientAPIFactory.getClient().getTopicRouteInfoFromNameServer(topic, Duration.ofSeconds(3).toMillis());
return buildMessageQueueView(topic, topicRouteData);
@@ -83,7 +86,8 @@ public abstract class TopicRouteService extends AbstractStartAndShutdown {
}
}
- @Override public @Nullable MessageQueueView reload(@NonNull String key,
+ @Override
+ public @Nullable MessageQueueView reload(@NonNull String key,
@NonNull MessageQueueView oldValue) throws Exception {
try {
return load(key);
--
2.32.0.windows.2
From 5f6dc90f9dab35809fcb0407d4d5cc2737d2335e Mon Sep 17 00:00:00 2001
From: Ziyi Tan <tanziyi0925@gmail.com>
Date: Fri, 25 Aug 2023 11:17:23 +0800
Subject: [PATCH 2/6] [ISSUE #7250] Beautify command rocksDBConfigToJson output
Co-authored-by: Ziy1-Tan <ajb4596984460@gmail.com>
.../metadata/RocksDBConfigToJsonCommand.java | 32 +++++++++++--------
1 file changed, 18 insertions(+), 14 deletions(-)
@@ -21,13 +21,13 @@ import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.config.RocksDBConfigManager;
import org.apache.rocketmq.common.utils.DataConverter;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.tools.command.SubCommand;
import org.apache.rocketmq.tools.command.SubCommandException;
-import java.io.File;
import java.util.HashMap;
import java.util.Map;
@@ -48,7 +48,7 @@ public class RocksDBConfigToJsonCommand implements SubCommand {
@Override
public Options buildCommandlineOptions(Options options) {
Option pathOption = new Option("p", "path", true,
- "Absolute path to the metadata directory");
+ "Absolute path for the metadata directory");
pathOption.setRequired(true);
options.addOption(pathOption);
@@ -63,15 +63,14 @@ public class RocksDBConfigToJsonCommand implements SubCommand {
@Override
public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException {
String path = commandLine.getOptionValue("path").trim();
- if (StringUtils.isEmpty(path) || !new File(path).exists()) {
+ if (StringUtils.isEmpty(path) || !UtilAll.isPathExists(path)) {
System.out.print("Rocksdb path is invalid.\n");
return;
}
String configType = commandLine.getOptionValue("configType").trim().toLowerCase();
- final long memTableFlushInterval = 60 * 60 * 1000L;
- RocksDBConfigManager kvConfigManager = new RocksDBConfigManager(memTableFlushInterval);
+ RocksDBConfigManager kvConfigManager = new RocksDBConfigManager(60 * 60 * 1000L);
try {
if (TOPICS_JSON_CONFIG.toLowerCase().equals(configType)) {
// for topics.json
@@ -84,13 +83,16 @@ public class RocksDBConfigToJsonCommand implements SubCommand {
topicConfigTable.put(topic, jsonObject);
});
- if (isLoad) {
- topicsJsonConfig.put("topicConfigTable", (JSONObject) JSONObject.toJSON(topicConfigTable));
- final String topicsJsonStr = JSONObject.toJSONString(topicsJsonConfig, true);
- System.out.print(topicsJsonStr + "\n");
+ if (!isLoad) {
+ System.out.print("RocksDB load error, path=" + path);
return;
}
+ topicsJsonConfig.put("topicConfigTable", (JSONObject) JSONObject.toJSON(topicConfigTable));
+ final String topicsJsonStr = JSONObject.toJSONString(topicsJsonConfig, true);
+ System.out.print(topicsJsonStr + "\n");
+ return;
}
+
if (SUBSCRIPTION_GROUP_JSON_CONFIG.toLowerCase().equals(configType)) {
// for subscriptionGroup.json
final Map<String, JSONObject> subscriptionGroupJsonConfig = new HashMap<>();
@@ -102,13 +104,15 @@ public class RocksDBConfigToJsonCommand implements SubCommand {
subscriptionGroupTable.put(subscriptionGroup, jsonObject);
});
- if (isLoad) {
- subscriptionGroupJsonConfig.put("subscriptionGroupTable",
- (JSONObject) JSONObject.toJSON(subscriptionGroupTable));
- final String subscriptionGroupJsonStr = JSONObject.toJSONString(subscriptionGroupJsonConfig, true);
- System.out.print(subscriptionGroupJsonStr + "\n");
+ if (!isLoad) {
+ System.out.print("RocksDB load error, path=" + path);
return;
}
+ subscriptionGroupJsonConfig.put("subscriptionGroupTable",
+ (JSONObject) JSONObject.toJSON(subscriptionGroupTable));
+ final String subscriptionGroupJsonStr = JSONObject.toJSONString(subscriptionGroupJsonConfig, true);
+ System.out.print(subscriptionGroupJsonStr + "\n");
+ return;
}
System.out.print("Config type was not recognized, configType=" + configType + "\n");
} finally {
--
2.32.0.windows.2
From b4f73e2aabc1b141cec98431899e4090340adf0f Mon Sep 17 00:00:00 2001
From: mxsm <ljbmxsm@gmail.com>
Date: Sun, 27 Aug 2023 20:58:58 +0800
Subject: [PATCH 3/6] [ISSUE #7271] Optimize the configuration for setting the
quantity of TimerDequeuePutMessageService (#7272)
.../java/org/apache/rocketmq/store/timer/TimerMessageStore.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
@@ -222,7 +222,7 @@ public class TimerMessageStore {
dequeueGetMessageServices[i] = new TimerDequeueGetMessageService();
}
- int putThreadNum = Math.max(storeConfig.getTimerGetMessageThreadNum(), 1);
+ int putThreadNum = Math.max(storeConfig.getTimerPutMessageThreadNum(), 1);
dequeuePutMessageServices = new TimerDequeuePutMessageService[putThreadNum];
for (int i = 0; i < dequeuePutMessageServices.length; i++) {
dequeuePutMessageServices[i] = new TimerDequeuePutMessageService();
--
2.32.0.windows.2
From 3e100103af68588528bf32f3752a85e8023f46f8 Mon Sep 17 00:00:00 2001
From: Ziyi Tan <tanziyi0925@gmail.com>
Date: Tue, 29 Aug 2023 13:48:51 +0800
Subject: [PATCH 4/6] [ISSUE #7277] Enhance rocksDBConfigToJson to support
metadata counting (#7276)
.../common/config/AbstractRocksDBStorage.java | 4 +-
.../common/config/ConfigRocksDBStorage.java | 6 +
.../tools/command/MQAdminStartup.java | 4 +-
.../ExportMetadataInRocksDBCommand.java | 138 ++++++++++++++++++
.../metadata/RocksDBConfigToJsonCommand.java | 122 ----------------
...> ExportMetadataInRocksDBCommandTest.java} | 38 +++--
6 files changed, 173 insertions(+), 139 deletions(-)
create mode 100644 tools/src/main/java/org/apache/rocketmq/tools/command/export/ExportMetadataInRocksDBCommand.java
delete mode 100644 tools/src/main/java/org/apache/rocketmq/tools/command/metadata/RocksDBConfigToJsonCommand.java
rename tools/src/test/java/org/apache/rocketmq/tools/command/metadata/{KvConfigToJsonCommandTest.java => ExportMetadataInRocksDBCommandTest.java} (62%)
@@ -385,8 +385,10 @@ public abstract class AbstractRocksDBStorage {
this.options.close();
}
//4. close db.
- if (db != null) {
+ if (db != null && !this.readOnly) {
this.db.syncWal();
+ }
+ if (db != null) {
this.db.closeE();
}
//5. help gc.
@@ -60,6 +60,12 @@ public class ConfigRocksDBStorage extends AbstractRocksDBStorage {
this.readOnly = false;
}
+ public ConfigRocksDBStorage(final String dbPath, boolean readOnly) {
+ super();
+ this.dbPath = dbPath;
+ this.readOnly = readOnly;
+ }
+
private void initOptions() {
this.options = createConfigDBOptions();
@@ -80,7 +80,7 @@ import org.apache.rocketmq.tools.command.message.QueryMsgByOffsetSubCommand;
import org.apache.rocketmq.tools.command.message.QueryMsgByUniqueKeySubCommand;
import org.apache.rocketmq.tools.command.message.QueryMsgTraceByIdSubCommand;
import org.apache.rocketmq.tools.command.message.SendMessageCommand;
-import org.apache.rocketmq.tools.command.metadata.RocksDBConfigToJsonCommand;
+import org.apache.rocketmq.tools.command.export.ExportMetadataInRocksDBCommand;
import org.apache.rocketmq.tools.command.namesrv.AddWritePermSubCommand;
import org.apache.rocketmq.tools.command.namesrv.DeleteKvConfigCommand;
import org.apache.rocketmq.tools.command.namesrv.GetNamesrvConfigCommand;
@@ -212,7 +212,6 @@ public class MQAdminStartup {
initCommand(new ClusterListSubCommand());
initCommand(new TopicListSubCommand());
- initCommand(new RocksDBConfigToJsonCommand());
initCommand(new UpdateKvConfigCommand());
initCommand(new DeleteKvConfigCommand());
@@ -257,6 +256,7 @@ public class MQAdminStartup {
initCommand(new ExportMetadataCommand());
initCommand(new ExportConfigsCommand());
initCommand(new ExportMetricsCommand());
+ initCommand(new ExportMetadataInRocksDBCommand());
initCommand(new HAStatusSubCommand());
new file mode 100644
@@ -0,0 +1,138 @@
+/*
+ * 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.tools.command.export;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.config.ConfigRocksDBStorage;
+import org.apache.rocketmq.common.utils.DataConverter;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.tools.command.SubCommand;
+import org.apache.rocketmq.tools.command.SubCommandException;
+import org.rocksdb.RocksIterator;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
+
+public class ExportMetadataInRocksDBCommand implements SubCommand {
+ private static final String TOPICS_JSON_CONFIG = "topics";
+ private static final String SUBSCRIPTION_GROUP_JSON_CONFIG = "subscriptionGroups";
+
+ @Override
+ public String commandName() {
+ return "exportMetadataInRocksDB";
+ }
+
+ @Override
+ public String commandDesc() {
+ return "export RocksDB kv config (topics/subscriptionGroups)";
+ }
+
+ @Override
+ public Options buildCommandlineOptions(Options options) {
+ Option pathOption = new Option("p", "path", true,
+ "Absolute path for the metadata directory");
+ pathOption.setRequired(true);
+ options.addOption(pathOption);
+
+ Option configTypeOption = new Option("t", "configType", true, "Name of kv config, e.g. " +
+ "topics/subscriptionGroups");
+ configTypeOption.setRequired(true);
+ options.addOption(configTypeOption);
+
+ Option jsonEnableOption = new Option("j", "jsonEnable", true,
+ "Json format enable, Default: false");
+ jsonEnableOption.setRequired(false);
+ options.addOption(jsonEnableOption);
+
+ return options;
+ }
+
+ @Override
+ public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException {
+ String path = commandLine.getOptionValue("path").trim();
+ if (StringUtils.isEmpty(path) || !UtilAll.isPathExists(path)) {
+ System.out.print("RocksDB path is invalid.\n");
+ return;
+ }
+
+ String configType = commandLine.getOptionValue("configType").trim().toLowerCase();
+
+ boolean jsonEnable = false;
+ if (commandLine.hasOption("jsonEnable")) {
+ jsonEnable = Boolean.parseBoolean(commandLine.getOptionValue("jsonEnable").trim());
+ }
+
+
+ ConfigRocksDBStorage kvStore = new ConfigRocksDBStorage(path, true /* readOnly */);
+ if (!kvStore.start()) {
+ System.out.print("RocksDB load error, path=" + path + "\n");
+ return;
+ }
+
+ try {
+ if (TOPICS_JSON_CONFIG.equalsIgnoreCase(configType) || SUBSCRIPTION_GROUP_JSON_CONFIG.equalsIgnoreCase(configType)) {
+ handleExportMetadata(kvStore, configType, jsonEnable);
+ } else {
+ System.out.printf("Invalid config type=%s, Options: topics,subscriptionGroups\n", configType);
+ }
+ } finally {
+ kvStore.shutdown();
+ }
+ }
+
+ private static void handleExportMetadata(ConfigRocksDBStorage kvStore, String configType, boolean jsonEnable) {
+ if (jsonEnable) {
+ final Map<String, JSONObject> jsonConfig = new HashMap<>();
+ final Map<String, JSONObject> configTable = new HashMap<>();
+ iterateKvStore(kvStore, (key, value) -> {
+ final String configKey = new String(key, DataConverter.charset);
+ final String configValue = new String(value, DataConverter.charset);
+ final JSONObject jsonObject = JSONObject.parseObject(configValue);
+ configTable.put(configKey, jsonObject);
+ }
+ );
+
+ jsonConfig.put(configType.equalsIgnoreCase(TOPICS_JSON_CONFIG) ? "topicConfigTable" : "subscriptionGroupTable",
+ (JSONObject) JSONObject.toJSON(configTable));
+ final String jsonConfigStr = JSONObject.toJSONString(jsonConfig, true);
+ System.out.print(jsonConfigStr + "\n");
+ } else {
+ AtomicLong count = new AtomicLong(0);
+ iterateKvStore(kvStore, (key, value) -> {
+ final String configKey = new String(key, DataConverter.charset);
+ final String configValue = new String(value, DataConverter.charset);
+ System.out.printf("%d, Key: %s, Value: %s%n", count.incrementAndGet(), configKey, configValue);
+ });
+ }
+ }
+
+ private static void iterateKvStore(ConfigRocksDBStorage kvStore, BiConsumer<byte[], byte[]> biConsumer) {
+ try (RocksIterator iterator = kvStore.iterator()) {
+ iterator.seekToFirst();
+ for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) {
+ biConsumer.accept(iterator.key(), iterator.value());
+ }
+ }
+ }
+}
deleted file mode 100644
@@ -1,122 +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.tools.command.metadata;
-
-import com.alibaba.fastjson.JSONObject;
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.rocketmq.common.UtilAll;
-import org.apache.rocketmq.common.config.RocksDBConfigManager;
-import org.apache.rocketmq.common.utils.DataConverter;
-import org.apache.rocketmq.remoting.RPCHook;
-import org.apache.rocketmq.tools.command.SubCommand;
-import org.apache.rocketmq.tools.command.SubCommandException;
-
-import java.util.HashMap;
-import java.util.Map;
-
-public class RocksDBConfigToJsonCommand implements SubCommand {
- private static final String TOPICS_JSON_CONFIG = "topics";
- private static final String SUBSCRIPTION_GROUP_JSON_CONFIG = "subscriptionGroups";
-
- @Override
- public String commandName() {
- return "rocksDBConfigToJson";
- }
-
- @Override
- public String commandDesc() {
- return "Convert RocksDB kv config (topics/subscriptionGroups) to json";
- }
-
- @Override
- public Options buildCommandlineOptions(Options options) {
- Option pathOption = new Option("p", "path", true,
- "Absolute path for the metadata directory");
- pathOption.setRequired(true);
- options.addOption(pathOption);
-
- Option configTypeOption = new Option("t", "configType", true, "Name of kv config, e.g. " +
- "topics/subscriptionGroups");
- configTypeOption.setRequired(true);
- options.addOption(configTypeOption);
-
- return options;
- }
-
- @Override
- public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException {
- String path = commandLine.getOptionValue("path").trim();
- if (StringUtils.isEmpty(path) || !UtilAll.isPathExists(path)) {
- System.out.print("Rocksdb path is invalid.\n");
- return;
- }
-
- String configType = commandLine.getOptionValue("configType").trim().toLowerCase();
-
- RocksDBConfigManager kvConfigManager = new RocksDBConfigManager(60 * 60 * 1000L);
- try {
- if (TOPICS_JSON_CONFIG.toLowerCase().equals(configType)) {
- // for topics.json
- final Map<String, JSONObject> topicsJsonConfig = new HashMap<>();
- final Map<String, JSONObject> topicConfigTable = new HashMap<>();
- boolean isLoad = kvConfigManager.load(path, (key, value) -> {
- final String topic = new String(key, DataConverter.charset);
- final String topicConfig = new String(value, DataConverter.charset);
- final JSONObject jsonObject = JSONObject.parseObject(topicConfig);
- topicConfigTable.put(topic, jsonObject);
- });
-
- if (!isLoad) {
- System.out.print("RocksDB load error, path=" + path);
- return;
- }
- topicsJsonConfig.put("topicConfigTable", (JSONObject) JSONObject.toJSON(topicConfigTable));
- final String topicsJsonStr = JSONObject.toJSONString(topicsJsonConfig, true);
- System.out.print(topicsJsonStr + "\n");
- return;
- }
-
- if (SUBSCRIPTION_GROUP_JSON_CONFIG.toLowerCase().equals(configType)) {
- // for subscriptionGroup.json
- final Map<String, JSONObject> subscriptionGroupJsonConfig = new HashMap<>();
- final Map<String, JSONObject> subscriptionGroupTable = new HashMap<>();
- boolean isLoad = kvConfigManager.load(path, (key, value) -> {
- final String subscriptionGroup = new String(key, DataConverter.charset);
- final String subscriptionGroupConfig = new String(value, DataConverter.charset);
- final JSONObject jsonObject = JSONObject.parseObject(subscriptionGroupConfig);
- subscriptionGroupTable.put(subscriptionGroup, jsonObject);
- });
-
- if (!isLoad) {
- System.out.print("RocksDB load error, path=" + path);
- return;
- }
- subscriptionGroupJsonConfig.put("subscriptionGroupTable",
- (JSONObject) JSONObject.toJSON(subscriptionGroupTable));
- final String subscriptionGroupJsonStr = JSONObject.toJSONString(subscriptionGroupJsonConfig, true);
- System.out.print(subscriptionGroupJsonStr + "\n");
- return;
- }
- System.out.print("Config type was not recognized, configType=" + configType + "\n");
- } finally {
- kvConfigManager.stop();
- }
- }
-}
similarity index 62%
rename from tools/src/test/java/org/apache/rocketmq/tools/command/metadata/KvConfigToJsonCommandTest.java
rename to tools/src/test/java/org/apache/rocketmq/tools/command/metadata/ExportMetadataInRocksDBCommandTest.java
@@ -21,43 +21,53 @@ import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Options;
import org.apache.rocketmq.srvutil.ServerUtil;
import org.apache.rocketmq.tools.command.SubCommandException;
+import org.apache.rocketmq.tools.command.export.ExportMetadataInRocksDBCommand;
import org.junit.Test;
import java.io.File;
import static org.assertj.core.api.Assertions.assertThat;
-public class KvConfigToJsonCommandTest {
+public class ExportMetadataInRocksDBCommandTest {
private static final String BASE_PATH = System.getProperty("user.home") + File.separator + "store/config/";
@Test
public void testExecute() throws SubCommandException {
{
- String[] cases = new String[]{"topics", "subscriptionGroups"};
- for (String c : cases) {
- RocksDBConfigToJsonCommand cmd = new RocksDBConfigToJsonCommand();
+ String[][] cases = new String[][] {
+ {"topics", "false"},
+ {"topics", "false1"},
+ {"topics", "true"},
+ {"subscriptionGroups", "false"},
+ {"subscriptionGroups", "false2"},
+ {"subscriptionGroups", "true"}
+ };
+
+ for (String[] c : cases) {
+ ExportMetadataInRocksDBCommand cmd = new ExportMetadataInRocksDBCommand();
Options options = ServerUtil.buildCommandlineOptions(new Options());
- String[] subargs = new String[]{"-p " + BASE_PATH + c, "-t " + c};
+ String[] subargs = new String[] {"-p " + BASE_PATH + c[0], "-t " + c[0], "-j " + c[1]};
final CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " + cmd.commandName(), subargs,
- cmd.buildCommandlineOptions(options), new DefaultParser());
+ cmd.buildCommandlineOptions(options), new DefaultParser());
cmd.execute(commandLine, options, null);
- assertThat(commandLine.getOptionValue("p").trim()).isEqualTo(BASE_PATH + c);
- assertThat(commandLine.getOptionValue("t").trim()).isEqualTo(c);
+ assertThat(commandLine.getOptionValue("p").trim()).isEqualTo(BASE_PATH + c[0]);
+ assertThat(commandLine.getOptionValue("t").trim()).isEqualTo(c[0]);
+ assertThat(commandLine.getOptionValue("j").trim()).isEqualTo(c[1]);
}
}
// invalid cases
{
- String[][] cases = new String[][]{
- {"-p " + BASE_PATH + "tmpPath", "-t topics"},
- {"-p ", "-t topics"},
- {"-p " + BASE_PATH + "topics", "-t invalid_type"}
+ String[][] cases = new String[][] {
+ {"-p " + BASE_PATH + "tmpPath", "-t topics", "-j true"},
+ {"-p ", "-t topics", "-j true"},
+ {"-p " + BASE_PATH + "topics", "-t invalid_type", "-j true"}
};
for (String[] c : cases) {
- RocksDBConfigToJsonCommand cmd = new RocksDBConfigToJsonCommand();
+ ExportMetadataInRocksDBCommand cmd = new ExportMetadataInRocksDBCommand();
Options options = ServerUtil.buildCommandlineOptions(new Options());
final CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " + cmd.commandName(), c,
- cmd.buildCommandlineOptions(options), new DefaultParser());
+ cmd.buildCommandlineOptions(options), new DefaultParser());
cmd.execute(commandLine, options, null);
}
}
--
2.32.0.windows.2
From fa549154370cb866a90e37c13a90d2c598d6b1f6 Mon Sep 17 00:00:00 2001
From: yuz10 <845238369@qq.com>
Date: Tue, 29 Aug 2023 15:22:09 +0800
Subject: [PATCH 5/6] [ISSUE #7261] Slave high CPU usage when
enableScheduleAsyncDeliver=true (#7262)
* [ISSUE #6390] Add break to the exception of WHEEL_TIMER_NOT_ENABLE.
* fix broker start fail if mapped file size is 0
* log
* only delete the last empty file
* change dataReadAheadEnable default to true
* fix endless loop when master change to slave.
.../rocketmq/broker/schedule/ScheduleMessageService.java | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
@@ -566,7 +566,8 @@ public class ScheduleMessageService extends ConfigManager {
pendingQueue.remove();
break;
case RUNNING:
- break;
+ scheduleNextTask();
+ return;
case EXCEPTION:
if (!isStarted()) {
log.warn("HandlePutResultTask shutdown, info={}", putResultProcess.toString());
@@ -586,6 +587,10 @@ public class ScheduleMessageService extends ConfigManager {
}
}
+ scheduleNextTask();
+ }
+
+ private void scheduleNextTask() {
if (isStarted()) {
ScheduleMessageService.this.handleExecutorService
.schedule(new HandlePutResultTask(this.delayLevel), DELAY_FOR_A_SLEEP, TimeUnit.MILLISECONDS);
--
2.32.0.windows.2
From 9f34f55e1dac495730c9cd5469f2ab3225b8f0b9 Mon Sep 17 00:00:00 2001
From: ShuangxiDing <dingshuangxi888@gmail.com>
Date: Tue, 29 Aug 2023 15:48:46 +0800
Subject: [PATCH 6/6] [ISSUE #7226] Filter tlvs in ppv2 which contents not are
spec-compliant ASCII characters and space (#7227)
Filter tlvs in ppv2 which not are spec-compliant ASCII characters and space
.../rocketmq/common/utils/BinaryUtil.java | 17 +++++++++++++++++
.../grpc/ProxyAndTlsProtocolNegotiator.java | 8 +++++++-
.../remoting/netty/NettyRemotingServer.java | 8 +++++++-
3 files changed, 31 insertions(+), 2 deletions(-)
@@ -43,4 +43,21 @@ public class BinaryUtil {
byte[] bytes = calculateMd5(content);
return Hex.encodeHexString(bytes, false);
}
+
+ /**
+ * Returns true if subject contains only bytes that are spec-compliant ASCII characters.
+ * @param subject
+ * @return
+ */
+ public static boolean isAscii(byte[] subject) {
+ if (subject == null) {
+ return false;
+ }
+ for (byte b : subject) {
+ if ((b & 0x80) != 0) {
+ return false;
+ }
+ }
+ return true;
+ }
}
\ No newline at end of file
@@ -24,6 +24,7 @@ import io.grpc.netty.shaded.io.grpc.netty.InternalProtocolNegotiator;
import io.grpc.netty.shaded.io.grpc.netty.InternalProtocolNegotiators;
import io.grpc.netty.shaded.io.grpc.netty.ProtocolNegotiationEvent;
import io.grpc.netty.shaded.io.netty.buffer.ByteBuf;
+import io.grpc.netty.shaded.io.netty.buffer.ByteBufUtil;
import io.grpc.netty.shaded.io.netty.channel.ChannelHandler;
import io.grpc.netty.shaded.io.netty.channel.ChannelHandlerContext;
import io.grpc.netty.shaded.io.netty.channel.ChannelInboundHandlerAdapter;
@@ -44,6 +45,7 @@ import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.constant.HAProxyConstants;
import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.utils.BinaryUtil;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
@@ -191,9 +193,13 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator
}
if (CollectionUtils.isNotEmpty(msg.tlvs())) {
msg.tlvs().forEach(tlv -> {
+ byte[] valueBytes = ByteBufUtil.getBytes(tlv.content());
+ if (!BinaryUtil.isAscii(valueBytes)) {
+ return;
+ }
Attributes.Key<String> key = AttributeKeys.valueOf(
HAProxyConstants.PROXY_PROTOCOL_TLV_PREFIX + String.format("%02x", tlv.typeByteValue()));
- String value = StringUtils.trim(tlv.content().toString(CharsetUtil.UTF_8));
+ String value = StringUtils.trim(new String(valueBytes, CharsetUtil.UTF_8));
builder.set(key, value);
});
}
@@ -18,6 +18,7 @@ package org.apache.rocketmq.remoting.netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
@@ -58,6 +59,7 @@ import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.HAProxyConstants;
import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.utils.BinaryUtil;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
@@ -787,9 +789,13 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
}
if (CollectionUtils.isNotEmpty(msg.tlvs())) {
msg.tlvs().forEach(tlv -> {
+ byte[] valueBytes = ByteBufUtil.getBytes(tlv.content());
+ if (!BinaryUtil.isAscii(valueBytes)) {
+ return;
+ }
AttributeKey<String> key = AttributeKeys.valueOf(
HAProxyConstants.PROXY_PROTOCOL_TLV_PREFIX + String.format("%02x", tlv.typeByteValue()));
- String value = StringUtils.trim(tlv.content().toString(CharsetUtil.UTF_8));
+ String value = StringUtils.trim(new String(valueBytes, CharsetUtil.UTF_8));
channel.attr(key).set(value);
});
}
--
2.32.0.windows.2