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 2a6810b99 fix(instance): stop a cancelled count batch before the
second lookup (#7142)
2a6810b99 is described below
commit 2a6810b99c62f0863f34f1592dc29d68ca0ad3c1
Author: btlqql <[email protected]>
AuthorDate: Mon Oct 26 20:50:33 2026 +0800
fix(instance): stop a cancelled count batch before the second lookup (#7142)
* fix(instance): stop a cancelled count batch before the second lookup
(#7077)
InstanceService.loadCounts started countGroups unconditionally after
countTopics. A batch that had
already reached its deadline or been cancelled interrupts its worker, but a
provider that ignores
the interrupt can still return the first count afterwards, so the worker
went on to start the
second provider lookup for a response nobody was waiting for.
loadCounts now checks the worker interrupt flag between the two lookups and
throws
InterruptedException when it is still set. The exception takes the runner's
existing failure path
(InstanceResourceCountRunner.execute -> CompletedCount.failure), so there
is no executor policy
change: the caller keeps its existing unavailable response and a healthy
batch still performs both
lookups. Thread.interrupted() consumes the flag the way any
InterruptedException producer does, so
the pooled worker does not inherit a stale interrupt for its next count
task.
Because that exception surfaces as a failed outcome, fillCountsInParallel
now treats an
InterruptedException as a cancellation rather than a provider failure: the
instance still loses its
counts, but the event is logged at debug instead of counting as a failed
lookup. A cancelled batch
is routine, and counting it as a failure would have made the failure signal
meaningless.
The pre-existing late-result test signalled its "provider finished" latch
from countGroups and thus
required the unwanted lookup; it now signals that latch when the first
lookup returns.
* test(instance): keep a blank line before the create-instance cases
Mechanical follow-up to the cancelled-batch change: the new case had been
appended
without the separating blank line. The extra coverage for the interrupt
reclassification was dropped on review instead — the runner reports a
deadline-cancelled worker as TIMED_OUT/INTERRUPTED with a null failure, so
the
instanceof-InterruptedException branch is not reachable from that scenario
and a
log-asserting case could not be made mutation-sensitive.
---
.../rocketmq/studio/instance/InstanceService.java | 19 ++++++-
.../studio/instance/InstanceServiceTest.java | 63 ++++++++++++++++++++--
2 files changed, 77 insertions(+), 5 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 6fce8cae2..633766836 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
@@ -166,6 +166,13 @@ public class InstanceService {
}
clearCounts(instance);
+ // A cancelled batch is routine (the caller stopped waiting or the
deadline passed),
+ // so the interrupt the worker raises at the checkpoint must not
be counted or logged as
+ // a provider failure: the instance keeps the same unavailable
counts as a timeout.
+ if (outcome.failure() instanceof InterruptedException) {
+ log.debug("Resource count lookup cancelled for instance {}",
instance.getId());
+ continue;
+ }
switch (outcome.status()) {
case REJECTED -> rejected++;
case TIMED_OUT -> timedOut++;
@@ -198,10 +205,20 @@ public class InstanceService {
* unique name first, so a name that happens to equal another instance's
numeric id would
* otherwise shadow it and attribute the counts to the wrong instance.
*/
- private InstanceResourceCountRunner.ResourceCounts loadCounts(InstanceVO
instance) {
+ private InstanceResourceCountRunner.ResourceCounts loadCounts(InstanceVO
instance)
+ throws InterruptedException {
InstanceVendor vendor = instance.getVendor() == null ?
InstanceVendor.APACHE : instance.getVendor();
InstanceProvider provider = providerRegistry.forVendor(vendor);
int topicCount = provider.countTopics(instance.getName());
+ // The batch deadline or a caller cancellation interrupts this worker,
but a provider that
+ // ignores the interrupt can still return a count afterwards. Check
the flag between the two
+ // lookups so a batch nobody waits for does not start the second
remote call.
+ // Thread.interrupted() consumes the flag the way any
InterruptedException producer does, so
+ // the pooled worker does not hand a stale interrupt to its next count
task.
+ if (Thread.interrupted()) {
+ throw new InterruptedException(
+ "resource count batch cancelled after counting topics of "
+ instance.getName());
+ }
int consumerGroupCount = provider.countGroups(instance.getName());
return new InstanceResourceCountRunner.ResourceCounts(topicCount,
consumerGroupCount);
}
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 a5182abe2..b1b041cdf 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
@@ -440,11 +440,10 @@ class InstanceServiceTest {
if (interrupted) {
Thread.currentThread().interrupt();
}
- return 7;
- });
- when(instanceProvider.countGroups("slow")).thenAnswer(invocation -> {
+ // Signal the finished lookup here: this test is about the
returned response, and it must
+ // not require the cancelled batch to reach the consumer-group
lookup.
providerFinished.countDown();
- return 5;
+ return 7;
});
try {
@@ -470,6 +469,62 @@ class InstanceServiceTest {
}
}
+ @Test
+ void cancelledCountBatchShouldNotStartTheGroupLookupTest() throws
Exception {
+ InstanceVO slow = InstanceVO.builder().name("slow").build();
+ slow.setId(25L);
+ 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 groupCountStarted = new CountDownLatch(1);
+ when(instanceProvider.countTopics("slow")).thenAnswer(invocation -> {
+ topicCountStarted.countDown();
+ boolean interrupted = false;
+ while (true) {
+ try {
+ releaseProvider.await();
+ break;
+ } catch (InterruptedException exception) {
+ // A provider that outlives Future.cancel(true): it keeps
waiting and restores
+ // the interrupt instead of aborting the in-flight lookup.
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ return 7;
+ });
+ // Lenient because the fixed path never reaches this lookup; on the
buggy path it is invoked.
+
org.mockito.Mockito.lenient().when(instanceProvider.countGroups("slow")).thenAnswer(invocation
-> {
+ groupCountStarted.countDown();
+ return 5;
+ });
+
+ try {
+ List<InstanceVO> result = instanceService.listInstances(null,
null, true);
+
+ // The batch deadline passed while the first lookup was still
running, so the caller
+ // keeps its existing unavailable response.
+ assertThat(topicCountStarted.await(1, TimeUnit.SECONDS)).isTrue();
+ assertThat(result).singleElement().satisfies(instance -> {
+ assertThat(instance.isResourceCountsAvailable()).isFalse();
+ assertThat(instance.getConsumerGroupCount()).isZero();
+ });
+
+ releaseProvider.countDown();
+ // The worker finishes the first lookup with the interruption
still pending. Wait for that
+ // detached worker within a bound (the unwanted lookup must never
complete, so the wait
+ // cannot be latched on it) and then require that it never started.
+ groupCountStarted.await(2, TimeUnit.SECONDS);
+ verify(instanceProvider, never()).countGroups("slow");
+ } finally {
+ releaseProvider.countDown();
+ }
+ }
+
@Test
void createInstanceShouldThrowWhenRequestIsNull() {
assertThatThrownBy(() -> instanceService.createInstance(null))
