RockteMQ-AI commented on code in PR #3168:
URL:
https://github.com/apache/rocketmq-dashboard/pull/3168#discussion_r3937761250
##########
server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java:
##########
@@ -122,13 +141,38 @@ private PageResult<DLQGroupVO> listDLQGroups(MQAdminExt
adminExt, String search,
long offset = Pagination.pageOffset(page, pageSize);
int from = (int) Math.min(offset, dlqTopics.size());
int to = (int) Math.min(offset + pageSize, dlqTopics.size());
- List<DLQGroupVO> groups = dlqTopics.subList(from, to).stream()
- .map(topic -> buildDLQGroup(adminExt,
-
topic.substring(MixAll.DLQ_GROUP_TOPIC_PREFIX.length()), topic))
- .toList();
+ List<String> pageTopics = dlqTopics.subList(from, to);
+ List<DLQGroupVO> groups = loadGroupStatsInParallel(adminExt,
pageTopics);
return PageResult.of(groups, dlqTopics.size(), page, pageSize);
}
+ private List<DLQGroupVO> loadGroupStatsInParallel(MQAdminExt adminExt,
List<String> pageTopics) {
+ List<Future<DLQGroupVO>> futures = new ArrayList<>(pageTopics.size());
+ for (String topic : pageTopics) {
+ String groupName =
topic.substring(MixAll.DLQ_GROUP_TOPIC_PREFIX.length());
+ futures.add(statsExecutor.submit(() -> buildDLQGroup(adminExt,
groupName, topic)));
+ }
+ List<DLQGroupVO> groups = new ArrayList<>(futures.size());
+ for (Future<DLQGroupVO> future : futures) {
+ groups.add(awaitGroupStats(future));
+ }
+ return groups;
+ }
+
+ private DLQGroupVO awaitGroupStats(Future<DLQGroupVO> future) {
+ try {
+ return future.get(STATS_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ } catch (InterruptedException e) {
+ future.cancel(true);
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException e) {
+ // buildDLQGroup reports single-topic failures inside the row
itself.
Review Comment:
When `awaitGroupStats` falls back (timeout / interrupt / execution
exception), the returned `DLQGroupVO` has no `groupName` or `dlqTopic` set —
the user will see a blank row in the DLQ table with only `UNAVAILABLE` status.
Consider passing `groupName` and `dlqTopic` into `awaitGroupStats` so the
fallback can populate them:
```java
private DLQGroupVO awaitGroupStats(Future<DLQGroupVO> future, String
groupName, String dlqTopic) {
try {
return future.get(STATS_TIMEOUT_SECONDS, TimeUnit.SECONDS);
} catch (TimeoutException e) {
future.cancel(true);
} catch (InterruptedException e) {
future.cancel(true);
Thread.currentThread().interrupt();
} catch (ExecutionException e) {
// single-topic failure
}
return DLQGroupVO.builder()
.groupName(groupName).dlqTopic(dlqTopic)
.statsAvailable(false).status("UNAVAILABLE").build();
}
```
And update `loadGroupStatsInParallel` to pass them through.
##########
server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java:
##########
@@ -122,13 +141,38 @@ private PageResult<DLQGroupVO> listDLQGroups(MQAdminExt
adminExt, String search,
long offset = Pagination.pageOffset(page, pageSize);
int from = (int) Math.min(offset, dlqTopics.size());
int to = (int) Math.min(offset + pageSize, dlqTopics.size());
- List<DLQGroupVO> groups = dlqTopics.subList(from, to).stream()
- .map(topic -> buildDLQGroup(adminExt,
-
topic.substring(MixAll.DLQ_GROUP_TOPIC_PREFIX.length()), topic))
- .toList();
+ List<String> pageTopics = dlqTopics.subList(from, to);
+ List<DLQGroupVO> groups = loadGroupStatsInParallel(adminExt,
pageTopics);
return PageResult.of(groups, dlqTopics.size(), page, pageSize);
}
+ private List<DLQGroupVO> loadGroupStatsInParallel(MQAdminExt adminExt,
List<String> pageTopics) {
+ List<Future<DLQGroupVO>> futures = new ArrayList<>(pageTopics.size());
+ for (String topic : pageTopics) {
+ String groupName =
topic.substring(MixAll.DLQ_GROUP_TOPIC_PREFIX.length());
+ futures.add(statsExecutor.submit(() -> buildDLQGroup(adminExt,
groupName, topic)));
+ }
+ List<DLQGroupVO> groups = new ArrayList<>(futures.size());
+ for (Future<DLQGroupVO> future : futures) {
+ groups.add(awaitGroupStats(future));
+ }
+ return groups;
+ }
+
+ private DLQGroupVO awaitGroupStats(Future<DLQGroupVO> future) {
+ try {
+ return future.get(STATS_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ } catch (InterruptedException e) {
+ future.cancel(true);
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException e) {
+ // buildDLQGroup reports single-topic failures inside the row
itself.
+ }
Review Comment:
The `ExecutionException` catch block silently swallows the underlying
exception. Consider adding a `log.debug` or `log.warn` so operators can
diagnose slow/failing brokers:
```java
} catch (ExecutionException e) {
log.warn("Failed to load DLQ stats for topic, row marked UNAVAILABLE",
e.getCause());
}
```
--
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]
