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 5f2566531 fix(instance): bound resource count job lifetime (#3044)
5f2566531 is described below
commit 5f256653114c56fca489e851f751becd3b3b2308
Author: xdz997 <[email protected]>
AuthorDate: Fri Sep 4 15:46:33 2026 +0800
fix(instance): bound resource count job lifetime (#3044)
---
.../instance/InstanceResourceCountRunner.java | 251 +++++++++++++++++++++
.../rocketmq/studio/instance/InstanceService.java | 121 +++++-----
.../instance/InstanceResourceCountRunnerTest.java | 251 +++++++++++++++++++++
.../studio/instance/InstanceServiceTest.java | 77 +++++++
4 files changed, 645 insertions(+), 55 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunner.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunner.java
new file mode 100644
index 000000000..319b01e58
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunner.java
@@ -0,0 +1,251 @@
+/*
+ * 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.studio.instance;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Runs instance resource-count lookups without exposing response objects to
worker threads.
+ *
+ * <p>The executor and its queue are both bounded because vendor SDK calls may
ignore thread
+ * interruption while an HTTP request is in flight. Each batch has one
absolute deadline. A
+ * result is usable only when the worker completed it before that deadline;
late results remain
+ * detached and therefore cannot mutate a response that has already been
returned.</p>
+ */
+final class InstanceResourceCountRunner implements AutoCloseable {
+
+ private static final AtomicInteger THREAD_SEQUENCE = new AtomicInteger();
+
+ private final ThreadPoolExecutor executor;
+ private final long timeoutNanos;
+
+ InstanceResourceCountRunner(int parallelism, int queueCapacity, long
timeout, TimeUnit timeoutUnit) {
+ if (parallelism <= 0) {
+ throw new IllegalArgumentException("parallelism must be positive");
+ }
+ if (queueCapacity <= 0) {
+ throw new IllegalArgumentException("queueCapacity must be
positive");
+ }
+ if (timeout <= 0) {
+ throw new IllegalArgumentException("timeout must be positive");
+ }
+ Objects.requireNonNull(timeoutUnit, "timeoutUnit");
+
+ this.timeoutNanos = timeoutUnit.toNanos(timeout);
+ ThreadFactory threadFactory = runnable -> {
+ Thread thread = new Thread(runnable,
+ "instance-resource-counts-" +
THREAD_SEQUENCE.incrementAndGet());
+ thread.setDaemon(true);
+ return thread;
+ };
+ this.executor = new ThreadPoolExecutor(
+ parallelism,
+ parallelism,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new ArrayBlockingQueue<>(queueCapacity),
+ threadFactory,
+ new ThreadPoolExecutor.AbortPolicy());
+ }
+
+ @FunctionalInterface
+ interface CountFunction {
+ ResourceCounts count(InstanceVO instance) throws Exception;
+ }
+
+ record ResourceCounts(int topicCount, int consumerGroupCount) {
+ }
+
+ enum OutcomeStatus {
+ SUCCESS,
+ FAILED,
+ TIMED_OUT,
+ REJECTED,
+ INTERRUPTED
+ }
+
+ record CountOutcome(OutcomeStatus status, ResourceCounts counts, Throwable
failure) {
+
+ static CountOutcome success(ResourceCounts counts) {
+ return new CountOutcome(OutcomeStatus.SUCCESS, counts, null);
+ }
+
+ static CountOutcome failed(Throwable failure) {
+ return new CountOutcome(OutcomeStatus.FAILED, null, failure);
+ }
+
+ static CountOutcome unavailable(OutcomeStatus status) {
+ return new CountOutcome(status, null, null);
+ }
+
+ boolean available() {
+ return status == OutcomeStatus.SUCCESS;
+ }
+ }
+
+ List<CountOutcome> countAll(List<InstanceVO> instances, CountFunction
function) {
+ Objects.requireNonNull(instances, "instances");
+ Objects.requireNonNull(function, "function");
+ if (instances.isEmpty()) {
+ return List.of();
+ }
+
+ long startedNanos = System.nanoTime();
+ List<SubmittedCount> submitted = submitAll(instances, function);
+ List<CountOutcome> outcomes = new ArrayList<>(submitted.size());
+ boolean interrupted = false;
+
+ for (SubmittedCount count : submitted) {
+ if (count.rejected()) {
+ outcomes.add(CountOutcome.unavailable(OutcomeStatus.REJECTED));
+ continue;
+ }
+ if (interrupted) {
+ count.future().cancel(true);
+
outcomes.add(CountOutcome.unavailable(OutcomeStatus.INTERRUPTED));
+ continue;
+ }
+
+ try {
+ CompletedCount completed = await(count.future(), startedNanos);
+ if (elapsedNanos(startedNanos, completed.completedNanos()) >
timeoutNanos) {
+
outcomes.add(CountOutcome.unavailable(OutcomeStatus.TIMED_OUT));
+ } else if (completed.failure() != null) {
+ outcomes.add(CountOutcome.failed(completed.failure()));
+ } else if (completed.counts() == null) {
+ outcomes.add(CountOutcome.failed(
+ new IllegalStateException("resource count provider
returned null")));
+ } else {
+ outcomes.add(CountOutcome.success(completed.counts()));
+ }
+ } catch (TimeoutException exception) {
+ count.future().cancel(true);
+
outcomes.add(CountOutcome.unavailable(OutcomeStatus.TIMED_OUT));
+ } catch (InterruptedException exception) {
+ count.future().cancel(true);
+ interrupted = true;
+
outcomes.add(CountOutcome.unavailable(OutcomeStatus.INTERRUPTED));
+ } catch (CancellationException exception) {
+
outcomes.add(CountOutcome.unavailable(OutcomeStatus.INTERRUPTED));
+ } catch (ExecutionException exception) {
+ outcomes.add(CountOutcome.failed(rootCause(exception)));
+ }
+ }
+
+ cancelOutstanding(submitted);
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ // Future.cancel does not remove queued FutureTasks from a
ThreadPoolExecutor queue.
+ // Purging prevents timed-out requests from leaving a cancelled
backlog behind.
+ executor.purge();
+ return List.copyOf(outcomes);
+ }
+
+ private List<SubmittedCount> submitAll(List<InstanceVO> instances,
CountFunction function) {
+ List<SubmittedCount> submitted = new ArrayList<>(instances.size());
+ for (InstanceVO instance : instances) {
+ try {
+ Future<CompletedCount> future = executor.submit(() ->
execute(instance, function));
+ submitted.add(SubmittedCount.accepted(future));
+ } catch (RejectedExecutionException exception) {
+ submitted.add(SubmittedCount.rejectedSubmission());
+ }
+ }
+ return submitted;
+ }
+
+ private static CompletedCount execute(InstanceVO instance, CountFunction
function) {
+ try {
+ return new CompletedCount(function.count(instance), null,
System.nanoTime());
+ } catch (Exception exception) {
+ return new CompletedCount(null, exception, System.nanoTime());
+ }
+ }
+
+ private CompletedCount await(Future<CompletedCount> future, long
startedNanos)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ if (future.isDone()) {
+ return future.get();
+ }
+ long remainingNanos = timeoutNanos - elapsedNanos(startedNanos,
System.nanoTime());
+ if (remainingNanos <= 0L) {
+ throw new TimeoutException("resource count batch deadline
reached");
+ }
+ return future.get(remainingNanos, TimeUnit.NANOSECONDS);
+ }
+
+ private static void cancelOutstanding(List<SubmittedCount> submitted) {
+ for (SubmittedCount count : submitted) {
+ if (!count.rejected() && !count.future().isDone()) {
+ count.future().cancel(true);
+ }
+ }
+ }
+
+ int activeCount() {
+ return executor.getActiveCount();
+ }
+
+ int queuedCount() {
+ return executor.getQueue().size();
+ }
+
+ @Override
+ public void close() {
+ executor.shutdownNow();
+ executor.purge();
+ }
+
+ private static long elapsedNanos(long startedNanos, long completedNanos) {
+ return completedNanos - startedNanos;
+ }
+
+ private static Throwable rootCause(Throwable error) {
+ Throwable current = error;
+ while (current.getCause() != null && current.getCause() != current) {
+ current = current.getCause();
+ }
+ return current;
+ }
+
+ private record CompletedCount(ResourceCounts counts, Throwable failure,
long completedNanos) {
+ }
+
+ private record SubmittedCount(Future<CompletedCount> future, boolean
rejected) {
+
+ static SubmittedCount accepted(Future<CompletedCount> future) {
+ return new SubmittedCount(future, false);
+ }
+
+ static SubmittedCount rejectedSubmission() {
+ return new SubmittedCount(null, true);
+ }
+ }
+}
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 e1a2faec1..f8de5b996 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
@@ -52,11 +52,6 @@ import java.util.List;
import java.util.Locale;
import java.util.Objects;
import java.util.Set;
-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;
@Slf4j
@@ -74,20 +69,18 @@ public class InstanceService {
private final RegionNames regionNames;
static final int COUNT_PARALLELISM = 8;
+ static final int COUNT_QUEUE_CAPACITY = 128;
static final long COUNT_TIMEOUT_SECONDS = 3;
private static final int MAX_BATCH_FAILURE_MESSAGE_LENGTH = 500;
static final int MAX_CLOUD_IMPORT_FAILURE_DETAILS = 100;
static final int MAX_CLOUD_IMPORT_FAILURE_MESSAGE_LENGTH = 500;
- private final ExecutorService countExecutor =
Executors.newFixedThreadPool(COUNT_PARALLELISM, runnable -> {
- Thread thread = new Thread(runnable, "instance-resource-counts");
- thread.setDaemon(true);
- return thread;
- });
+ private final InstanceResourceCountRunner countRunner = new
InstanceResourceCountRunner(
+ COUNT_PARALLELISM, COUNT_QUEUE_CAPACITY, COUNT_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
@PreDestroy
- void shutdownCountExecutor() {
- countExecutor.shutdownNow();
+ void shutdownCountRunner() {
+ countRunner.close();
}
public List<InstanceVO> listInstances(InstanceType type, String search) {
@@ -125,61 +118,79 @@ public class InstanceService {
if (instances.isEmpty()) {
return;
}
- 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 < 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());
- } else {
- try {
- future.get();
- } catch (InterruptedException ex) {
- Thread.currentThread().interrupt();
- instance.setResourceCountsAvailable(false);
- } catch (ExecutionException ex) {
- instance.setResourceCountsAvailable(false);
+ List<InstanceResourceCountRunner.CountOutcome> outcomes =
+ countRunner.countAll(instances, this::loadCounts);
+ int rejected = 0;
+ int timedOut = 0;
+ int failed = 0;
+ for (int index = 0; index < instances.size(); index++) {
+ InstanceVO instance = instances.get(index);
+ InstanceResourceCountRunner.CountOutcome outcome =
outcomes.get(index);
+ if (outcome.available()) {
+ applyCounts(instance, outcome.counts());
+ continue;
+ }
+
+ clearCounts(instance);
+ switch (outcome.status()) {
+ case REJECTED -> rejected++;
+ case TIMED_OUT -> timedOut++;
+ case FAILED -> {
+ failed++;
log.warn("Failed to load resource counts for instance {}:
{}",
- instance.getId(), ex.getMessage());
+ instance.getId(), rootMessage(outcome.failure()));
}
+ case INTERRUPTED -> log.debug(
+ "Resource count lookup interrupted for instance {}",
instance.getId());
+ case SUCCESS -> throw new IllegalStateException("available
count outcome has no values");
}
}
+ if (timedOut > 0) {
+ log.warn("Resource count deadline reached after {}s for {}
instance(s)",
+ COUNT_TIMEOUT_SECONDS, timedOut);
+ }
+ if (rejected > 0) {
+ log.warn("Resource count executor saturated; {} instance(s) marked
unavailable", rejected);
+ }
+ if (failed > 0) {
+ log.debug("Resource count lookup failed for {} instance(s)",
failed);
+ }
}
/**
* Resource counts live on the vendor side (cloud APIs) or in the local
tables (Apache),
* so resolve them uniformly through the vendor provider.
*/
- private void fillCounts(InstanceVO instance) {
+ private InstanceResourceCountRunner.ResourceCounts loadCounts(InstanceVO
instance) {
InstanceVendor vendor = instance.getVendor() == null ?
InstanceVendor.APACHE : instance.getVendor();
- try {
- InstanceProvider provider = providerRegistry.forVendor(vendor);
-
instance.setTopicCount(provider.countTopics(String.valueOf(instance.getId())));
-
instance.setConsumerGroupCount(provider.countGroups(String.valueOf(instance.getId())));
- instance.setResourceCountsAvailable(true);
- } catch (RuntimeException ex) {
- instance.setResourceCountsAvailable(false);
- log.warn("Failed to load resource counts for instance {}: {}",
- instance.getId(), ex.getMessage());
+ InstanceProvider provider = providerRegistry.forVendor(vendor);
+ int topicCount =
provider.countTopics(String.valueOf(instance.getId()));
+ int consumerGroupCount =
provider.countGroups(String.valueOf(instance.getId()));
+ return new InstanceResourceCountRunner.ResourceCounts(topicCount,
consumerGroupCount);
+ }
+
+ private static void applyCounts(InstanceVO instance,
InstanceResourceCountRunner.ResourceCounts counts) {
+ instance.setTopicCount(counts.topicCount());
+ instance.setConsumerGroupCount(counts.consumerGroupCount());
+ instance.setResourceCountsAvailable(true);
+ }
+
+ private static void clearCounts(InstanceVO instance) {
+ instance.setTopicCount(0);
+ instance.setConsumerGroupCount(0);
+ instance.setResourceCountsAvailable(false);
+ }
+
+ private static String rootMessage(Throwable failure) {
+ if (failure == null) {
+ return "unknown failure";
+ }
+ Throwable current = failure;
+ while (current.getCause() != null && current.getCause() != current) {
+ current = current.getCause();
}
+ String message = current.getMessage();
+ return StringUtils.hasText(message) ? message :
current.getClass().getSimpleName();
}
public InstanceVO createInstance(InstanceVO instance) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunnerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunnerTest.java
new file mode 100644
index 000000000..4fab0ebc5
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceResourceCountRunnerTest.java
@@ -0,0 +1,251 @@
+/*
+ * 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.studio.instance;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class InstanceResourceCountRunnerTest {
+
+ @Test
+ void
countAllShouldRunIndependentLookupsConcurrentlyAndKeepInputOrderTest() throws
Exception {
+ InstanceVO first = instance(1L);
+ InstanceVO second = instance(2L);
+ CountDownLatch bothStarted = new CountDownLatch(2);
+ CountDownLatch releaseFirst = new CountDownLatch(1);
+
+ try (InstanceResourceCountRunner runner =
+ new InstanceResourceCountRunner(2, 2, 2,
TimeUnit.SECONDS)) {
+ CompletableFuture<List<InstanceResourceCountRunner.CountOutcome>>
call =
+ CompletableFuture.supplyAsync(() ->
runner.countAll(List.of(first, second), instance -> {
+ bothStarted.countDown();
+ if (instance.getId().equals(1L)) {
+ releaseFirst.await();
+ }
+ int id = instance.getId().intValue();
+ return new
InstanceResourceCountRunner.ResourceCounts(id * 10, id * 100);
+ }));
+
+ assertThat(bothStarted.await(1, TimeUnit.SECONDS)).isTrue();
+ releaseFirst.countDown();
+
+ List<InstanceResourceCountRunner.CountOutcome> outcomes =
call.get(1, TimeUnit.SECONDS);
+
assertThat(outcomes).extracting(InstanceResourceCountRunner.CountOutcome::status)
+ .containsExactly(
+ InstanceResourceCountRunner.OutcomeStatus.SUCCESS,
+ InstanceResourceCountRunner.OutcomeStatus.SUCCESS);
+ assertThat(outcomes).extracting(outcome ->
outcome.counts().topicCount())
+ .containsExactly(10, 20);
+ } finally {
+ releaseFirst.countDown();
+ }
+ }
+
+ @Test
+ void countAllShouldDiscardAResultThatFinishesAfterTheBatchDeadlineTest()
throws Exception {
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch finished = new CountDownLatch(1);
+
+ try (InstanceResourceCountRunner runner =
+ new InstanceResourceCountRunner(1, 1, 100,
TimeUnit.MILLISECONDS)) {
+ long startedAt = System.nanoTime();
+ List<InstanceResourceCountRunner.CountOutcome> outcomes =
runner.countAll(
+ List.of(instance(1L)), ignored -> {
+ started.countDown();
+ awaitIgnoringInterrupt(release);
+ finished.countDown();
+ return new
InstanceResourceCountRunner.ResourceCounts(7, 5);
+ });
+ long elapsedMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAt);
+
+ assertThat(started.await(1, TimeUnit.SECONDS)).isTrue();
+ assertThat(elapsedMillis).isLessThan(1_000);
+ assertThat(outcomes).singleElement().satisfies(outcome -> {
+
assertThat(outcome.status()).isEqualTo(InstanceResourceCountRunner.OutcomeStatus.TIMED_OUT);
+ assertThat(outcome.counts()).isNull();
+ });
+
+ release.countDown();
+ assertThat(finished.await(1, TimeUnit.SECONDS)).isTrue();
+ assertThat(outcomes).singleElement().satisfies(outcome ->
+ assertThat(outcome.status()).isEqualTo(
+
InstanceResourceCountRunner.OutcomeStatus.TIMED_OUT));
+ } finally {
+ release.countDown();
+ }
+ }
+
+ @Test
+ void countAllShouldRejectRowsBeyondTheBoundedQueueTest() throws Exception {
+ List<InstanceVO> instances = List.of(instance(1L), instance(2L),
instance(3L));
+ CountDownLatch firstStarted = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+
+ try (InstanceResourceCountRunner runner =
+ new InstanceResourceCountRunner(1, 1, 2,
TimeUnit.SECONDS)) {
+ CompletableFuture<List<InstanceResourceCountRunner.CountOutcome>>
call =
+ CompletableFuture.supplyAsync(() ->
runner.countAll(instances, instance -> {
+ firstStarted.countDown();
+ release.await();
+ int id = instance.getId().intValue();
+ return new
InstanceResourceCountRunner.ResourceCounts(id, id);
+ }));
+
+ assertThat(firstStarted.await(1, TimeUnit.SECONDS)).isTrue();
+ assertThat(waitForQueueSize(runner, 1, 1,
TimeUnit.SECONDS)).isTrue();
+ assertThat(runner.queuedCount()).isLessThanOrEqualTo(1);
+ release.countDown();
+
+ List<InstanceResourceCountRunner.CountOutcome> outcomes =
call.get(1, TimeUnit.SECONDS);
+
assertThat(outcomes).extracting(InstanceResourceCountRunner.CountOutcome::status)
+ .containsExactly(
+ InstanceResourceCountRunner.OutcomeStatus.SUCCESS,
+ InstanceResourceCountRunner.OutcomeStatus.SUCCESS,
+
InstanceResourceCountRunner.OutcomeStatus.REJECTED);
+ } finally {
+ release.countDown();
+ }
+ }
+
+ @Test
+ void countAllShouldUseOneDeadlineForTheWholeBatchTest() {
+ CountDownLatch release = new CountDownLatch(1);
+
+ try (InstanceResourceCountRunner runner =
+ new InstanceResourceCountRunner(2, 2, 150,
TimeUnit.MILLISECONDS)) {
+ long startedAt = System.nanoTime();
+ List<InstanceResourceCountRunner.CountOutcome> outcomes =
runner.countAll(
+ List.of(instance(1L), instance(2L)), ignored -> {
+ release.await();
+ return new
InstanceResourceCountRunner.ResourceCounts(1, 1);
+ });
+ long elapsedMillis =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAt);
+
+ assertThat(elapsedMillis).isLessThan(1_000);
+
assertThat(outcomes).extracting(InstanceResourceCountRunner.CountOutcome::status)
+
.containsOnly(InstanceResourceCountRunner.OutcomeStatus.TIMED_OUT);
+ } finally {
+ release.countDown();
+ }
+ }
+
+ @Test
+ void countAllShouldKeepFailuresAndNullResultsLocalToTheirRowsTest() {
+ try (InstanceResourceCountRunner runner =
+ new InstanceResourceCountRunner(3, 3, 1,
TimeUnit.SECONDS)) {
+ List<InstanceResourceCountRunner.CountOutcome> outcomes =
runner.countAll(
+ List.of(instance(1L), instance(2L), instance(3L)),
instance -> {
+ if (instance.getId().equals(1L)) {
+ throw new IllegalStateException("provider
unavailable");
+ }
+ if (instance.getId().equals(2L)) {
+ return null;
+ }
+ return new
InstanceResourceCountRunner.ResourceCounts(3, 4);
+ });
+
+
assertThat(outcomes).extracting(InstanceResourceCountRunner.CountOutcome::status)
+ .containsExactly(
+ InstanceResourceCountRunner.OutcomeStatus.FAILED,
+ InstanceResourceCountRunner.OutcomeStatus.FAILED,
+ InstanceResourceCountRunner.OutcomeStatus.SUCCESS);
+ assertThat(outcomes.get(0).failure()).hasMessage("provider
unavailable");
+ assertThat(outcomes.get(1).failure()).hasMessage("resource count
provider returned null");
+ assertThat(outcomes.get(2).counts())
+ .isEqualTo(new
InstanceResourceCountRunner.ResourceCounts(3, 4));
+ }
+ }
+
+ @Test
+ void countAllShouldCancelOutstandingWorkAndRestoreCallerInterruptTest()
throws Exception {
+ CountDownLatch workerStarted = new CountDownLatch(1);
+ CountDownLatch workerInterrupted = new CountDownLatch(1);
+ AtomicReference<List<InstanceResourceCountRunner.CountOutcome>>
outcomes = new AtomicReference<>();
+ AtomicBoolean interruptRestored = new AtomicBoolean();
+
+ try (InstanceResourceCountRunner runner =
+ new InstanceResourceCountRunner(1, 2, 10,
TimeUnit.SECONDS)) {
+ Thread caller = new Thread(() -> {
+ outcomes.set(runner.countAll(List.of(instance(1L),
instance(2L)), ignored -> {
+ workerStarted.countDown();
+ try {
+ Thread.sleep(10_000);
+ } catch (InterruptedException exception) {
+ workerInterrupted.countDown();
+ throw exception;
+ }
+ return new InstanceResourceCountRunner.ResourceCounts(1,
1);
+ }));
+ interruptRestored.set(Thread.currentThread().isInterrupted());
+ });
+ caller.start();
+
+ assertThat(workerStarted.await(1, TimeUnit.SECONDS)).isTrue();
+ caller.interrupt();
+ caller.join(1_000);
+
+ assertThat(caller.isAlive()).isFalse();
+ assertThat(workerInterrupted.await(1, TimeUnit.SECONDS)).isTrue();
+ assertThat(interruptRestored).isTrue();
+
assertThat(outcomes.get()).extracting(InstanceResourceCountRunner.CountOutcome::status)
+
.containsOnly(InstanceResourceCountRunner.OutcomeStatus.INTERRUPTED);
+ assertThat(runner.queuedCount()).isZero();
+ }
+ }
+
+ private static InstanceVO instance(long id) {
+ InstanceVO instance = InstanceVO.builder().name("instance-" +
id).build();
+ instance.setId(id);
+ return instance;
+ }
+
+ private static void awaitIgnoringInterrupt(CountDownLatch latch) {
+ boolean interrupted = false;
+ while (true) {
+ try {
+ latch.await();
+ break;
+ } catch (InterruptedException exception) {
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private static boolean waitForQueueSize(InstanceResourceCountRunner
runner, int expected,
+ long timeout, TimeUnit unit)
throws InterruptedException {
+ long deadline = System.nanoTime() + unit.toNanos(timeout);
+ while (System.nanoTime() < deadline) {
+ if (runner.queuedCount() == expected) {
+ return true;
+ }
+ Thread.sleep(5);
+ }
+ return runner.queuedCount() == expected;
+ }
+}
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 9367ab4c9..6ef5a5277 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
@@ -48,6 +48,8 @@ import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -297,6 +299,25 @@ class InstanceServiceTest {
assertThat(result.get(0).getConsumerGroupCount()).isZero();
}
+ @Test
+ void listInstancesShouldClearPreviouslyLoadedCountsWhenRefreshFailsTest() {
+ InstanceVO instance =
InstanceVO.builder().name("stale-counts").build();
+ instance.setId(6L);
+ instance.setTopicCount(12);
+ instance.setConsumerGroupCount(8);
+ instance.setResourceCountsAvailable(true);
+ when(instanceRepository.findAll()).thenReturn(List.of(instance));
+
when(providerRegistry.forVendor(InstanceVendor.APACHE)).thenReturn(instanceProvider);
+ when(instanceProvider.countTopics("6"))
+ .thenThrow(new IllegalStateException("provider unavailable"));
+
+ InstanceVO result = instanceService.listInstances(null, null).get(0);
+
+ assertThat(result.isResourceCountsAvailable()).isFalse();
+ assertThat(result.getTopicCount()).isZero();
+ assertThat(result.getConsumerGroupCount()).isZero();
+ }
+
@Test
void listInstancesShouldApplyASingleDeadlineToSlowCountsTest() throws
InterruptedException {
// Three hung providers: waiting per future would cost 3s each (9s
total); a shared
@@ -324,6 +345,62 @@ class InstanceServiceTest {
assertThat(result).allSatisfy(vo ->
assertThat(vo.isResourceCountsAvailable()).isFalse());
}
+ @Test
+ void
timedOutCountTaskShouldNotMutateReturnedInstanceAfterRequestCompletesTest()
throws Exception {
+ InstanceVO slow = InstanceVO.builder().name("slow").build();
+ slow.setId(24L);
+ when(instanceRepository.findAll()).thenReturn(List.of(slow));
+
when(providerRegistry.forVendor(InstanceVendor.APACHE)).thenReturn(instanceProvider);
+
+ CountDownLatch topicCountStarted = new CountDownLatch(1);
+ CountDownLatch releaseProvider = new CountDownLatch(1);
+ CountDownLatch providerFinished = new CountDownLatch(1);
+ when(instanceProvider.countTopics("24")).thenAnswer(invocation -> {
+ topicCountStarted.countDown();
+ boolean interrupted = false;
+ while (true) {
+ try {
+ releaseProvider.await();
+ break;
+ } catch (InterruptedException exception) {
+ // Simulate a vendor SDK that does not abort an in-flight
HTTP request
+ // when Future.cancel(true) interrupts the worker.
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ return 7;
+ });
+ when(instanceProvider.countGroups("24")).thenAnswer(invocation -> {
+ providerFinished.countDown();
+ return 5;
+ });
+
+ try {
+ List<InstanceVO> result = instanceService.listInstances(null,
null);
+
+ assertThat(topicCountStarted.await(1, TimeUnit.SECONDS)).isTrue();
+ assertThat(result).singleElement().satisfies(instance -> {
+ assertThat(instance.isResourceCountsAvailable()).isFalse();
+ assertThat(instance.getTopicCount()).isZero();
+ assertThat(instance.getConsumerGroupCount()).isZero();
+ });
+
+ releaseProvider.countDown();
+ assertThat(providerFinished.await(1, TimeUnit.SECONDS)).isTrue();
+
+ assertThat(result).singleElement().satisfies(instance -> {
+ assertThat(instance.isResourceCountsAvailable()).isFalse();
+ assertThat(instance.getTopicCount()).isZero();
+ assertThat(instance.getConsumerGroupCount()).isZero();
+ });
+ } finally {
+ releaseProvider.countDown();
+ }
+ }
+
@Test
void createInstanceShouldThrowWhenRequestIsNull() {
assertThatThrownBy(() -> instanceService.createInstance(null))
