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 2df0a70e1 fix(studio): stop presenting unmeasured values as measured
(#5918)
2df0a70e1 is described below
commit 2df0a70e17335ce597ac213054e46e6d1ad7441f
Author: 烤化の初雪 <[email protected]>
AuthorDate: Tue Oct 27 21:38:48 2026 +0800
fix(studio): stop presenting unmeasured values as measured (#5918)
* fix(cluster): stop rendering a failed broker runtime-stats read as 0% and
0 TPS (#5918)
The cluster page sorted and rendered disk usage and TPS from whatever the
runtime
stats read returned, so a failed read showed as 0% and 0 TPS - a broker
that could
not be queried looked like an empty, healthy one. Track statsUnavailable and
render "unavailable" with direction-aware sorting, the way
BrokerCluster.tsx on
the studio page already did. The three bare sorters are the ones the page
had.
* fix(consumer): stop presenting an unmeasurable consume delay as 0 seconds
(#5920)
delaySeconds had no availability signal, so a group whose delay could not be
measured rendered as "0 seconds" in the list, the detail drawer, the CSV
export
and the diagnostics summary. Propagate the flag the backend already sends
(ConsumerGroupVO) through metadata.ts, consumerService.ts and
consumerGroupDiagnostics.ts, render "unavailable" at all five consumption
points,
and sort unmeasured rows to the bottom in both directions. While landing it
we
added the ascending-sort assertion the review asked for - without it the
MAX_SAFE_INTEGER branch of the local delaySortValue had no coverage - and
fixed a
line-wrap in the PR's own consumerService.test.ts that failed prettier.
* fix(consumer): keep unmeasured rows out of the top of a descending sort
(#5902)
The total-lag sorter treated an unmeasured value as the largest possible
number,
so a descending sort put the rows we know least about at the top. Add a
sortOrder
so unmeasured rows sink in both directions, following compareResourceCounts
on the
instance page and unmeasuredSortKey on the LiteTopic page. Scope note: this
covers
the total-lag column only. The PR as submitted also touched
consumerConnections.ts,
but trunk has since replaced onlineInstancesSortValue with
onlineInstancesCount
(#6916, where an unavailable count renders as 0 by design), and the
author's own
merge of trunk had already dropped that part - so it is not included here.
* fix(acl): apply the rule version filter the console sends (#5900)
The console has always sent aclVersion (all / 1.0 / 2.0) and the repository
predicate already existed, but the controller never declared the parameter
and the
service hard-coded null, so the version dropdown was a dead control. Thread
it
through the controller and the service and add the version comparison to the
Tencent in-memory filter chain. The 7-argument listRules overload stays:
AclListToolHandler still calls it.
* fix(alert): stop the notification template from rendering values in
scientific notation (#5889)
Thresholds and values went through String.valueOf, so a large threshold
reached
on-call as "1.0E+7". Route both render points through formatNumber.
* fix(auth): stamp the user and session audit columns in UTC (#5886)
The user and session tables left gmt_create / gmt_modified to database
defaults,
so the recorded instant depended on the server's timezone while the rest of
the
auth path is UTC - and #7311 had started surfacing gmt_create to the UI
through
StudioUserSessionDetailVO and StudioUserVO. Write both columns explicitly
from
AuthService.now(). Landing this needed the write points added since the PR
was
opened (#7311 and #7312) to be covered too, so all 13 write points now stamp
explicitly rather than half the table being UTC and half the database
default;
the conflicts in AuthService were resolved as a union, keeping #7312's
passwordMustChange and #7230's monotonic last_seen_at guard.
Batch integration of six related fixes by the same contributor; #5886,
#5889,
#5900, #5902 and #5920 are closed as merged into this one. Verified on the
combined branch: 4157 backend tests, 2016 frontend tests across 166 files,
the
production build and type check, license gate 8/8, and eslint with no new
warnings against the trunk baseline.
---
docs/api-spec.md | 3 +-
.../apache/rocketmq/studio/auth/AuthService.java | 48 ++++++--
.../studio/instance/acl/AclController.java | 3 +-
.../rocketmq/studio/instance/acl/AclService.java | 18 ++-
.../ops/alert/AlertNotificationTemplate.java | 14 ++-
.../studio/auth/AuthServiceDatabaseTest.java | 137 ++++++++++++++++++++-
.../studio/instance/acl/AclControllerTest.java | 23 +++-
.../studio/instance/acl/AclServiceTest.java | 32 +++++
.../ops/alert/AlertNotificationTemplateTest.java | 27 ++++
web/src/api/metadata.ts | 5 +
.../pages/cluster/__tests__/ClusterPage.test.tsx | 50 ++++++++
web/src/pages/cluster/index.tsx | 60 +++++++--
.../pages/instance/__tests__/ConsumerPage.test.tsx | 73 +++++++++++
web/src/pages/instance/consumer.tsx | 37 +++++-
web/src/services/consumerService.test.ts | 31 +++++
web/src/services/consumerService.ts | 8 +-
web/src/utils/consumerGroupDiagnostics.test.ts | 30 +++++
web/src/utils/consumerGroupDiagnostics.ts | 45 ++++---
web/src/utils/consumerLag.test.ts | 8 ++
web/src/utils/consumerLag.ts | 17 ++-
20 files changed, 612 insertions(+), 57 deletions(-)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index 32c161128..8f124bf49 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -1833,7 +1833,7 @@ POST /api/groups/reset-offset/preview
### 7.1 获取 ACL 规则列表
```
-GET
/api/acl/rules?principal={principal}&resource={resource}&scope={scope}&decision={decision}&instanceId={instanceId}&page={page}&pageSize={pageSize}
+GET
/api/acl/rules?principal={principal}&resource={resource}&scope={scope}&decision={decision}&aclVersion={aclVersion}&instanceId={instanceId}&page={page}&pageSize={pageSize}
```
**Query Parameters:**
@@ -1844,6 +1844,7 @@ GET
/api/acl/rules?principal={principal}&resource={resource}&scope={scope}&decis
| `resource` | `string` | 否 | 按资源过滤 |
| `scope` | `string` | 否 | 按作用域过滤 |
| `decision` | `string` | 否 | 按决策过滤 |
+| `aclVersion` | `string` | 否 | 按 ACL 版本过滤（`1.0` / `2.0`），省略时返回全部版本 |
| `instanceId` | `string` | 否 | 按所属实例过滤 |
| `page` | `number` | 否 | 页码，默认 `1` |
| `pageSize` | `number` | 否 | 每页条数，默认 `20` |
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
index 0d6c349bb..2a03ae3d0 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
@@ -190,10 +190,12 @@ public class AuthService {
public void logout(String authorization) {
tokenFromAuthorization(authorization).ifPresent(token -> {
if (databaseBacked()) {
+ LocalDateTime current = now();
sessionMapper.update(null, new
UpdateWrapper<RmqStudioSession>()
.eq("token_hash", tokenHash(token))
.isNull("revoked_at")
- .set("revoked_at", now()));
+ .set("revoked_at", current)
+ .set("gmt_modified", current));
} else {
activeTokens.remove(token);
}
@@ -350,6 +352,7 @@ public class AuthService {
wrapper.ne("user_id", Long.parseLong(currentUserId));
}
wrapper.set("revoked_at", current);
+ wrapper.set("gmt_modified", current);
return sessionMapper.update(null, wrapper);
}
@@ -371,7 +374,8 @@ public class AuthService {
sessionMapper.update(null, new UpdateWrapper<RmqStudioSession>()
.eq("id", sessionId)
.isNull("revoked_at")
- .set("revoked_at", current));
+ .set("revoked_at", current)
+ .set("gmt_modified", current));
}
public RmqStudioUser createUser(String username, String password, boolean
admin) {
@@ -386,10 +390,16 @@ public class AuthService {
user.setPasswordHash(passwordHasher.hash(password));
user.setAdmin(admin);
user.setEnabled(true);
- user.setPasswordChangedAt(now());
+ LocalDateTime current = now();
+ user.setPasswordChangedAt(current);
// The initial password was chosen by the creator, not the owner: the
owner rotates it
// at first login.
user.setPasswordMustChange(true);
+ // An omitted gmt_create/gmt_modified falls back to the database
session's
+ // CURRENT_TIMESTAMP, which is not UTC on a server running in another
zone, while the
+ // API serializes every zone-less timestamp as UTC.
+ user.setGmtCreate(current);
+ user.setGmtModified(current);
try {
userMapper.insert(user);
} catch (DuplicateKeyException exception) {
@@ -418,7 +428,11 @@ public class AuthService {
throw new BusinessException(409, "The last enabled
administrator cannot be disabled");
}
}
- userMapper.updateById(userWithEnabled(user, enabled));
+ RmqStudioUser enableUpdate = userWithEnabled(user, enabled);
+ // updateById only writes non-null fields, so without this the
column's ON UPDATE
+ // CURRENT_TIMESTAMP - the database session's zone, not UTC - would be
the audit instant.
+ enableUpdate.setGmtModified(now());
+ userMapper.updateById(enableUpdate);
if (!enabled) {
revokeUserSessions(user.getId());
}
@@ -454,6 +468,10 @@ public class AuthService {
RmqStudioUser update = new RmqStudioUser();
update.setId(user.getId());
update.setAdmin(admin);
+ // Same reason as in setUserEnabled: updateById writes only non-null
fields, so the audit
+ // instant has to be stamped here or the column's ON UPDATE
CURRENT_TIMESTAMP - the
+ // database session's zone, not UTC - would be it.
+ update.setGmtModified(now());
userMapper.updateById(update);
revokeUserSessions(user.getId());
user.setAdmin(admin);
@@ -516,7 +534,8 @@ public class AuthService {
.set("password_changed_at", now())
// A password chosen by its owner needs no rotation; one
handed out by an
// administrator does, until the owner replaces it.
- .set("password_must_change", !requireCurrentPassword));
+ .set("password_must_change", !requireCurrentPassword)
+ .set("gmt_modified", now()));
revokeUserSessions(user.getId());
}
@@ -535,7 +554,8 @@ public class AuthService {
.isNotNull("last_seen_at")
.ge("expires_at", current)
.lt("last_seen_at",
current.minusMinutes(idleTimeoutMinutes))
- .set("revoked_at", current));
+ .set("revoked_at", current)
+ .set("gmt_modified", current));
}
sessionMapper.delete(new
QueryWrapper<RmqStudioSession>().lt("expires_at", current));
} else {
@@ -576,6 +596,8 @@ public class AuthService {
session.setExpiresAt(current.plusSeconds(tokenTtlSeconds));
session.setClientIp(truncate(clientIp, MAX_CLIENT_IP_LENGTH));
session.setUserAgent(truncate(userAgent, MAX_USER_AGENT_LENGTH));
+ session.setGmtCreate(current);
+ session.setGmtModified(current);
sessionMapper.insert(session);
return loginResponse(userInfo(user), token, tokenTtlSeconds);
}
@@ -616,7 +638,8 @@ public class AuthService {
&&
!session.getLastSeenAt().plusMinutes(idleTimeoutMinutes).isAfter(current)) {
sessionMapper.update(null, new UpdateWrapper<RmqStudioSession>()
.eq("id", session.getId())
- .set("revoked_at", current));
+ .set("revoked_at", current)
+ .set("gmt_modified", current));
return Optional.empty();
}
if (session.getLastSeenAt() == null
@@ -625,7 +648,8 @@ public class AuthService {
.eq("id", session.getId())
// A delayed request must not overwrite activity a newer
request already recorded.
.and(activity ->
activity.isNull("last_seen_at").or().lt("last_seen_at", current))
- .set("last_seen_at", current));
+ .set("last_seen_at", current)
+ .set("gmt_modified", current));
}
return Optional.of(userInfo(user));
}
@@ -668,7 +692,10 @@ public class AuthService {
user.setPasswordHash(passwordHasher.hash(configuredUser.getPassword()));
user.setAdmin(configuredUser.isAdmin());
user.setEnabled(true);
- user.setPasswordChangedAt(now());
+ LocalDateTime current = now();
+ user.setPasswordChangedAt(current);
+ user.setGmtCreate(current);
+ user.setGmtModified(current);
try {
userMapper.insert(user);
} catch (DuplicateKeyException exception) {
@@ -704,7 +731,8 @@ public class AuthService {
.eq("user_id", userId)
.isNull("revoked_at")
.gt("expires_at", current)
- .set("revoked_at", current));
+ .set("revoked_at", current)
+ .set("gmt_modified", current));
}
private Set<Long> normalizeUserIds(Collection<Long> userIds) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclController.java
index 693927999..9451caaa2 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclController.java
@@ -61,10 +61,11 @@ public class AclController {
@RequestParam(required = false) String resource,
@RequestParam(required = false) String scope,
@RequestParam(required = false) String decision,
+ @RequestParam(required = false) String aclVersion,
@RequestParam(required = false) String instanceId,
@RequestParam(defaultValue = "1") Integer page,
@RequestParam(defaultValue = "20") Integer pageSize) {
- return Result.ok(aclService.listRules(principal, resource, scope,
decision,
+ return Result.ok(aclService.listRules(principal, resource, scope,
decision, aclVersion,
instanceId, page, pageSize));
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
index 666350a90..1f3636532 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
@@ -101,6 +101,16 @@ public class AclService {
public PageResult<AclRuleVO> listRules(String principal, String resource,
String scope, String decision,
String instanceId, Integer page, Integer pageSize) {
+ return listRules(principal, resource, scope, decision, null,
instanceId, page, pageSize);
+ }
+
+ /**
+ * @param aclVersion restricts the page to one ACL version ({@code
1.0}/{@code 2.0}); the console
+ * has offered this filter all along, and the repository
predicate existed, but
+ * nothing carried the value from the request to it.
+ */
+ public PageResult<AclRuleVO> listRules(String principal, String resource,
String scope, String decision,
+ String aclVersion, String instanceId, Integer page, Integer
pageSize) {
int normalizedPage = requireValidPage(page);
int normalizedPageSize = requireValidPageSize(pageSize);
requireAcl2Supported(instanceId);
@@ -109,12 +119,14 @@ public class AclService {
.filter(rule -> containsIgnoreCase(rule.getResource(),
resource))
.filter(rule -> equalsIgnoreCase(rule.getScope(), scope))
.filter(rule -> equalsIgnoreCase(rule.getDecision(),
decision))
+ .filter(rule -> equalsIgnoreCase(rule.getAclVersion(),
aclVersion))
.toList();
return toUtcWire(paginateRules(filtered, normalizedPage,
normalizedPageSize));
}
- log.info("Listing ACL rules for principal={}, resource={}, scope={},
decision={}, page={}, pageSize={}",
- principal, resource, scope, decision, normalizedPage,
normalizedPageSize);
- return toUtcWire(aclRepository.findRulePage(principal, resource,
scope, decision, null,
+ log.info("Listing ACL rules for principal={}, resource={}, scope={},
decision={}, aclVersion={},"
+ + " page={}, pageSize={}",
+ principal, resource, scope, decision, aclVersion,
normalizedPage, normalizedPageSize);
+ return toUtcWire(aclRepository.findRulePage(principal, resource,
scope, decision, aclVersion,
normalizedPage, normalizedPageSize));
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationTemplate.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationTemplate.java
index 1c75373cb..6017ee3e9 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationTemplate.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationTemplate.java
@@ -51,7 +51,7 @@ final class AlertNotificationTemplate {
values.put("metric", text(rule == null ? null : rule.getMetric()));
values.put("instanceId", text(alert.getInstanceId()));
values.put("value", formattedValue(alert, rule));
- values.put("threshold", rule == null ? "" :
String.valueOf(rule.getThreshold()));
+ values.put("threshold", rule == null ? "" :
formatNumber(rule.getThreshold()));
values.put("thresholdUnit", text(rule == null ? null :
rule.getThresholdUnit()));
values.put("level", text(alert.getLevel()));
values.put("time", alert.getTime() == null ? "" :
alert.getTime().toString());
@@ -73,7 +73,17 @@ final class AlertNotificationTemplate {
?
BigDecimal.valueOf(currentValue).movePointRight(2).stripTrailingZeros().toPlainString()
: String.valueOf(currentValue);
}
- return String.valueOf(currentValue);
+ return formatNumber(currentValue);
+ }
+
+ /**
+ * Renders a measurement with {@code Double.toString} instead of the plain
decimal the rule
+ * dialog previews. {@code Double.toString} switches to computerized
scientific notation at 1e7,
+ * so a consumer-lag alert of 12,500,000 messages reached the operator as
"1.25E7" while the
+ * preview of the same template showed "12500000". Only a non-finite value
has no decimal form.
+ */
+ private static String formatNumber(double value) {
+ return Double.isFinite(value) ?
BigDecimal.valueOf(value).toPlainString() : String.valueOf(value);
}
private static String formatLabels(Map<String, String> labels) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
index 1cb77c6d1..3c8e1d7d5 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
@@ -56,6 +56,8 @@ import static org.mockito.Mockito.when;
class AuthServiceDatabaseTest {
+ private static final LocalDateTime UTC_CLOCK_INSTANT =
LocalDateTime.of(2026, 8, 13, 0, 0);
+
private AuthService authService;
private SettingsRepository settingsRepository;
private RmqStudioUserMapper userMapper;
@@ -293,6 +295,54 @@ class AuthServiceDatabaseTest {
assertThat(captor.getValue().getUserAgent()).isNull();
}
+ @Test
+ void createdUserCarriesTheUtcAuditInstantsTheApiSerializesTest() {
+ when(userMapper.selectOne(any(Wrapper.class))).thenReturn(null);
+
+ authService.createUser("operator", "password-1", false);
+
+ org.mockito.ArgumentCaptor<RmqStudioUser> captor =
+ org.mockito.ArgumentCaptor.forClass(RmqStudioUser.class);
+ verify(userMapper).insert(captor.capture());
+ // rmq_studio_user.gmt_create/gmt_modified otherwise fall back to the
database session's
+ // CURRENT_TIMESTAMP, while the API serializes every zone-less
timestamp as UTC.
+
assertThat(captor.getValue().getGmtCreate()).isEqualTo(UTC_CLOCK_INSTANT);
+
assertThat(captor.getValue().getGmtModified()).isEqualTo(UTC_CLOCK_INSTANT);
+ }
+
+ @Test
+ void createdSessionCarriesTheUtcAuditInstantsTheApiSerializesTest() {
+ RmqStudioUser user = user(1L, "operator", true, true, "password-1");
+ when(userMapper.selectCount(isNull())).thenReturn(1L);
+ when(userMapper.selectOne(any(Wrapper.class))).thenReturn(user);
+ LoginDTO request = new LoginDTO();
+ request.setUsername("operator");
+ request.setPassword("password-1");
+
+ authService.login(request);
+
+ org.mockito.ArgumentCaptor<RmqStudioSession> captor =
+ org.mockito.ArgumentCaptor.forClass(RmqStudioSession.class);
+ verify(sessionMapper).insert(captor.capture());
+
assertThat(captor.getValue().getGmtCreate()).isEqualTo(UTC_CLOCK_INSTANT);
+
assertThat(captor.getValue().getGmtModified()).isEqualTo(UTC_CLOCK_INSTANT);
+ }
+
+ @Test
+ void passwordChangeStampsTheUtcModifiedInstantTest() {
+ RmqStudioUser user = user(1L, "operator", false, true, "password-1");
+ when(userMapper.selectById(1L)).thenReturn(user);
+
+ authService.changePassword(1L, "password-1", "password-2", true);
+
+ org.mockito.ArgumentCaptor<UpdateWrapper<RmqStudioUser>> updateCaptor =
+ org.mockito.ArgumentCaptor.forClass(UpdateWrapper.class);
+ verify(userMapper).update(isNull(), updateCaptor.capture());
+
assertThat(updateCaptor.getValue().getSqlSet()).contains("gmt_modified");
+ assertThat(updateCaptor.getValue().getParamNameValuePairs())
+ .containsValue(UTC_CLOCK_INSTANT);
+ }
+
@Test
void passwordChangeRevokesExistingSessions() {
RmqStudioUser user = user(1L, "operator", false, true, "password-1");
@@ -708,7 +758,92 @@ class AuthServiceDatabaseTest {
assertThat(authService.isAuthenticated("Bearer token-1")).isTrue();
- verify(sessionMapper).update(isNull(), any(Wrapper.class));
+ org.mockito.ArgumentCaptor<UpdateWrapper<RmqStudioSession>>
updateCaptor =
+ org.mockito.ArgumentCaptor.forClass(UpdateWrapper.class);
+ verify(sessionMapper).update(isNull(), updateCaptor.capture());
+
assertThat(updateCaptor.getValue().getSqlSet()).contains("last_seen_at",
"gmt_modified");
+
assertThat(updateCaptor.getValue().getParamNameValuePairs()).containsValue(UTC_CLOCK_INSTANT);
+ }
+
+ @Test
+ void userAndSessionUpdatesStampTheUtcAuditInstantTest() {
+ RmqStudioUser target = user(1L, "admin-one", true, true, "password-1");
+ RmqStudioUser other = user(2L, "admin-two", true, true, "password-1");
+ when(userMapper.selectById(1L)).thenReturn(target);
+
when(userMapper.selectList(any(Wrapper.class))).thenReturn(List.of(target,
other));
+
+ authService.setUserEnabled(1L, false);
+
+ // rmq_studio_user.gmt_modified is ON UPDATE CURRENT_TIMESTAMP, which
is the database
+ // session's zone; the API serializes every zone-less timestamp as
UTC, so the app has to
+ // write the instant itself.
+ org.mockito.ArgumentCaptor<RmqStudioUser> updateCaptor =
+ org.mockito.ArgumentCaptor.forClass(RmqStudioUser.class);
+ verify(userMapper).updateById(updateCaptor.capture());
+
assertThat(updateCaptor.getValue().getGmtModified()).isEqualTo(UTC_CLOCK_INSTANT);
+ }
+
+ @Test
+ void revokingSessionsStampsTheUtcModifiedInstantTest() {
+ RmqStudioUser user = user(1L, "operator", false, true, "password-1");
+ when(userMapper.selectById(1L)).thenReturn(user);
+ when(sessionMapper.update(isNull(), any(Wrapper.class))).thenReturn(2);
+
+ authService.revokeSessionsForUser(1L);
+
+ org.mockito.ArgumentCaptor<UpdateWrapper<RmqStudioSession>>
updateCaptor =
+ org.mockito.ArgumentCaptor.forClass(UpdateWrapper.class);
+ verify(sessionMapper).update(isNull(), updateCaptor.capture());
+ assertThat(updateCaptor.getValue().getSqlSet()).contains("revoked_at",
"gmt_modified");
+
assertThat(updateCaptor.getValue().getParamNameValuePairs()).containsValue(UTC_CLOCK_INSTANT);
+ }
+
+ @Test
+ void loggingOutStampsTheUtcModifiedInstantTest() {
+ authService.logout("Bearer token-1");
+
+ org.mockito.ArgumentCaptor<UpdateWrapper<RmqStudioSession>>
updateCaptor =
+ org.mockito.ArgumentCaptor.forClass(UpdateWrapper.class);
+ verify(sessionMapper).update(isNull(), updateCaptor.capture());
+ assertThat(updateCaptor.getValue().getSqlSet()).contains("revoked_at",
"gmt_modified");
+
assertThat(updateCaptor.getValue().getParamNameValuePairs()).containsValue(UTC_CLOCK_INSTANT);
+ }
+
+ @Test
+ void revokingEverySessionStampsTheUtcModifiedInstantTest() {
+ authService.revokeAllSessions();
+
+ org.mockito.ArgumentCaptor<UpdateWrapper<RmqStudioSession>>
updateCaptor =
+ org.mockito.ArgumentCaptor.forClass(UpdateWrapper.class);
+ verify(sessionMapper).update(isNull(), updateCaptor.capture());
+ assertThat(updateCaptor.getValue().getSqlSet()).contains("revoked_at",
"gmt_modified");
+ }
+
+ @Test
+ void revokingASessionByIdStampsTheUtcModifiedInstantTest() {
+ RmqStudioSession session = activeSession(19L, 7L,
+ LocalDateTime.parse("2026-08-13T00:00:00"));
+ when(sessionMapper.selectById(19L)).thenReturn(session);
+
+ authService.revokeSessionById(19L);
+
+ org.mockito.ArgumentCaptor<UpdateWrapper<RmqStudioSession>>
updateCaptor =
+ org.mockito.ArgumentCaptor.forClass(UpdateWrapper.class);
+ verify(sessionMapper).update(isNull(), updateCaptor.capture());
+ assertThat(updateCaptor.getValue().getSqlSet()).contains("revoked_at",
"gmt_modified");
+ }
+
+ @Test
+ void grantingTheAdministratorRoleStampsTheUtcAuditInstantTest() {
+ RmqStudioUser operator = user(2L, "operator", false, true,
"password-1");
+ when(userMapper.selectById(2L)).thenReturn(operator);
+
+ authService.setUserAdmin(2L, true);
+
+ org.mockito.ArgumentCaptor<RmqStudioUser> updateCaptor =
+ org.mockito.ArgumentCaptor.forClass(RmqStudioUser.class);
+ verify(userMapper).updateById(updateCaptor.capture());
+
assertThat(updateCaptor.getValue().getGmtModified()).isEqualTo(UTC_CLOCK_INSTANT);
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclControllerTest.java
index 388d6e7de..b4001fc36 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclControllerTest.java
@@ -109,7 +109,7 @@ class AclControllerTest extends WebMvcAuthTestSupport {
rule.setId(1L);
rule.setGmtCreate(LocalDateTime.of(2026, 1, 1, 0, 0));
- when(aclService.listRules(isNull(), isNull(), isNull(), isNull(),
isNull(), eq(1), eq(20)))
+ when(aclService.listRules(isNull(), isNull(), isNull(), isNull(),
isNull(), isNull(), eq(1), eq(20)))
.thenReturn(PageResult.of(java.util.List.of(rule), 1, 1, 20));
mockMvc.perform(get("/api/acl/rules"))
@@ -127,7 +127,7 @@ class AclControllerTest extends WebMvcAuthTestSupport {
@Test
void listRulesShouldPassQueryParams() throws Exception {
when(aclService.listRules(eq("user1"), eq("topic-a"), eq("namespace"),
eq("DENY"),
- isNull(), eq(3), eq(5))).thenReturn(PageResult.empty(3, 5));
+ isNull(), isNull(), eq(3),
eq(5))).thenReturn(PageResult.empty(3, 5));
mockMvc.perform(get("/api/acl/rules")
.param("principal", "user1")
@@ -142,12 +142,25 @@ class AclControllerTest extends WebMvcAuthTestSupport {
.andExpect(jsonPath("$.data.size").value(5));
verify(aclService).listRules(eq("user1"), eq("topic-a"),
eq("namespace"), eq("DENY"),
- isNull(), eq(3), eq(5));
+ isNull(), isNull(), eq(3), eq(5));
+ }
+
+ @Test
+ void listRulesShouldApplyTheAclVersionFilterTest() throws Exception {
+ when(aclService.listRules(isNull(), isNull(), isNull(), isNull(),
eq("1.0"),
+ isNull(), eq(1), eq(20))).thenReturn(PageResult.empty(1, 20));
+
+ mockMvc.perform(get("/api/acl/rules").param("aclVersion", "1.0"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200));
+
+ verify(aclService).listRules(isNull(), isNull(), isNull(), isNull(),
eq("1.0"),
+ isNull(), eq(1), eq(20));
}
@Test
void listRulesShouldRejectPageSizeAboveTheInventoryLimit() throws
Exception {
- when(aclService.listRules(isNull(), isNull(), isNull(), isNull(),
isNull(),
+ when(aclService.listRules(isNull(), isNull(), isNull(), isNull(),
isNull(), isNull(),
eq(1), eq(101))).thenThrow(new BusinessException(400,
"page must be >= 1 and pageSize must be between 1 and 100"));
@@ -159,7 +172,7 @@ class AclControllerTest extends WebMvcAuthTestSupport {
.andExpect(jsonPath("$.message")
.value("page must be >= 1 and pageSize must be between
1 and 100"));
- verify(aclService).listRules(isNull(), isNull(), isNull(), isNull(),
isNull(),
+ verify(aclService).listRules(isNull(), isNull(), isNull(), isNull(),
isNull(), isNull(),
eq(1), eq(101));
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
index 0547fa804..71915a5aa 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
@@ -129,6 +129,38 @@ class AclServiceTest {
verify(aclRepository).findRulePage("user1", "topic", "cluster",
"ALLOW", null, 2, 5);
}
+ @Test
+ void listRulesShouldForwardTheAclVersionFilterToTheRepositoryTest() {
+ when(aclRepository.findRulePage(null, null, null, null, "1.0", 1, 20))
+ .thenReturn(PageResult.of(List.of(
+
AclRuleVO.builder().principal("user1").aclVersion("1.0").build()), 1, 1, 20));
+
+ PageResult<AclRuleVO> result = aclService.listRules(null, null, null,
null, "1.0",
+ null, 1, 20);
+
+ assertThat(result.getItems()).hasSize(1);
+ verify(aclRepository).findRulePage(null, null, null, null, "1.0", 1,
20);
+ }
+
+ @Test
+ void listRulesShouldFilterTencentRulesByAclVersionTest() {
+ InstanceVO instance = InstanceVO.builder()
+ .name("tencent-instance")
+ .vendor(InstanceVendor.TENCENT)
+ .type(InstanceType.CLOUD)
+ .build();
+
when(instanceResolver.findByIdentifier("tencent-instance")).thenReturn(Optional.of(instance));
+ when(tencentAclService.listRules("tencent-instance",
null)).thenReturn(List.of(
+
AclRuleVO.builder().principal("role-legacy").aclVersion("1.0").build(),
+
AclRuleVO.builder().principal("role-acl2").aclVersion("2.0").build()));
+
+ PageResult<AclRuleVO> result = aclService.listRules(null, null, null,
null, "2.0",
+ "tencent-instance", 1, 20);
+
+
assertThat(result.getItems()).extracting(AclRuleVO::getPrincipal).containsExactly("role-acl2");
+ assertThat(result.getTotal()).isEqualTo(1);
+ }
+
@Test
void capabilitiesShouldDescribeApacheRemoteReadSupport() {
InstanceVO instance = InstanceVO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationTemplateTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationTemplateTest.java
index 032fe9aad..e3730ca11 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationTemplateTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertNotificationTemplateTest.java
@@ -32,6 +32,33 @@ class AlertNotificationTemplateTest {
+ "brokerAddr=127.0.0.1:10911,
brokerName=broker-a|${missing}");
}
+ @Test
+ void rendersLargeValuesAndThresholdsWithoutScientificNotationTest() {
+ AlertRuleVO rule = AlertRuleVO.builder().name("Consumer
lag").metric("consumer.lag.total")
+ .threshold(12000000).thresholdUnit("messages").build();
+ SystemAlertVO alert =
SystemAlertVO.builder().level(AlertLevel.warning).title("Consumer lag")
+ .description("FIRING").transition("FIRING").instanceId("local")
+ .currentValue(12500000.0).time(LocalDateTime.of(2026, 8, 23,
12, 0))
+ .labels(Map.of()).build();
+
+ String rendered =
AlertNotificationTemplate.render("${value}/${threshold}${thresholdUnit}",
alert, rule);
+
+ // Double.toString would deliver the same alert as
"1.25E7/1.2E7messages".
+ assertThat(rendered).isEqualTo("12500000/12000000messages");
+ }
+
+ @Test
+ void rendersNonFiniteValuesWithoutAPlainDecimalFormTest() {
+ AlertRuleVO rule = AlertRuleVO.builder().name("Consumer
lag").metric("consumer.lag.total")
+ .threshold(100).thresholdUnit("messages").build();
+ SystemAlertVO alert =
SystemAlertVO.builder().level(AlertLevel.warning).title("Consumer lag")
+ .description("FIRING").transition("FIRING").instanceId("local")
+
.currentValue(Double.POSITIVE_INFINITY).time(LocalDateTime.of(2026, 8, 23, 12,
0))
+ .labels(Map.of()).build();
+
+ assertThat(AlertNotificationTemplate.render("${value}", alert,
rule)).isEqualTo("Infinity");
+ }
+
@Test
void rendersPercentageValuesForPaddedStoredMetricsTest() {
AlertRuleVO rule = AlertRuleVO.builder().name("Disk
threshold").metric(" broker.disk.usage_ratio ")
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 02361644c..32e87af0c 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -111,6 +111,11 @@ export interface ConsumerGroup {
gmtCreate: string;
gmtModified: string;
delaySeconds: number;
+ /**
+ * Whether the broker consume stats carried a consumed-message timestamp.
When false,
+ * `delaySeconds` is the provider's placeholder zero, not a measured delay.
+ */
+ consumptionTimestampAvailable?: boolean;
instances: ConsumerInstance[];
}
diff --git a/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
b/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
index 286a90af6..3a9bb775a 100644
--- a/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
+++ b/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
@@ -404,6 +404,56 @@ describe('Cluster page', () => {
expect(clusterServiceMocks.restartProxy).toHaveBeenCalledTimes(1);
});
+ it('reports a failed broker runtime-stats read as unavailable instead of 0%
and 0 TPS', async () => {
+ const cluster = buildCluster();
+ clusterServiceMocks.listRegistryClusters.mockResolvedValue([
+ {
+ ...cluster,
+ brokers: [
+ {
+ ...cluster.brokers[0],
+ // The shape RocketMQClusterProvider leaves behind when the stats
read fails.
+ runtimeStatsAvailable: false,
+ diskUsage: 0,
+ tpsIn: 0,
+ tpsOut: 0,
+ },
+ // A second, measured broker: without one the order assertions below
could never fail.
+ {
+ ...cluster.brokers[1],
+ runtimeStatsAvailable: true,
+ diskUsage: 42,
+ tpsIn: 950_000,
+ tpsOut: 12_345,
+ },
+ ],
+ },
+ ]);
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ renderWithProviders(<ClusterPage />);
+
+ const brokerRow = await screen.findByRole('row', { name:
/10\.101\.2\.11:10911/ });
+ expect(within(brokerRow).queryByText('0%')).not.toBeInTheDocument();
+ expect(within(brokerRow).getAllByText('不可用')).toHaveLength(3);
+ const measuredRow = screen.getByRole('row', { name: /10\.101\.2\.12:10911/
});
+ expect(within(measuredRow).getByText('950,000')).toBeInTheDocument();
+ expect(within(measuredRow).queryByText('不可用')).not.toBeInTheDocument();
+
+ const tpsInHeader = screen.getAllByText('TPS In')[0];
+ const brokerRowOrder = () =>
+ Array.from(document.querySelectorAll('tbody tr')).map((row) =>
row.textContent ?? '');
+ const unmeasuredIndex = () =>
+ brokerRowOrder().findIndex((text) => text.includes('10.101.2.11:10911'));
+ const measuredIndex = () =>
+ brokerRowOrder().findIndex((text) => text.includes('10.101.2.12:10911'));
+
+ // Neither direction may promote the unmeasured broker above the measured
one.
+ await user.click(tpsInHeader);
+ expect(measuredIndex()).toBeLessThan(unmeasuredIndex());
+ await user.click(tpsInHeader);
+ expect(measuredIndex()).toBeLessThan(unmeasuredIndex());
+ });
+
it('centers the broker status column', async () => {
renderWithProviders(<ClusterPage />);
const brokerRow = await screen.findByRole('row', { name:
/10\.101\.2\.11:10911/ });
diff --git a/web/src/pages/cluster/index.tsx b/web/src/pages/cluster/index.tsx
index babd276db..e2514f82e 100644
--- a/web/src/pages/cluster/index.tsx
+++ b/web/src/pages/cluster/index.tsx
@@ -42,6 +42,7 @@ import {
Col,
} from 'antd';
import type { ColumnsType } from 'antd/es/table';
+import type { SortOrder } from 'antd/es/table/interface';
import {
ReloadOutlined,
SettingOutlined,
@@ -1265,6 +1266,28 @@ const ClusterPage = () => {
nsClusterName: string;
cluster: ClusterInfo;
};
+ // A broker whose runtime stats read failed carries diskUsage/tpsIn/tpsOut
as Java zero
+ // defaults plus runtimeStatsAvailable=false
(BrokerVO/RocketMQClusterProvider). Rendering the
+ // zero claims an empty disk and no traffic, so the cell and the order
have to treat it as
+ // unknown - the same way the studio cluster page does.
+ const statsUnavailable = (broker: BrokerWithCluster) =>
broker.runtimeStatsAvailable === false;
+ const compareBrokerStats = (
+ left: BrokerWithCluster,
+ right: BrokerWithCluster,
+ field: 'diskUsage' | 'tpsIn' | 'tpsOut',
+ sortOrder?: SortOrder,
+ ): number => {
+ const leftUnavailable = statsUnavailable(left);
+ const rightUnavailable = statsUnavailable(right);
+ if (leftUnavailable || rightUnavailable) {
+ if (leftUnavailable === rightUnavailable) return 0;
+ // Ant Design reverses the comparator for a descending column, so
invert this branch to
+ // keep unavailable brokers after measured ones in either order.
+ const unavailableAfterAvailable = sortOrder === 'descend' ? -1 : 1;
+ return leftUnavailable ? unavailableAfterAvailable :
-unavailableAfterAvailable;
+ }
+ return left[field] - right[field];
+ };
const brokerSearchText = searchText(brokerSearch);
const allBrokers: BrokerWithCluster[] = registryClusters.flatMap((c) =>
@@ -1342,14 +1365,17 @@ const ClusterPage = () => {
dataIndex: 'diskUsage',
key: 'diskUsage',
width: 150,
- sorter: (a, b) => a.diskUsage - b.diskUsage,
- render: (v: number) => (
- <Progress
- percent={roundPercent(v)}
- size="small"
- strokeColor={v > 85 ? '#ff4d4f' : v > 70 ? '#faad14' : '#1677ff'}
- />
- ),
+ sorter: (a, b, sortOrder) => compareBrokerStats(a, b, 'diskUsage',
sortOrder),
+ render: (v: number, record: BrokerWithCluster) =>
+ statsUnavailable(record) ? (
+ <Text type="secondary">{t('common.unavailable')}</Text>
+ ) : (
+ <Progress
+ percent={roundPercent(v)}
+ size="small"
+ strokeColor={v > 85 ? '#ff4d4f' : v > 70 ? '#faad14' : '#1677ff'}
+ />
+ ),
},
{
title: t('common.address'),
@@ -1366,8 +1392,13 @@ const ClusterPage = () => {
key: 'tpsIn',
width: 90,
align: 'right',
- sorter: (a, b) => a.tpsIn - b.tpsIn,
- render: (v: number) => v.toLocaleString(),
+ sorter: (a, b, sortOrder) => compareBrokerStats(a, b, 'tpsIn',
sortOrder),
+ render: (v: number, record: BrokerWithCluster) =>
+ statsUnavailable(record) ? (
+ <Text type="secondary">{t('common.unavailable')}</Text>
+ ) : (
+ v.toLocaleString()
+ ),
},
{
title: 'TPS Out',
@@ -1375,8 +1406,13 @@ const ClusterPage = () => {
key: 'tpsOut',
width: 90,
align: 'right',
- sorter: (a, b) => a.tpsOut - b.tpsOut,
- render: (v: number) => v.toLocaleString(),
+ sorter: (a, b, sortOrder) => compareBrokerStats(a, b, 'tpsOut',
sortOrder),
+ render: (v: number, record: BrokerWithCluster) =>
+ statsUnavailable(record) ? (
+ <Text type="secondary">{t('common.unavailable')}</Text>
+ ) : (
+ v.toLocaleString()
+ ),
},
{
title: t('common.actions'),
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 89f671747..cb77cabaf 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -2663,6 +2663,49 @@ describe('Consumer page', () => {
expect(within(dialog).getByText('不可用')).toBeInTheDocument();
});
+ it('reports a group without a consumed-message timestamp as an unknown
delay', async () => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ vi.mocked(consumerService.listConsumerGroupPage).mockResolvedValue(
+ groupPage([
+ {
+ ...group,
+ name: 'unknown-delay-cg',
+ delaySeconds: 0,
+ consumptionTimestampAvailable: false,
+ },
+ {
+ ...group,
+ name: 'known-delay-cg',
+ delaySeconds: 120,
+ consumptionTimestampAvailable: true,
+ },
+ ]),
+ );
+ renderWithProviders(<ConsumerPage />);
+
+ const unknownRow = await screen.findByRole('row', { name:
/unknown-delay-cg/ });
+ // The provider's placeholder zero must not be presented as "caught up".
+ expect(within(unknownRow).getByText('不可用')).toBeInTheDocument();
+ expect(within(unknownRow).queryByText('0秒')).not.toBeInTheDocument();
+ await user.click(within(unknownRow).getByRole('button', { name: /详\s*情/
}));
+ const dialog = await screen.findByRole('dialog', { name:
/unknown-delay-cg/ });
+ expect(within(dialog).getByText('不可用')).toBeInTheDocument();
+
+ // ... and neither sort direction may promote it.
+ const [delayHeader] = screen.getAllByText('消费延迟');
+ const rowTexts = () =>
+ Array.from(document.querySelectorAll('tbody tr')).map((row) =>
row.textContent ?? '');
+ const unknownIndex = () => rowTexts().findIndex((text) =>
text.includes('unknown-delay-cg'));
+ const knownIndex = () => rowTexts().findIndex((text) =>
/\bknown-delay-cg\b/.test(text));
+
+ // Ascending: pins the MAX_SAFE_INTEGER sentinel that keeps an unmeasured
row last.
+ await user.click(delayHeader);
+ await waitFor(() => expect(unknownIndex()).toBeGreaterThan(knownIndex()));
+ // Descending: antd negates the comparator, so the sentinel has to flip
with the direction.
+ await user.click(delayHeader);
+ await waitFor(() => expect(unknownIndex()).toBeGreaterThan(knownIndex()));
+ });
+
it('sorts groups with an unknown lag after known backlogs in lag order',
async () => {
const user = userEvent.setup({ pointerEventsCheck: 0 });
vi.mocked(consumerService.listConsumerGroupPage).mockResolvedValue(
@@ -2692,6 +2735,36 @@ describe('Consumer page', () => {
});
});
+ it('keeps an unknown lag after known backlogs in the descending order too',
async () => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ vi.mocked(consumerService.listConsumerGroupPage).mockResolvedValue(
+ groupPage([
+ { ...group, name: 'unknown-lag-cg', totalLag: -1 },
+ { ...group, name: 'known-lag-cg', totalLag: 15000 },
+ ]),
+ );
+ renderWithProviders(<ConsumerPage />);
+ await screen.findByRole('row', { name: /unknown-lag-cg/ });
+
+ const [lagHeader] = screen.getAllByText('总堆积量');
+ // Ascending, then descending: the row with no measurement must not lead
either list.
+ await user.click(lagHeader);
+ await user.click(lagHeader);
+ await waitFor(() => {
+ const order = Array.from(document.querySelectorAll('tbody tr'))
+ .map((row) => row.textContent ?? '')
+ .map((text) =>
+ text.includes('unknown-lag-cg')
+ ? 'unknown'
+ : /\bknown-lag-cg\b/.test(text)
+ ? 'known'
+ : '',
+ )
+ .filter(Boolean);
+ expect(order).toEqual(['known', 'unknown']);
+ });
+ });
+
it('renders an unavailable connection inventory as zero and sorts it as
zero', async () => {
const user = userEvent.setup({ pointerEventsCheck: 0 });
vi.mocked(consumerService.listConsumerGroupPage).mockResolvedValue(
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index 1a02b099a..35925203d 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -15,6 +15,7 @@
* limitations under the License.
*/
+import type { SortOrder } from 'antd/es/table/interface';
import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
import { useSearchParams } from 'react-router-dom';
import {
@@ -121,6 +122,24 @@ const { Text } = Typography;
const UNKNOWN_LAG_COLOR = '#8c8c8c';
const UNAVAILABLE_LAG_LABEL = '不可用';
+/**
+ * The provider leaves delaySeconds at zero when the broker consume stats
carried no
+ * consumed-message timestamp (RocketMQMetadataProvider: "keep zeros") and
says so through
+ * consumptionTimestampAvailable. Rendering that zero claims the group is
caught up, so the
+ * column and its order have to treat it as unknown.
+ */
+const isDelayAvailable = (group: ConsumerGroup): boolean =>
+ group.consumptionTimestampAvailable !== false;
+
+const delaySortValue = (group: ConsumerGroup, sortOrder?: SortOrder): number
=> {
+ if (isDelayAvailable(group) && Number.isFinite(group.delaySeconds)) {
+ return group.delaySeconds;
+ }
+ // Ant Design negates the comparator for a descending column, so the
sentinel has to flip with
+ // the direction to keep unmeasured rows after measured ones in both orders.
+ return sortOrder === 'descend' ? Number.MIN_SAFE_INTEGER :
Number.MAX_SAFE_INTEGER;
+};
+
const lagColor = (lag: number): string => {
// The backend reports -1 when the lag cannot be determined; do not color it
// as healthy (green) or backlogged.
@@ -1044,7 +1063,8 @@ const ConsumerPageContent = ({
key: 'totalLag',
width: 96,
align: 'right',
- sorter: (a, b) => lagSortValue(a.totalLag) - lagSortValue(b.totalLag),
+ sorter: (a, b, sortOrder) =>
+ lagSortValue(a.totalLag, sortOrder) - lagSortValue(b.totalLag,
sortOrder),
render: (lag: number) =>
isLagAvailable(lag) ? (
lag.toLocaleString()
@@ -1058,8 +1078,13 @@ const ConsumerPageContent = ({
key: 'delaySeconds',
width: 100,
align: 'right',
- sorter: (a, b) => (a.delaySeconds ?? 0) - (b.delaySeconds ?? 0),
- render: (seconds: number) => formatDelay(seconds ?? 0, lang),
+ sorter: (a, b, sortOrder) => delaySortValue(a, sortOrder) -
delaySortValue(b, sortOrder),
+ render: (seconds: number, record: ConsumerGroup) =>
+ isDelayAvailable(record) ? (
+ formatDelay(seconds ?? 0, lang)
+ ) : (
+ <Text type="secondary">{UNAVAILABLE_LAG_LABEL}</Text>
+ ),
},
{
title: '创建时间',
@@ -1859,7 +1884,11 @@ const ConsumerPageContent = ({
</Tag>
</Descriptions.Item>
<Descriptions.Item label="消费延迟">
- <Text strong>{formatDelay(selectedGroup.delaySeconds,
lang)}</Text>
+ <Text strong>
+ {isDelayAvailable(selectedGroup)
+ ? formatDelay(selectedGroup.delaySeconds, lang)
+ : UNAVAILABLE_LAG_LABEL}
+ </Text>
</Descriptions.Item>
<Descriptions.Item label="最大重试次数">
<Text strong>{selectedGroup.retryMaxTimes}</Text> 次
diff --git a/web/src/services/consumerService.test.ts
b/web/src/services/consumerService.test.ts
index bfec41bfc..aee2c601e 100644
--- a/web/src/services/consumerService.test.ts
+++ b/web/src/services/consumerService.test.ts
@@ -550,6 +550,37 @@ describe('consumer service mock data', () => {
});
describe('consumer group CSV export', () => {
+ it('renders an unmeasurable consume delay as unknown instead of the
placeholder zero', async () => {
+ const unmeasured = {
+ ...mockConsumerGroups[0],
+ name: 'cg-delay-unmeasured',
+ namespace: 'ns',
+ clusterId: 'cluster-a',
+ subscriptionMode: 'Push' as const,
+ consumeType: 'CLUSTERING' as const,
+ delaySeconds: 0,
+ consumptionTimestampAvailable: false,
+ };
+ const measuredZero = {
+ ...unmeasured,
+ name: 'cg-delay-zero',
+ consumptionTimestampAvailable: true,
+ };
+ mockConsumerGroups.push(unmeasured, measuredZero);
+ try {
+ const csv = await exportConsumerGroups({ names: ['cg-delay-unmeasured',
'cg-delay-zero'] });
+
+ expect(csv).toContain(
+
'"cg-delay-unmeasured","ns","cluster-a","Push","CLUSTERING","0","0","unknown"',
+ );
+ // A measured zero must stay numeric, otherwise the column loses its
meaning entirely.
+
expect(csv).toContain('"cg-delay-zero","ns","cluster-a","Push","CLUSTERING","0","0","0"');
+ } finally {
+ mockConsumerGroups.pop();
+ mockConsumerGroups.pop();
+ }
+ });
+
it('exports an unavailable total lag as unknown while preserving measured
values', async () => {
const unavailable = {
...mockConsumerGroups[0],
diff --git a/web/src/services/consumerService.ts
b/web/src/services/consumerService.ts
index 8381f33a9..5b3fd0f8c 100644
--- a/web/src/services/consumerService.ts
+++ b/web/src/services/consumerService.ts
@@ -53,7 +53,13 @@ const GROUP_EXPORT_COLUMNS: CsvColumn<ConsumerGroup>[] = [
header: 'Total Lag',
value: (group) => (isLagAvailable(group.totalLag) ? group.totalLag :
'unknown'),
},
- { header: 'Delay Seconds', value: (group) => group.delaySeconds },
+ {
+ header: 'Delay Seconds',
+ // The same placeholder zero as behind the table column: an unmeasured
delay must not be
+ // exported as "0", which reads as caught up.
+ value: (group) =>
+ group.consumptionTimestampAvailable === false ? 'unknown' :
group.delaySeconds,
+ },
{ header: 'Subscription Data Type', value: (group) =>
group.subscriptionDataType },
{ header: 'Delivery Order Type', value: (group) => group.deliveryOrderType },
{ header: 'Retry Max Times', value: (group) => group.retryMaxTimes },
diff --git a/web/src/utils/consumerGroupDiagnostics.test.ts
b/web/src/utils/consumerGroupDiagnostics.test.ts
index 3eaa205fe..19f6cce89 100644
--- a/web/src/utils/consumerGroupDiagnostics.test.ts
+++ b/web/src/utils/consumerGroupDiagnostics.test.ts
@@ -183,4 +183,34 @@ describe('consumer group diagnostics', () => {
expect.arrayContaining(['SUBSCRIPTION_UNKNOWN', 'UNKNOWN_QUEUE_LAG',
'STALE_HEARTBEAT']),
);
});
+
+ it('reports an unmeasurable consume delay as unknown instead of a healthy
zero', () => {
+ const diagnostics = analyzeConsumerGroupHealth(
+ group({ delaySeconds: 0, consumptionTimestampAvailable: false, totalLag:
0 }),
+ [subscription({})],
+ [queue({ diffTotal: 0 })],
+ { now: '2026-08-31T12:01:00Z' },
+ );
+
+ // The provider leaves delaySeconds at zero when the broker stats carried
no consumed-message
+ // timestamp; judging that zero reported the group as caught up rather
than unmeasurable.
+ expect(diagnostics.issues.map((item) =>
item.code)).toContain('CONSUME_DELAY_UNKNOWN');
+ expect(diagnostics.issues.map((item) =>
item.code)).not.toContain('HIGH_CONSUME_DELAY');
+ expect(diagnostics.status).toBe('warning');
+ expect(diagnostics.recommendations).toContain(
+ '消费延迟不可用时，确认 Broker 消费统计可读且消费端仍在提交 offset。',
+ );
+ });
+
+ it('keeps judging a measured delay', () => {
+ const diagnostics = analyzeConsumerGroupHealth(
+ group({ delaySeconds: 1_900, consumptionTimestampAvailable: true,
totalLag: 0 }),
+ [subscription({})],
+ [queue({ diffTotal: 0 })],
+ { now: '2026-08-31T12:01:00Z' },
+ );
+
+ expect(diagnostics.issues.map((item) =>
item.code)).toContain('HIGH_CONSUME_DELAY');
+ expect(diagnostics.issues.map((item) =>
item.code)).not.toContain('CONSUME_DELAY_UNKNOWN');
+ });
});
diff --git a/web/src/utils/consumerGroupDiagnostics.ts
b/web/src/utils/consumerGroupDiagnostics.ts
index c7a3f6883..e402512dc 100644
--- a/web/src/utils/consumerGroupDiagnostics.ts
+++ b/web/src/utils/consumerGroupDiagnostics.ts
@@ -29,6 +29,7 @@ export type ConsumerGroupHealthIssueCode =
| 'QUEUE_LAG_SKEW'
| 'HIGH_GROUP_LAG'
| 'HIGH_CONSUME_DELAY'
+ | 'CONSUME_DELAY_UNKNOWN'
| 'UNKNOWN_QUEUE_LAG'
| 'STALE_HEARTBEAT';
@@ -340,25 +341,38 @@ const runtimeIssues = (
}
}
- const delaySeconds = Number.isFinite(group.delaySeconds) ?
group.delaySeconds : 0;
- if (delaySeconds >= options.criticalDelaySeconds) {
+ if (group.consumptionTimestampAvailable === false) {
+ // delaySeconds is the provider's placeholder zero here, not a
measurement, so judging it
+ // would report an unmeasurable group as caught up. Say the delay is
unknown instead.
issues.push(
issue(
- 'HIGH_CONSUME_DELAY',
- 'critical',
- '消费延迟过高',
- `Group 当前消费延迟约 ${delaySeconds.toLocaleString()} 秒，业务可能已经感知延迟。`,
- ),
- );
- } else if (delaySeconds >= options.highDelaySeconds) {
- issues.push(
- issue(
- 'HIGH_CONSUME_DELAY',
+ 'CONSUME_DELAY_UNKNOWN',
'warning',
- '消费延迟偏高',
- `Group 当前消费延迟约 ${delaySeconds.toLocaleString()} 秒，建议继续观察趋势。`,
+ '消费延迟不可用',
+ 'Broker 消费统计未返回已消费消息时间戳，无法判断该 Group 的消费延迟。',
),
);
+ } else {
+ const delaySeconds = Number.isFinite(group.delaySeconds) ?
group.delaySeconds : 0;
+ if (delaySeconds >= options.criticalDelaySeconds) {
+ issues.push(
+ issue(
+ 'HIGH_CONSUME_DELAY',
+ 'critical',
+ '消费延迟过高',
+ `Group 当前消费延迟约 ${delaySeconds.toLocaleString()} 秒，业务可能已经感知延迟。`,
+ ),
+ );
+ } else if (delaySeconds >= options.highDelaySeconds) {
+ issues.push(
+ issue(
+ 'HIGH_CONSUME_DELAY',
+ 'warning',
+ '消费延迟偏高',
+ `Group 当前消费延迟约 ${delaySeconds.toLocaleString()} 秒，建议继续观察趋势。`,
+ ),
+ );
+ }
}
return issues;
};
@@ -385,6 +399,9 @@ const recommendations = (issues:
ConsumerGroupHealthIssue[]): string[] => {
if (codes.has('UNKNOWN_QUEUE_LAG')) {
result.push('当堆积不可用时，优先确认 Proxy 指标采集和 Broker offset 查询权限。');
}
+ if (codes.has('CONSUME_DELAY_UNKNOWN')) {
+ result.push('消费延迟不可用时，确认 Broker 消费统计可读且消费端仍在提交 offset。');
+ }
return result;
};
diff --git a/web/src/utils/consumerLag.test.ts
b/web/src/utils/consumerLag.test.ts
index 8d3da77a3..96a799c9e 100644
--- a/web/src/utils/consumerLag.test.ts
+++ b/web/src/utils/consumerLag.test.ts
@@ -42,4 +42,12 @@ describe('consumer lag helpers', () => {
expect(lagSortValue(UNKNOWN_LAG)).toBe(Number.MAX_SAFE_INTEGER);
expect(lagSortValue(null)).toBe(Number.MAX_SAFE_INTEGER);
});
+
+ it('keeps unknown lags after known ones in the descending order too', () => {
+ // Ant Design negates the comparator for a descending column, so an
unknown row would land
+ // first if the sentinel stayed the largest number.
+ expect(lagSortValue(UNKNOWN_LAG, 'descend')).toBe(Number.MIN_SAFE_INTEGER);
+ expect(lagSortValue(null, 'descend')).toBe(Number.MIN_SAFE_INTEGER);
+ expect(lagSortValue(15000, 'descend')).toBe(15000);
+ });
});
diff --git a/web/src/utils/consumerLag.ts b/web/src/utils/consumerLag.ts
index ffb6e8392..052a73bc3 100644
--- a/web/src/utils/consumerLag.ts
+++ b/web/src/utils/consumerLag.ts
@@ -28,6 +28,17 @@ export const formatLag = (
unavailableLabel: string = String(UNKNOWN_LAG),
): string => (isLagAvailable(lag) ? lag.toLocaleString() : unavailableLabel);
-/** Sort key that pushes unknown lags to the end of an ascending list. */
-export const lagSortValue = (lag: number | null | undefined): number =>
- isLagAvailable(lag) ? lag : Number.MAX_SAFE_INTEGER;
+/**
+ * Sort key that keeps an unknown lag after every known value in both sort
directions.
+ *
+ * Ant Design computes a descending column by negating the comparator's
result, so an unknown
+ * value cannot simply be the largest number: the sentinel has to flip with
the direction, or the
+ * descending order lists the rows that carry no measurement first.
+ */
+export const lagSortValue = (
+ lag: number | null | undefined,
+ sortOrder?: 'ascend' | 'descend' | null,
+): number => {
+ if (isLagAvailable(lag)) return lag;
+ return sortOrder === 'descend' ? Number.MIN_SAFE_INTEGER :
Number.MAX_SAFE_INTEGER;
+};
