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 aae11ff0d fix(lite): surface session metric read failures (#4536)
aae11ff0d is described below
commit aae11ff0da98b1f51eb23e4180740bb90cebddda
Author: zmuxuny <[email protected]>
AuthorDate: Mon Sep 21 21:01:36 2026 +0800
fix(lite): surface session metric read failures (#4536)
`RocketMQLiteTopicProvider.getSession` folded broker metric failures into
zeros. The backlog came from `groupLag`, which swallowed a failed
`getLiteGroupInfo` into `0`, and `consumedMessages` logged each per-LiteTopic
offset read failure at debug and skipped the topic. A session whose master was
unreachable therefore rendered as a healthy group with no pending messages, on
the one screen where the operator explicitly asked for that group's numbers.
The session path now reads its backlog through a new `sessionGroupLag` that
throws `BusinessException(502)` when the call fails or returns no body, and a
failed offset read throws instead of being skipped; `InterruptedException`
restores the interrupt flag before the rethrow. The best-effort `groupLag` is
left in place for the list and summary paths, where one unreadable group must
not take the whole page down.
Fixes #4535
---
.../provider/apache/RocketMQLiteTopicProvider.java | 29 ++++++++++-
.../apache/RocketMQLiteTopicProviderTest.java | 58 ++++++++++++++++++++++
2 files changed, 85 insertions(+), 2 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
index f86f69a80..f7b160744 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
@@ -275,7 +275,7 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
.toList();
session.setLiteTopics(new LinkedHashSet<>(liteTopics));
- long pending = groupLag(admin, located.master, group);
+ long pending = sessionGroupLag(admin, located.master, group);
long consumed = consumedMessages(admin, located.master, group,
liteTopics);
session.setPendingMessages(pending);
session.setConsumedMessages(consumed);
@@ -317,7 +317,10 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
consumed += wrapper.getConsumerOffset();
}
} catch (Exception failure) {
- log.debug("Failed to read lite offset for {}|{}: {}", group,
liteTopic, failure.getMessage());
+ restoreInterrupt(failure);
+ throw new BusinessException(502,
+ "Failed to read LiteTopic consumed offset for " +
group + "|" + liteTopic
+ + ": " + failure.getMessage());
}
}
return consumed;
@@ -521,6 +524,28 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
return null;
}
+ private long sessionGroupLag(MQAdminExt admin, String brokerAddr, String
group) {
+ try {
+ GetLiteGroupInfoResponseBody body =
admin.getLiteGroupInfo(brokerAddr, group, null, 1);
+ if (body == null) {
+ throw new BusinessException(502, "Broker returned no LiteTopic
backlog for group " + group);
+ }
+ return Math.max(body.getTotalLagCount(), 0);
+ } catch (BusinessException failure) {
+ throw failure;
+ } catch (Exception failure) {
+ restoreInterrupt(failure);
+ throw new BusinessException(502,
+ "Failed to read LiteTopic backlog for group " + group + ":
" + failure.getMessage());
+ }
+ }
+
+ private static void restoreInterrupt(Exception failure) {
+ if (failure instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
private long groupLag(MQAdminExt admin, String brokerAddr, String group) {
try {
GetLiteGroupInfoResponseBody body =
admin.getLiteGroupInfo(brokerAddr, group, null, 1);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
index 58f234e3b..0fbf4c004 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
@@ -184,6 +184,64 @@ class RocketMQLiteTopicProviderTest {
assertThat(session.getTotalMessages()).isEqualTo(15L);
}
+ @Test
+ void getSessionShouldSurfaceBacklogReadFailureTest() throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.getLiteClientInfo(BROKER_A, PARENT, GROUP, "c1"))
+ .thenReturn(clientInfo(1, System.currentTimeMillis(),
LiteUtil.toLmqName(PARENT, "bob")));
+ when(admin.getLiteGroupInfo(BROKER_A, GROUP, null, 1))
+ .thenThrow(new IllegalStateException("backlog unavailable"));
+
+ assertThatThrownBy(() -> provider.getSession(
+ RocketMQLiteTopicProvider.encodeSessionId(PARENT, GROUP,
"c1")))
+ .isInstanceOfSatisfying(BusinessException.class,
+ ex -> assertThat(ex.getCode()).isEqualTo(502));
+ }
+
+ @Test
+ void getSessionShouldSurfaceConsumedOffsetReadFailureTest() throws
Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.getLiteClientInfo(BROKER_A, PARENT, GROUP, "c1"))
+ .thenReturn(clientInfo(1, System.currentTimeMillis(),
LiteUtil.toLmqName(PARENT, "bob")));
+ when(admin.getLiteGroupInfo(BROKER_A, GROUP, null,
1)).thenReturn(lag(5));
+ when(admin.getLiteGroupInfo(BROKER_A, GROUP, "bob", 1))
+ .thenThrow(new IllegalStateException("offset unavailable"));
+
+ assertThatThrownBy(() -> provider.getSession(
+ RocketMQLiteTopicProvider.encodeSessionId(PARENT, GROUP,
"c1")))
+ .isInstanceOfSatisfying(BusinessException.class,
+ ex -> assertThat(ex.getCode()).isEqualTo(502));
+ }
+
+ @Test
+ void getSessionShouldRejectMissingBacklogBodyTest() throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.getLiteClientInfo(BROKER_A, PARENT, GROUP, "c1"))
+ .thenReturn(clientInfo(0, System.currentTimeMillis()));
+
+ assertThatThrownBy(() -> provider.getSession(
+ RocketMQLiteTopicProvider.encodeSessionId(PARENT, GROUP,
"c1")))
+ .isInstanceOfSatisfying(BusinessException.class,
+ ex -> assertThat(ex.getCode()).isEqualTo(502));
+ }
+
+ @Test
+ void getSessionShouldKeepZeroConsumedWhenNoOffsetIsCommittedTest() throws
Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+ when(admin.getLiteClientInfo(BROKER_A, PARENT, GROUP, "c1"))
+ .thenReturn(clientInfo(1, System.currentTimeMillis(),
LiteUtil.toLmqName(PARENT, "bob")));
+ when(admin.getLiteGroupInfo(BROKER_A, GROUP, null,
1)).thenReturn(lag(0));
+ when(admin.getLiteGroupInfo(BROKER_A, GROUP, "bob", 1))
+ .thenReturn(new GetLiteGroupInfoResponseBody());
+
+ LiteTopicSession session = provider.getSession(
+ RocketMQLiteTopicProvider.encodeSessionId(PARENT, GROUP,
"c1"));
+
+ assertThat(session.getPendingMessages()).isZero();
+ assertThat(session.getConsumedMessages()).isZero();
+ assertThat(session.getTotalMessages()).isZero();
+ }
+
@Test
void getSessionFailsWhenNoBrokerReportsTheClient() throws Exception {
when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
