This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 1196e7124 fix(instance): apply a single deadline to resource counts
(#2525)
1196e7124 is described below
commit 1196e7124f679b313b413ddbdce2589030253a4d
Author: 0 <[email protected]>
AuthorDate: Sat Aug 22 15:30:59 2026 +0800
fix(instance): apply a single deadline to resource counts (#2525)
fillCountsInParallel submitted one task per instance and then waited on
each future with its own 3s get(). A batch of hung vendor calls therefore cost
3s * instance-count of response latency.
Submit the whole batch through invokeAll with one shared deadline: the
request waits at most COUNT_TIMEOUT_SECONDS in total, the tasks that miss the
deadline are cancelled (interrupted) by invokeAll and their rows are marked
counts-unavailable like any other failure.
Fixes #2495
---
.../rocketmq/studio/instance/InstanceService.java | 50 ++++++++++++++--------
.../studio/instance/InstanceServiceTest.java | 29 +++++++++++++
2 files changed, 61 insertions(+), 18 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
index b31464a3b..6a7b48336 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
@@ -47,12 +47,12 @@ import java.util.Comparator;
import java.util.List;
import java.util.Locale;
import java.util.Objects;
+import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
@Slf4j
@Service
@@ -116,27 +116,41 @@ public class InstanceService {
if (instances.isEmpty()) {
return;
}
- List<InstanceVO> pending = new ArrayList<>(instances);
- List<Future<?>> futures = new ArrayList<>(instances.size());
- for (InstanceVO instance : pending) {
- futures.add(countExecutor.submit(() -> fillCounts(instance)));
+ List<Callable<Void>> tasks = instances.stream()
+ .<Callable<Void>>map(instance -> () -> {
+ fillCounts(instance);
+ return null;
+ })
+ .toList();
+ List<Future<Void>> futures;
+ try {
+ // A single deadline for the whole batch. Waiting per future would
let one hung
+ // vendor add COUNT_TIMEOUT_SECONDS to the response for every
instance.
+ futures = countExecutor.invokeAll(tasks, COUNT_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ instances.forEach(instance ->
instance.setResourceCountsAvailable(false));
+ return;
}
- for (int i = 0; i < pending.size(); i++) {
- InstanceVO instance = pending.get(i);
- try {
- futures.get(i).get(COUNT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- } catch (TimeoutException ex) {
- futures.get(i).cancel(true);
+ for (int i = 0; i < futures.size(); i++) {
+ InstanceVO instance = instances.get(i);
+ Future<Void> future = futures.get(i);
+ if (future.isCancelled()) {
+ // Missed the shared deadline; invokeAll already interrupted
the task.
instance.setResourceCountsAvailable(false);
log.warn("Resource counts timed out after {}s for instance {}",
COUNT_TIMEOUT_SECONDS, instance.getId());
- } catch (InterruptedException ex) {
- Thread.currentThread().interrupt();
- instance.setResourceCountsAvailable(false);
- } catch (ExecutionException ex) {
- instance.setResourceCountsAvailable(false);
- log.warn("Failed to load resource counts for instance {}: {}",
- instance.getId(), ex.getMessage());
+ } else {
+ try {
+ future.get();
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ instance.setResourceCountsAvailable(false);
+ } catch (ExecutionException ex) {
+ instance.setResourceCountsAvailable(false);
+ log.warn("Failed to load resource counts for instance {}:
{}",
+ instance.getId(), ex.getMessage());
+ }
}
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
index 405569b9f..d490791f3 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
@@ -41,6 +41,7 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.time.LocalDateTime;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Locale;
@@ -49,6 +50,7 @@ import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
@@ -289,6 +291,33 @@ class InstanceServiceTest {
assertThat(result.get(0).getConsumerGroupCount()).isZero();
}
+ @Test
+ void listInstancesShouldApplyASingleDeadlineToSlowCountsTest() throws
InterruptedException {
+ // Three hung providers: waiting per future would cost 3s each (9s
total); a shared
+ // deadline must bound the whole batch to roughly one timeout.
+ List<InstanceVO> slow = new ArrayList<>();
+ for (long id = 20L; id < 23L; id++) {
+ InstanceVO vo = InstanceVO.builder().name("slow-" + id).build();
+ vo.setId(id);
+ slow.add(vo);
+ }
+ when(instanceRepository.findAll()).thenReturn(slow);
+
when(providerRegistry.forVendor(InstanceVendor.APACHE)).thenReturn(instanceProvider);
+ when(instanceProvider.countTopics(anyString())).thenAnswer(invocation
-> {
+ Thread.sleep(10_000);
+ return 0;
+ });
+
+ long start = System.nanoTime();
+ List<InstanceVO> result = instanceService.listInstances(null, null);
+ long elapsedMillis = (System.nanoTime() - start) / 1_000_000;
+
+ assertThat(elapsedMillis)
+ .as("batch must not wait one timeout per instance")
+ .isLessThan(2L * InstanceService.COUNT_TIMEOUT_SECONDS * 1000);
+ assertThat(result).allSatisfy(vo ->
assertThat(vo.isResourceCountsAvailable()).isFalse());
+ }
+
@Test
void createInstanceShouldThrowWhenRequestIsNull() {
assertThatThrownBy(() -> instanceService.createInstance(null))
