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 d33acc5b5 fix(group): prevent late enrichment from changing returned
groups (#5229)
d33acc5b5 is described below
commit d33acc5b57798fc04e40f701eac447509a2e637c
Author: coder999o <[email protected]>
AuthorDate: Fri Oct 9 11:30:55 2026 +0800
fix(group): prevent late enrichment from changing returned groups (#5229)
The enrichment workers mutated the very `ConsumerGroupVO` the request
thread was about to
serialize, and `future.cancel(true)` does not guarantee a worker has
stopped - `PageResult.of` is a
shallow copy, so a late broker answer could still change a group after the
response had been built.
Workers now write an immutable `GroupLiveSnapshot` into their own
`AtomicReference`, and the request
thread copies the values back once, in a `finally`, after the batch
deadline. Same shape as the
existing `InstanceResourceCountRunner` precedent.
---
.../provider/apache/RocketMQMetadataProvider.java | 81 ++++++++++-----
.../apache/RocketMQMetadataProviderTest.java | 114 +++++++++++++++++++++
2 files changed, 170 insertions(+), 25 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
index a516abc77..50bce4e40 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
@@ -77,6 +77,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
/**
* Real MetadataProvider implementation.
@@ -290,6 +291,14 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
private static final int ONLINE_ENRICHMENT_THREADS = 8;
private static final long ONLINE_ENRICHMENT_TIMEOUT_SECONDS = 3;
+ private record GroupLiveSnapshot(List<ConsumerInstanceVO> instances, int
onlineInstances,
+ boolean consumeStatsAvailable, long totalLag, int delaySeconds,
+ boolean consumptionTimestampAvailable) {
+ private static GroupLiveSnapshot empty() {
+ return new GroupLiveSnapshot(List.of(), 0, false, 0L, 0, false);
+ }
+ }
+
private final ExecutorService onlineEnrichmentExecutor =
Executors.newFixedThreadPool(
ONLINE_ENRICHMENT_THREADS, runnable -> {
Thread thread = new Thread(runnable,
"group-online-enrichment");
@@ -326,52 +335,71 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
return;
}
List<Future<?>> futures = new ArrayList<>(groups.size());
+ List<AtomicReference<GroupLiveSnapshot>> snapshots = new
ArrayList<>(groups.size());
for (ConsumerGroupVO vo : groups) {
- futures.add(onlineEnrichmentExecutor.submit(() ->
enrichGroupLiveStats(instanceId, vo)));
+ String groupName = vo.getName();
+ AtomicReference<GroupLiveSnapshot> snapshot = new
AtomicReference<>(GroupLiveSnapshot.empty());
+ snapshots.add(snapshot);
+ futures.add(onlineEnrichmentExecutor.submit(() ->
enrichGroupLiveStats(instanceId, groupName, snapshot)));
}
// One absolute deadline for the whole batch (the
InstanceResourceCountRunner
// precedent): a hanging broker must not multiply the wait by the
number of
// groups, so each future only gets the time the batch has left.
long startedNanos = System.nanoTime();
long timeoutNanos =
TimeUnit.SECONDS.toNanos(ONLINE_ENRICHMENT_TIMEOUT_SECONDS);
- for (Future<?> future : futures) {
- long remainingNanos = timeoutNanos - (System.nanoTime() -
startedNanos);
- try {
- if (remainingNanos <= 0) {
- throw new TimeoutException("enrichment batch deadline
reached");
+ try {
+ for (Future<?> future : futures) {
+ long remainingNanos = timeoutNanos - (System.nanoTime() -
startedNanos);
+ try {
+ if (remainingNanos <= 0) {
+ throw new TimeoutException("enrichment batch deadline
reached");
+ }
+ future.get(remainingNanos, TimeUnit.NANOSECONDS);
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ } catch (ExecutionException e) {
+ // leave the stats at the last completed phase
}
- future.get(remainingNanos, TimeUnit.NANOSECONDS);
- } catch (TimeoutException e) {
- future.cancel(true);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- return;
- } catch (ExecutionException e) {
- // leave the stats at zero
+ }
+ } finally {
+ // A canceled broker call may continue, but it can only update its
private snapshot.
+ for (int index = 0; index < groups.size(); index++) {
+ GroupLiveSnapshot liveStats = snapshots.get(index).get();
+ ConsumerGroupVO vo = groups.get(index);
+ vo.setInstances(liveStats.instances());
+ vo.setOnlineInstances(liveStats.onlineInstances());
+ vo.setConsumeStatsAvailable(liveStats.consumeStatsAvailable());
+ vo.setTotalLag(liveStats.totalLag());
+ vo.setDelaySeconds(liveStats.delaySeconds());
+
vo.setConsumptionTimestampAvailable(liveStats.consumptionTimestampAvailable());
}
}
}
- private void enrichGroupLiveStats(String instanceId, ConsumerGroupVO vo) {
+ private void enrichGroupLiveStats(String instanceId, String groupName,
+ AtomicReference<GroupLiveSnapshot> snapshot) {
// The detail modal reuses the listed group as-is, so the online
instance list has to be
// filled here too; both fields come from the same connection set to
stay consistent.
ProxyConsumerResolver.ConsumerConnectionResolution connection =
- resolveConsumerConnection(instanceId, vo.getName());
+ resolveConsumerConnection(instanceId, groupName);
List<ConsumerInstanceVO> instances =
ConsumerConnections.toInstances(connection.connection());
- vo.setInstances(instances);
- vo.setOnlineInstances(connection.available() ? instances.size() : -1);
+ int onlineInstances = connection.available() ? instances.size() : -1;
+ snapshot.set(new GroupLiveSnapshot(instances, onlineInstances, false,
0L, 0, false));
try {
ConsumeStats stats;
if (StringUtils.hasText(instanceId)) {
stats = runtimeAdminClientResolver.execute(instanceId,
- admin -> admin.examineConsumeStats(vo.getName()));
+ admin -> admin.examineConsumeStats(groupName));
} else {
- stats = adminExecute(admin ->
admin.examineConsumeStats(vo.getName()));
+ stats = adminExecute(admin ->
admin.examineConsumeStats(groupName));
}
if (stats == null) {
return;
}
- vo.setConsumeStatsAvailable(true);
+ snapshot.set(new GroupLiveSnapshot(instances, onlineInstances,
true, 0L, 0, false));
if (stats.getOffsetTable() == null ||
stats.getOffsetTable().isEmpty()) {
return;
}
@@ -392,12 +420,15 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
newestConsumedTimestamp = lastTimestamp;
}
}
- vo.setTotalLag(lagUnknown ? ConsumerLagResolver.UNKNOWN :
totalLag);
+ int delaySeconds = 0;
+ boolean timestampAvailable = false;
if (newestConsumedTimestamp > 0) {
- long delaySeconds = (System.currentTimeMillis() -
newestConsumedTimestamp) / 1000;
- vo.setDelaySeconds((int) Math.max(delaySeconds, 0));
- vo.setConsumptionTimestampAvailable(true);
+ long elapsedSeconds = (System.currentTimeMillis() -
newestConsumedTimestamp) / 1000;
+ delaySeconds = (int) Math.max(elapsedSeconds, 0);
+ timestampAvailable = true;
}
+ snapshot.set(new GroupLiveSnapshot(instances, onlineInstances,
true,
+ lagUnknown ? ConsumerLagResolver.UNKNOWN : totalLag,
delaySeconds, timestampAvailable));
} catch (Exception e) {
// No consume stats (e.g. POP-only group without an offset table):
keep zeros.
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
index 308d55127..5bfbeabea 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
@@ -62,6 +62,9 @@ import java.util.HashSet;
import java.util.List;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -771,6 +774,117 @@ class RocketMQMetadataProviderTest {
assertThat(elapsedMillis).isLessThan(6_000L);
}
+ @Test
+ void timedOutGroupEnrichmentMustNotMutateReturnedGroupTest() throws
Exception {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("cg-late");
+ entity.setInstanceId("instance-a");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+
+ CountDownLatch connectionStarted = new CountDownLatch(1);
+ CountDownLatch releaseConnection = new CountDownLatch(1);
+ CountDownLatch statsRead = new CountDownLatch(1);
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ org.apache.rocketmq.remoting.protocol.body.ConsumerConnection
connection =
+ new
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
+ org.apache.rocketmq.remoting.protocol.body.Connection client =
+ new org.apache.rocketmq.remoting.protocol.body.Connection();
+ client.setClientId("late-client");
+ connection.setConnectionSet(new HashSet<>(List.of(client)));
+
when(admin.examineConsumerConnectionInfo("cg-late")).thenAnswer(invocation -> {
+ connectionStarted.countDown();
+ while (releaseConnection.getCount() > 0) {
+ try {
+ releaseConnection.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException ignored) {
+ // Simulate a broker call that ignores cancellation and
returns later.
+ }
+ }
+ return connection;
+ });
+ when(admin.examineConsumeStats("cg-late")).thenAnswer(invocation -> {
+ statsRead.countDown();
+ return null;
+ });
+ when(runtimeAdminClientResolver.execute(eq("instance-a"), any()))
+ .thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(admin));
+ RocketMQMetadataProvider provider = newLiveProvider(admin);
+
+ try {
+ ConsumerGroupVO returned =
provider.listConsumerGroups("instance-a", null, null).get(0);
+ assertThat(connectionStarted.getCount()).isZero();
+ assertThat(returned.getOnlineInstances()).isZero();
+
+ releaseConnection.countDown();
+ assertThat(statsRead.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(returned.getOnlineInstances()).isZero();
+ assertThat(returned.getInstances()).isEmpty();
+ } finally {
+ releaseConnection.countDown();
+ provider.shutdownOnlineEnrichmentExecutor();
+ }
+ }
+
+ @Test
+ void timedOutConsumeStatsKeepsCompletedConnectionInfoTest() throws
Exception {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("cg-partial");
+ entity.setInstanceId("instance-a");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+
+ CountDownLatch statsStarted = new CountDownLatch(1);
+ CountDownLatch releaseStats = new CountDownLatch(1);
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ org.apache.rocketmq.remoting.protocol.body.ConsumerConnection
connection =
+ new
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
+ org.apache.rocketmq.remoting.protocol.body.Connection client =
+ new org.apache.rocketmq.remoting.protocol.body.Connection();
+ client.setClientId("online-client");
+ connection.setConnectionSet(new HashSet<>(List.of(client)));
+
when(admin.examineConsumerConnectionInfo("cg-partial")).thenReturn(connection);
+ ConsumeStats stats = new ConsumeStats();
+ OffsetWrapper wrapper = new OffsetWrapper();
+ wrapper.setBrokerOffset(100L);
+ wrapper.setConsumerOffset(60L);
+ stats.getOffsetTable().put(new MessageQueue("orders", "broker-a", 0),
wrapper);
+ when(admin.examineConsumeStats("cg-partial")).thenAnswer(invocation ->
{
+ statsStarted.countDown();
+ while (releaseStats.getCount() > 0) {
+ try {
+ releaseStats.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException ignored) {
+ // Simulate a broker call that does not stop immediately
on cancellation.
+ }
+ }
+ return stats;
+ });
+ when(runtimeAdminClientResolver.execute(eq("instance-a"), any()))
+ .thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(admin));
+ RocketMQMetadataProvider provider = newLiveProvider(admin);
+ ExecutorService executor = (ExecutorService)
org.springframework.test.util.ReflectionTestUtils.getField(
+ provider, "onlineEnrichmentExecutor");
+
+ try {
+ ConsumerGroupVO returned =
provider.listConsumerGroups("instance-a", null, null).get(0);
+ assertThat(statsStarted.getCount()).isZero();
+ assertThat(returned.getOnlineInstances()).isEqualTo(1);
+ assertThat(returned.getInstances())
+
.extracting(org.apache.rocketmq.studio.instance.group.ConsumerInstanceVO::getClientId)
+ .containsExactly("online-client");
+ assertThat(returned.isConsumeStatsAvailable()).isFalse();
+ releaseStats.countDown();
+ provider.shutdownOnlineEnrichmentExecutor();
+ assertThat(executor.awaitTermination(5,
TimeUnit.SECONDS)).isTrue();
+ assertThat(returned.isConsumeStatsAvailable()).isFalse();
+ assertThat(returned.getTotalLag()).isZero();
+ } finally {
+ releaseStats.countDown();
+ provider.shutdownOnlineEnrichmentExecutor();
+ }
+ }
+
@Test
void
listConsumerGroupsShouldMarkOnlineInstancesUnknownWhenConnectionLookupFailsTest()
throws Exception {
RmqGroup entity = new RmqGroup();
