[ISSUE#9097] Add new command to check async task status in broker. #9162

KiteSoar · 2025-01-27T03:29:09Z

Which Issue(s) This PR Fixes

Fixes #9097

Brief Description

How Did You Test This Change?

codecov-commenter · 2025-02-01T08:56:49Z

Codecov Report

Attention: Patch coverage is 35.52632% with 147 lines in your changes missing coverage. Please review.

Project coverage is 48.04%. Comparing base (de4e48d) to head (bf4a370).
Report is 4 commits behind head on develop.

Files with missing lines	Patch %	Lines
...ain/java/org/apache/rocketmq/common/AsyncTask.java	0.00%	28 Missing ⚠️
.../apache/rocketmq/broker/AdminAsyncTaskManager.java	20.58%	27 Missing ⚠️
...ocketmq/broker/processor/AdminBrokerProcessor.java	4.54%	21 Missing ⚠️
.../command/stats/CheckAsyncTaskStatusSubCommand.java	77.65%	14 Missing and 7 partials ⚠️
...ocol/header/CheckAsyncTaskStatusRequestHeader.java	0.00%	17 Missing ⚠️
...col/header/CheckAsyncTaskStatusResponseHeader.java	0.00%	14 Missing ⚠️
...in/java/org/apache/rocketmq/common/TaskStatus.java	0.00%	11 Missing ⚠️
...g/apache/rocketmq/client/impl/MQClientAPIImpl.java	0.00%	6 Missing ⚠️
...apache/rocketmq/tools/admin/DefaultMQAdminExt.java	0.00%	1 Missing ⚠️
...he/rocketmq/tools/admin/DefaultMQAdminExtImpl.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop    #9162      +/-   ##
=============================================
- Coverage      48.11%   48.04%   -0.07%     
- Complexity     12091    12112      +21     
=============================================
  Files           1321     1327       +6     
  Lines          93024    93263     +239     
  Branches       11926    11951      +25     
=============================================
+ Hits           44755    44807      +52     
- Misses         42754    42920     +166     
- Partials        5515     5536      +21

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

qianye1001 · 2025-02-05T02:14:50Z

...ain/java/org/apache/rocketmq/remoting/protocol/header/CheckAsyncTaskStatusRequestHeader.java

+import org.apache.rocketmq.remoting.protocol.RequestCode;
+
+@RocketMQAction(value = RequestCode.CHECK_ASYNC_TASK_STATUS, action = Action.GET)
+public class CheckAsyncTaskStatusRequestHeader implements CommandCustomHeader {


add resourceType for ACL

@RocketMQAction(value = RequestCode.CHECK_ASYNC_TASK_STATUS, resource = ResourceType.CLUSTER, action = Action.GET)

qianye1001 · 2025-02-05T02:19:21Z

tools/src/main/java/org/apache/rocketmq/tools/command/stats/CheckAsyncTaskStatusSubCommand.java

+
+    @Override
+    public Options buildCommandlineOptions(Options options) {
+        Option opt = new Option("t", "taskName", true, "The name of the asynchronous task");


maybe need [-b] to choose single broker and [-n] [-c] to collect status from all brokers in name server or cluster

qianye1001 · 2025-02-05T02:29:15Z

broker/src/main/java/org/apache/rocketmq/broker/AdminAsyncTaskManager.java

+public class AdminAsyncTaskManager {
+
+    // taskId -> AsyncTask
+    private static final Map<String, AsyncTask> ASYNC_TASK_MAP = new ConcurrentHashMap<>();


do not use static, it may conflict with other design, use object and make sure to be used only in admin Process

qianye1001 · 2025-02-05T02:39:15Z

broker/src/main/java/org/apache/rocketmq/broker/AdminAsyncTaskManager.java

+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class AdminAsyncTaskManager {


if using map to manage task, maybe need a thread to clean up the expired task regularly. Using Caffeine LoadingCache may be better

qianye1001 · 2025-02-05T02:48:02Z

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

@@ -487,11 +494,14 @@ private RemotingCommand updateAndGetGroupForbidden(ChannelHandlerContext ctx, Re
    private RemotingCommand checkRocksdbCqWriteProgress(ChannelHandlerContext ctx, RemotingCommand request) {
        CheckRocksdbCqWriteResult result = new CheckRocksdbCqWriteResult();
        result.setCheckStatus(CheckRocksdbCqWriteResult.CheckStatus.CHECK_IN_PROGRESS.getValue());
+        String taskId = AdminAsyncTaskManager.createTask("checkRocksdbCqWriteProgress");


use CompletableFuture when complete to record task success or error, you can Change this runable to a future and add future arg in createTask like this ：

CompletableFuture future = ......
adminAsyncTaskManager.createTask("checkRocksdbCqWriteProgress", future);

qianye1001 · 2025-02-07T06:42:49Z

broker/src/main/java/org/apache/rocketmq/broker/AdminAsyncTaskManager.java

+    public AdminAsyncTaskManager() {
+        this.asyncTaskCache = Caffeine.newBuilder()
+            .expireAfterWrite(30, TimeUnit.MINUTES)
+            .maximumSize(10000)


add removalListener to clean dirty data in taskNameToIdsMap

make expire time and max size configurable, expire time default one day may be better

qianye1001 · 2025-02-07T06:46:18Z

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

            try {
-                CheckRocksdbCqWriteResult checkResult = doCheckRocksdbCqWriteProgress(ctx, request);
-                LOGGER.info("checkRocksdbCqWriteProgress result: {}", JSON.toJSONString(checkResult));
+                return doCheckRocksdbCqWriteProgress(ctx, request);


qianye1001 · 2025-02-07T06:51:01Z

broker/src/main/java/org/apache/rocketmq/broker/AdminAsyncTaskManager.java

+        AsyncTask task = new AsyncTask(taskName, taskId, future);
+
+        asyncTaskCache.put(taskId, task);
+        taskNameToIdsMap.computeIfAbsent(taskName, k -> new ArrayList<>()).add(taskId);


make sure [ add and remove object in ArrayList ] in map.compute to avoid concurrent errors

qianye1001 · 2025-02-07T06:56:07Z

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

-        asyncExecuteWorker.submit(runnable);
+        }, asyncExecuteWorker);
+
+        asyncTaskManager.createTask("checkRocksdbCqWriteProgress", future);


return task id

qianye1001 · 2025-02-07T06:57:28Z

...ain/java/org/apache/rocketmq/remoting/protocol/header/CheckAsyncTaskStatusRequestHeader.java

+@RocketMQAction(value = RequestCode.CHECK_ASYNC_TASK_STATUS, resource = ResourceType.CLUSTER, action = Action.GET)
+public class CheckAsyncTaskStatusRequestHeader implements CommandCustomHeader {
+
+    private String taskName;


Provides a method to accurately query a task

private String taskId;

qianye1001 · 2025-02-07T07:05:16Z

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

+
+    private RemotingCommand checkAsyncTaskStatus(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
+        final CheckAsyncTaskStatusRequestHeader requestHeader = request.decodeCommandCustomHeader(CheckAsyncTaskStatusRequestHeader.class);
+        List<String> taskIds = asyncTaskManager.getTaskIdsByName(requestHeader.getTaskName());


Consider the order of the task

Return the status of the newly created task first?

modified: broker/src/main/java/org/apache/rocketmq/broker/AdminAsyncTaskManager.java modified: broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java modified: common/src/main/java/org/apache/rocketmq/common/CheckRocksdbCqWriteResult.java modified: remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/CheckAsyncTaskStatusRequestHeader.java modified: tools/src/main/java/org/apache/rocketmq/tools/command/queue/CheckRocksdbCqWriteProgressCommand.java modified: tools/src/main/java/org/apache/rocketmq/tools/command/stats/CheckAsyncTaskStatusSubCommand.java modified: tools/src/test/java/org/apache/rocketmq/tools/command/CheckAsyncTaskStatusSubCommandTest.java modified: broker/src/main/java/org/apache/rocketmq/broker/AdminAsyncTaskManager.java modified: broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java modified: common/src/main/java/org/apache/rocketmq/common/CheckRocksdbCqWriteResult.java modified: remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/CheckAsyncTaskStatusRequestHeader.java modified: tools/src/main/java/org/apache/rocketmq/tools/command/queue/CheckRocksdbCqWriteProgressCommand.java modified: tools/src/main/java/org/apache/rocketmq/tools/command/stats/CheckAsyncTaskStatusSubCommand.java modified: tools/src/test/java/org/apache/rocketmq/tools/command/CheckAsyncTaskStatusSubCommandTest.java modified: broker/src/main/java/org/apache/rocketmq/broker/AdminAsyncTaskManager.java modified: broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java modified: common/src/main/java/org/apache/rocketmq/common/CheckRocksdbCqWriteResult.java modified: remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/CheckAsyncTaskStatusRequestHeader.java modified: tools/src/main/java/org/apache/rocketmq/tools/command/queue/CheckRocksdbCqWriteProgressCommand.java modified: tools/src/main/java/org/apache/rocketmq/tools/command/stats/CheckAsyncTaskStatusSubCommand.java modified: tools/src/test/java/org/apache/rocketmq/tools/command/CheckAsyncTaskStatusSubCommandTest.java modified: broker/src/main/java/org/apache/rocketmq/broker/AdminAsyncTaskManager.java modified: broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java modified: common/src/main/java/org/apache/rocketmq/common/CheckRocksdbCqWriteResult.java modified: remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/CheckAsyncTaskStatusRequestHeader.java modified: tools/src/main/java/org/apache/rocketmq/tools/command/queue/CheckRocksdbCqWriteProgressCommand.java modified: tools/src/main/java/org/apache/rocketmq/tools/command/stats/CheckAsyncTaskStatusSubCommand.java modified: tools/src/test/java/org/apache/rocketmq/tools/command/CheckAsyncTaskStatusSubCommandTest.java

KiteSoar force-pushed the develop-#9097 branch 5 times, most recently from 53d1b72 to a4a7703 Compare February 1, 2025 08:27

qianye1001 reviewed Feb 5, 2025

View reviewed changes

[ISSUE apache#9097]Add new command to check async task status in broker.

bf4a370

KiteSoar force-pushed the develop-#9097 branch from 2104751 to bf4a370 Compare February 7, 2025 01:44

qianye1001 reviewed Feb 7, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[ISSUE#9097] Add new command to check async task status in broker. #9162

[ISSUE#9097] Add new command to check async task status in broker. #9162

KiteSoar commented Jan 27, 2025

codecov-commenter commented Feb 1, 2025 •

edited

Loading

qianye1001 Feb 5, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 5, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 5, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 5, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 5, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 7, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 7, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 7, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 7, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 7, 2025

KiteSoar Feb 7, 2025

qianye1001 Feb 7, 2025

KiteSoar Feb 7, 2025

KiteSoar Feb 7, 2025

[ISSUE#9097] Add new command to check async task status in broker. #9162

Are you sure you want to change the base?

[ISSUE#9097] Add new command to check async task status in broker. #9162

Conversation

KiteSoar commented Jan 27, 2025

Which Issue(s) This PR Fixes

Brief Description

How Did You Test This Change?

codecov-commenter commented Feb 1, 2025 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Feb 1, 2025 •

edited

Loading