[ISSUE #9111] Support export broker RocksDB Config to json file

broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBConsumerOffsetManager.java

@@ -38,7 +38,7 @@ public class RocksDBConsumerOffsetManager extends ConsumerOffsetManager {
 
     protected static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
 
-    protected RocksDBConfigManager rocksDBConfigManager;
+    protected transient RocksDBConfigManager rocksDBConfigManager;
 
     public RocksDBConsumerOffsetManager(BrokerController brokerController) {
         super(brokerController);
@@ -138,6 +138,11 @@ public synchronized void persist() {
         }
     }
 
+    public synchronized void exportToJson() {
+        LOG.info("export Consumer Offset to json file");
+        super.persist();
+    }
+
     private void putWriteBatch(final WriteBatch writeBatch, final String topicGroupName, final ConcurrentMap<Integer, Long> offsetMap) throws Exception {
         byte[] keyBytes = topicGroupName.getBytes(DataConverter.CHARSET_UTF8);
         RocksDBOffsetSerializeWrapper wrapper = new RocksDBOffsetSerializeWrapper();

broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBSubscriptionGroupManager.java

@@ -37,7 +37,7 @@
 
 public class RocksDBSubscriptionGroupManager extends SubscriptionGroupManager {
 
-    protected RocksDBConfigManager rocksDBConfigManager;
+    protected transient RocksDBConfigManager rocksDBConfigManager;
 
     public RocksDBSubscriptionGroupManager(BrokerController brokerController) {
         super(brokerController, false);
@@ -184,6 +184,11 @@ public synchronized void persist() {
         }
     }
 
+    public synchronized void exportToJson() {
+        log.info("export Subscription Group to json file");
+        super.persist();
+    }
+
     public String rocksdbConfigFilePath() {
         return this.brokerController.getMessageStoreConfig().getStorePathRootDir() + File.separator + "config" + File.separator + "subscriptionGroups" + File.separator;
     }

broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBTopicConfigManager.java

@@ -32,7 +32,7 @@
 
 public class RocksDBTopicConfigManager extends TopicConfigManager {
 
-    protected RocksDBConfigManager rocksDBConfigManager;
+    protected transient RocksDBConfigManager rocksDBConfigManager;
 
     public RocksDBTopicConfigManager(BrokerController brokerController) {
         super(brokerController, false);
@@ -139,6 +139,11 @@ public synchronized void persist() {
         }
     }
 
+    public synchronized void exportToJson() {
+        log.info("export Topic Config to json file");
+        super.persist();
+    }
+
     public String rocksdbConfigFilePath() {
         return this.brokerController.getMessageStoreConfig().getStorePathRootDir() + File.separator + "config" + File.separator + "topics" + File.separator;
     }

broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java

@@ -35,6 +35,7 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
@@ -60,6 +61,9 @@
 import org.apache.rocketmq.broker.auth.converter.UserConverter;
 import org.apache.rocketmq.broker.client.ClientChannelInfo;
 import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
+import org.apache.rocketmq.broker.config.v1.RocksDBConsumerOffsetManager;
+import org.apache.rocketmq.broker.config.v1.RocksDBSubscriptionGroupManager;
+import org.apache.rocketmq.broker.config.v1.RocksDBTopicConfigManager;
 import org.apache.rocketmq.broker.controller.ReplicasManager;
 import org.apache.rocketmq.broker.filter.ConsumerFilterData;
 import org.apache.rocketmq.broker.filter.ExpressionMessageFilter;
@@ -193,6 +197,7 @@
 import org.apache.rocketmq.remoting.protocol.header.ResetMasterFlushOffsetHeader;
 import org.apache.rocketmq.remoting.protocol.header.ResetOffsetRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.ResumeCheckHalfMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.RocksDBConfigToJsonRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.SearchOffsetRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.SearchOffsetResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.UpdateAclRequestHeader;
@@ -239,7 +244,7 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
     private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
     protected final BrokerController brokerController;
     protected Set<String> configBlackList = new HashSet<>();
-    private final ExecutorService asyncExecuteWorker = new ThreadPoolExecutor(0, 1, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());
+    private final ExecutorService asyncExecuteWorker = new ThreadPoolExecutor(0, 4, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());
 
     public AdminBrokerProcessor(final BrokerController brokerController) {
         this.brokerController = brokerController;
@@ -356,6 +361,8 @@ public RemotingCommand processRequest(ChannelHandlerContext ctx,
                 return queryConsumeQueue(ctx, request);
             case RequestCode.CHECK_ROCKSDB_CQ_WRITE_PROGRESS:
                 return this.checkRocksdbCqWriteProgress(ctx, request);
+            case RequestCode.EXPORT_ROCKSDB_CONFIG_TO_JSON:
+                return this.exportRocksDBConfigToJson(ctx, request);
             case RequestCode.UPDATE_AND_GET_GROUP_FORBIDDEN:
                 return this.updateAndGetGroupForbidden(ctx, request);
             case RequestCode.GET_SUBSCRIPTIONGROUP_CONFIG:
@@ -495,6 +502,60 @@ private RemotingCommand checkRocksdbCqWriteProgress(ChannelHandlerContext ctx, R
         return response;
     }
 
+    private RemotingCommand exportRocksDBConfigToJson(ChannelHandlerContext ctx,
+        RemotingCommand request) throws RemotingCommandException {
+        RocksDBConfigToJsonRequestHeader requestHeader = request.decodeCommandCustomHeader(RocksDBConfigToJsonRequestHeader.class);
+        List<RocksDBConfigToJsonRequestHeader.ConfigType> configType = requestHeader.getConfigType();
+        List<CompletableFuture<Void>> futureList = new ArrayList<>();
+        for (RocksDBConfigToJsonRequestHeader.ConfigType type : configType) {
+            switch (type) {
+                case topics:
+                    if (this.brokerController.getTopicConfigManager() instanceof RocksDBTopicConfigManager) {
+                        RocksDBTopicConfigManager rocksDBTopicConfigManager = (RocksDBTopicConfigManager) this.brokerController.getTopicConfigManager();
+                        futureList.add(CompletableFuture.supplyAsync(() -> {
+                            rocksDBTopicConfigManager.exportToJson();
+                            return null;
+                        }, asyncExecuteWorker));
+                    }
+                    break;
+                case subscriptionGroups:
+                    if (this.brokerController.getSubscriptionGroupManager() instanceof RocksDBSubscriptionGroupManager) {
+                        RocksDBSubscriptionGroupManager rocksDBSubscriptionGroupManager = (RocksDBSubscriptionGroupManager) this.brokerController.getSubscriptionGroupManager();
+                        futureList.add(CompletableFuture.supplyAsync(() -> {
+                            rocksDBSubscriptionGroupManager.exportToJson();
+                            return null;
+                        }, asyncExecuteWorker));
+                    }
+                    break;
+                case consumerOffsets:
+                    if (this.brokerController.getConsumerOffsetManager() instanceof RocksDBConsumerOffsetManager) {
+                        RocksDBConsumerOffsetManager rocksDBConsumerOffsetManager = (RocksDBConsumerOffsetManager) this.brokerController.getConsumerOffsetManager();
+                        futureList.add(CompletableFuture.supplyAsync(() -> {
+                            rocksDBConsumerOffsetManager.exportToJson();
+                            return null;
+                        }, asyncExecuteWorker));
+                    }
+                    break;
+                default:
+                    break;
+            }
+        }
+
+        try {
+            CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
+        } catch (CompletionException e) {
+            RemotingCommand response = RemotingCommand.createResponseCommand(null);
+            response.setCode(ResponseCode.SYSTEM_ERROR);
+            response.setRemark("export failed." + e);
+            return response;
+        }
+
+        RemotingCommand response = RemotingCommand.createResponseCommand(null);
+        response.setCode(ResponseCode.SUCCESS);
+        response.setRemark("export done.");
+        return response;
+    }
+
     @Override
     public boolean rejectRequest() {
         return false;

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -210,6 +210,7 @@
 import org.apache.rocketmq.remoting.protocol.header.ResetMasterFlushOffsetHeader;
 import org.apache.rocketmq.remoting.protocol.header.ResetOffsetRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.ResumeCheckHalfMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.RocksDBConfigToJsonRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.SearchOffsetRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.SearchOffsetResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
@@ -3036,6 +3037,21 @@ public CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(final String broker
         throw new MQClientException(response.getCode(), response.getRemark());
     }
 
+    public void exportRocksDBConfigToJson(final String brokerAddr,
+        final List<RocksDBConfigToJsonRequestHeader.ConfigType> configType,
+        final long timeoutMillis) throws InterruptedException,
+        RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
+        RocksDBConfigToJsonRequestHeader header = new RocksDBConfigToJsonRequestHeader();
+        header.setConfigType(configType);
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.EXPORT_ROCKSDB_CONFIG_TO_JSON, header);
+        RemotingCommand response = this.remotingClient.invokeSync(brokerAddr, request, timeoutMillis);
+        assert response != null;
+
+        if (ResponseCode.SUCCESS != response.getCode()) {
+            throw new MQClientException(response.getCode(), response.getRemark());
+        }
+    }
+
     public void checkClientInBroker(final String brokerAddr, final String consumerGroup,
         final String clientId, final SubscriptionData subscriptionData,
         final long timeoutMillis)

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RequestCode.java

@@ -219,6 +219,7 @@ public class RequestCode {
     public static final int GET_SUBSCRIPTIONGROUP_CONFIG = 352;
     public static final int UPDATE_AND_GET_GROUP_FORBIDDEN = 353;
     public static final int CHECK_ROCKSDB_CQ_WRITE_PROGRESS = 354;
+    public static final int EXPORT_ROCKSDB_CONFIG_TO_JSON = 355;
 
     public static final int LITE_PULL_MESSAGE = 361;
     public static final int RECALL_MESSAGE = 370;

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/RocksDBConfigToJsonRequestHeader.java

@@ -0,0 +1,73 @@
+/*
+ * 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.remoting.protocol.header;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.action.Action;
+import org.apache.rocketmq.common.action.RocketMQAction;
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNotNull;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+
+@RocketMQAction(value = RequestCode.EXPORT_ROCKSDB_CONFIG_TO_JSON, action = Action.GET)
+public class RocksDBConfigToJsonRequestHeader implements CommandCustomHeader {
+    private static final String CONFIG_TYPE_SEPARATOR = ";";
+
+    public enum ConfigType {
+        topics,
+        subscriptionGroups,
+        consumerOffsets;
+
+        public static List<ConfigType> fromString(String ordinal) {
+            String[] configTypeStrings = StringUtils.split(ordinal, CONFIG_TYPE_SEPARATOR);
+            List<ConfigType> configTypes = new ArrayList<>();
+            for (String configTypeString : configTypeStrings) {
+                if (StringUtils.isNotEmpty(configTypeString)) {
+                    configTypes.add(ConfigType.valueOf(configTypeString));
+                }
+            }
+            return configTypes;
+        }
+
+        public static String toString(List<ConfigType> configTypes) {
+            StringBuilder sb = new StringBuilder();
+            for (ConfigType configType : configTypes) {
+                sb.append(configType.name()).append(CONFIG_TYPE_SEPARATOR);
+            }
+            return sb.toString();
+        }
+    }
+
+    @CFNotNull
+    private String configType;
+
+    @Override
+    public void checkFields() throws RemotingCommandException {
+
+    }
+
+    public List<ConfigType> getConfigType() {
+        return ConfigType.fromString(configType);
+    }
+
+    public void setConfigType(List<ConfigType> configType) {
+        this.configType = ConfigType.toString(configType);
+    }
+}

remoting/src/test/java/org/apache/rocketmq/remoting/protocol/header/RocksDBConfigToJsonRequestHeaderTest.java

@@ -0,0 +1,40 @@
+/*
+ * 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.remoting.protocol.header;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.Test;
+
+public class RocksDBConfigToJsonRequestHeaderTest {
+    @Test
+    public void configTypeTest() {
+        List<RocksDBConfigToJsonRequestHeader.ConfigType> configTypes = new ArrayList<>();
+        configTypes.add(RocksDBConfigToJsonRequestHeader.ConfigType.topics);
+        configTypes.add(RocksDBConfigToJsonRequestHeader.ConfigType.subscriptionGroups);
+
+        String string = RocksDBConfigToJsonRequestHeader.ConfigType.toString(configTypes);
+
+        List<RocksDBConfigToJsonRequestHeader.ConfigType> newConfigTypes = RocksDBConfigToJsonRequestHeader.ConfigType.fromString(string);
+        assert newConfigTypes.size() == 2;
+        assert configTypes.equals(newConfigTypes);
+
+        List<RocksDBConfigToJsonRequestHeader.ConfigType> topics = RocksDBConfigToJsonRequestHeader.ConfigType.fromString("topics");
+        assert topics.size() == 1;
+        assert topics.get(0).equals(RocksDBConfigToJsonRequestHeader.ConfigType.topics);
+    }
+}

tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java

@@ -65,6 +65,7 @@
 import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
 import org.apache.rocketmq.remoting.protocol.body.TopicList;
 import org.apache.rocketmq.remoting.protocol.body.UserInfo;
+import org.apache.rocketmq.remoting.protocol.header.RocksDBConfigToJsonRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader;
 import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
@@ -778,6 +779,13 @@ public CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(String brokerAddr,
         return this.defaultMQAdminExtImpl.checkRocksdbCqWriteProgress(brokerAddr, topic, checkStoreTime);
     }
 
+    @Override
+    public void exportRocksDBConfigToJson(String brokerAddr,
+        List<RocksDBConfigToJsonRequestHeader.ConfigType> configType)
+        throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
+        this.defaultMQAdminExtImpl.exportRocksDBConfigToJson(brokerAddr, configType);
+    }
+
     @Override
     public boolean resumeCheckHalfMessage(String topic,
         String msgId)

tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java

@@ -103,6 +103,7 @@
 import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
 import org.apache.rocketmq.remoting.protocol.body.TopicList;
 import org.apache.rocketmq.remoting.protocol.body.UserInfo;
+import org.apache.rocketmq.remoting.protocol.header.RocksDBConfigToJsonRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.UpdateConsumerOffsetRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.UpdateGroupForbiddenRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
@@ -1824,6 +1825,13 @@ public CheckRocksdbCqWriteResult checkRocksdbCqWriteProgress(String brokerAddr,
         return this.mqClientInstance.getMQClientAPIImpl().checkRocksdbCqWriteProgress(brokerAddr, topic, checkStoreTime, timeoutMillis);
     }
 
+    @Override
+    public void exportRocksDBConfigToJson(String brokerAddr,
+        List<RocksDBConfigToJsonRequestHeader.ConfigType> configType)
+        throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
+        this.mqClientInstance.getMQClientAPIImpl().exportRocksDBConfigToJson(brokerAddr, configType, timeoutMillis);
+    }
+
     @Override
     public boolean resumeCheckHalfMessage(final String topic,
         final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {

tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java

@@ -61,6 +61,7 @@
 import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
 import org.apache.rocketmq.remoting.protocol.body.TopicList;
 import org.apache.rocketmq.remoting.protocol.body.UserInfo;
+import org.apache.rocketmq.remoting.protocol.header.RocksDBConfigToJsonRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader;
 import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
@@ -392,6 +393,10 @@ QueryConsumeQueueResponseBody queryConsumeQueue(final String brokerAddr,
         final long index, final int count, final String consumerGroup)
         throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException;
 
+    void exportRocksDBConfigToJson(String brokerAddr,
+        List<RocksDBConfigToJsonRequestHeader.ConfigType> configType)
+        throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException;
+
     boolean resumeCheckHalfMessage(final String topic,
         final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException;