lollipopjin commented on code in PR #8823:
URL: https://github.com/apache/rocketmq/pull/8823#discussion_r1805957112
##########
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java:
##########
@@ -479,62 +483,93 @@ private RemotingCommand
checkRocksdbCqWriteProgress(ChannelHandlerContext ctx, R
defaultMessageStore = (DefaultMessageStore) messageStore;
}
RocksDBMessageStore rocksDBMessageStore =
defaultMessageStore.getRocksDBMessageStore();
+ HashMap<String, String> resultMap = new HashMap<>();
+
+ if
(defaultMessageStore.getMessageStoreConfig().getStoreType().equals(StoreType.DEFAULT_ROCKSDB.getStoreType()))
{
+ resultMap.put("diffResult", "storeType is DEFAULT_ROCKSDB, no need
check");
+ resultMap.put("checkStatus", "0");
+ response.setBody(JSON.toJSONBytes(resultMap));
+ return response;
+ }
+
if
(!defaultMessageStore.getMessageStoreConfig().isRocksdbCQDoubleWriteEnable()) {
- response.setBody(JSON.toJSONBytes(ImmutableMap.of("diffResult",
"rocksdbCQWriteEnable is false, checkRocksdbCqWriteProgressCommand is
invalid")));
+ resultMap.put("diffResult", "rocksdbCQWriteEnable is false,
checkRocksdbCqWriteProgressCommand is invalid");
+ resultMap.put("checkStatus", "1");
+ response.setBody(JSON.toJSONBytes(resultMap));
return response;
}
ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>>
cqTable = defaultMessageStore.getConsumeQueueTable();
StringBuilder diffResult = new StringBuilder();
try {
if (StringUtils.isNotBlank(requestTopic)) {
- processConsumeQueuesForTopic(cqTable.get(requestTopic),
requestTopic, rocksDBMessageStore, diffResult,false);
-
response.setBody(JSON.toJSONBytes(ImmutableMap.of("diffResult",
diffResult.toString())));
+ boolean checkResult =
processConsumeQueuesForTopic(cqTable.get(requestTopic), requestTopic,
rocksDBMessageStore, diffResult, false, requestHeader.getCheckStoreTime());
+ resultMap.put("diffResult", diffResult.toString());
+ resultMap.put("checkStatus", checkResult ? "0" : "1");
+ response.setBody(JSON.toJSONBytes(resultMap));
return response;
}
+ boolean checkResult = true;
for (Map.Entry<String, ConcurrentMap<Integer,
ConsumeQueueInterface>> topicEntry : cqTable.entrySet()) {
String topic = topicEntry.getKey();
- processConsumeQueuesForTopic(topicEntry.getValue(), topic,
rocksDBMessageStore, diffResult,true);
+ checkResult =
processConsumeQueuesForTopic(topicEntry.getValue(), topic, rocksDBMessageStore,
diffResult, true, requestHeader.getCheckStoreTime());
Review Comment:
If checkResult is not true for any topic, should consider to return the
right result, or records detail log for the specific topic.
--
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: commits-unsubscr...@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
