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 0ea72dbf0 feat(message): store query result snapshots and simplify
topic/consumer pages (#2839)
0ea72dbf0 is described below
commit 0ea72dbf0f1896a80e63f95b3e40cb567c13467c
Author: lizhimins <[email protected]>
AuthorDate: Mon Aug 31 16:36:08 2026 +0800
feat(message): store query result snapshots and simplify topic/consumer
pages (#2839)
---
.../studio/instance/message/MessageService.java | 7 +-
.../instance/message/QueryHistoryController.java | 8 ++
.../instance/message/QueryHistoryService.java | 68 +++++++++-
.../studio/ops/alert/AlertSchemaMigration.java | 3 +-
.../studio/persistence/entity/RmqMessageQuery.java | 2 +
server/src/main/resources/db/schema.sql | 1 +
.../instance/message/MessageServiceTest.java | 4 +-
.../QueryHistoryServiceIntegrationTest.java | 2 +-
.../instance/message/QueryHistoryServiceTest.java | 5 +-
web/src/api/messageHistory.ts | 21 +++
.../pages/instance/__tests__/ConsumerPage.test.tsx | 29 -----
web/src/pages/instance/consumer.tsx | 36 +-----
web/src/pages/instance/message.tsx | 40 ++++--
web/src/pages/instance/topic.tsx | 143 +++++----------------
14 files changed, 172 insertions(+), 197 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
index 78aef0178..e676cfd64 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
@@ -54,7 +54,7 @@ public class MessageService {
.map(provider -> provider.queryMessages(instanceId, topic,
msgId, tag, key, startTime, endTime))
.orElseGet(() -> messageProvider.queryMessages(instanceId,
topic, msgId, tag, key, startTime, endTime));
if (recordHistory) {
- recordMessageQuery(instanceId, topic, msgId, tag, key, startTime,
endTime, result.size());
+ recordMessageQuery(instanceId, topic, msgId, tag, key, startTime,
endTime, result);
}
return result;
}
@@ -151,11 +151,12 @@ public class MessageService {
.orElseGet(() ->
messageProvider.getMessageTraceByKey(instanceId, key, topic, traceTopic));
}
private void recordMessageQuery(String instanceId, String topic, String
msgId, String tag,
- String key, Long startTime, Long endTime,
int resultCount) {
+ String key, Long startTime, Long endTime,
List<MessageRecordVO> result) {
String queryType = StringUtils.hasText(msgId) ? "MSG_ID" :
StringUtils.hasText(key) ? "KEY" : "TOPIC";
try {
+ String snapshot = queryHistoryService.buildResultSnapshot(result);
queryHistoryService.recordMessageQuery(instanceId, queryType,
topic, msgId, tag, key,
- startTime, endTime, resultCount);
+ startTime, endTime, result.size(), snapshot);
} catch (RuntimeException failure) {
log.warn("Failed to record message query history: {}",
failure.getMessage());
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryController.java
index 017e4290e..c43258374 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryController.java
@@ -11,11 +11,14 @@ import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.domain.Result;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.util.StringUtils;
+import java.util.List;
+
@RestController
@RequestMapping("/api/query-history")
@RequiredArgsConstructor
@@ -52,6 +55,11 @@ public class QueryHistoryController {
return
Result.ok(queryHistoryService.summarize(normalizeFilter(clusterId)));
}
+ @GetMapping("/messages/{id}/results")
+ public Result<List<MessageRecordVO>> messageQueryResults(@PathVariable
long id) {
+ return Result.ok(queryHistoryService.getMessageQueryResults(id));
+ }
+
private void validatePage(int page, int pageSize) {
if (page < 1) {
throw new BusinessException(400, "page must be at least 1");
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
index 0bb21e87f..699d83797 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
@@ -19,6 +19,8 @@ package org.apache.rocketmq.studio.instance.message;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.apache.rocketmq.studio.auth.AuthenticatedUserContext;
@@ -33,7 +35,9 @@ import org.springframework.util.StringUtils;
import java.time.Clock;
import java.time.LocalDateTime;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
@Slf4j
@Service
@@ -43,27 +47,31 @@ public class QueryHistoryService {
private final RmqTraceQueryMapper traceQueryMapper;
private final QueryHistoryProperties properties;
private final Clock clock;
+ private final ObjectMapper objectMapper;
@Autowired
public QueryHistoryService(RmqMessageQueryMapper messageQueryMapper,
RmqTraceQueryMapper traceQueryMapper,
- QueryHistoryProperties properties) {
- this(messageQueryMapper, traceQueryMapper, properties,
Clock.systemUTC());
+ QueryHistoryProperties properties,
+ ObjectMapper objectMapper) {
+ this(messageQueryMapper, traceQueryMapper, properties,
Clock.systemUTC(), objectMapper);
}
QueryHistoryService(RmqMessageQueryMapper messageQueryMapper,
RmqTraceQueryMapper traceQueryMapper,
QueryHistoryProperties properties,
- Clock clock) {
+ Clock clock,
+ ObjectMapper objectMapper) {
this.messageQueryMapper = messageQueryMapper;
this.traceQueryMapper = traceQueryMapper;
this.properties = properties;
this.clock = clock;
+ this.objectMapper = objectMapper;
}
public void recordMessageQuery(String clusterId, String queryType, String
topic, String msgId,
String tag, String key, Long startTime,
- Long endTime, int resultCount) {
+ Long endTime, int resultCount, String
resultSnapshot) {
RmqMessageQuery query = new RmqMessageQuery();
query.setQueryType(queryType);
query.setTopic(topic);
@@ -73,6 +81,7 @@ public class QueryHistoryService {
query.setStartTime(startTime);
query.setEndTime(endTime);
query.setResultCount(resultCount);
+ query.setResultSnapshot(resultSnapshot);
query.setClusterId(clusterId);
query.setQueriedBy(AuthenticatedUserContext.currentUsernameOrSystem());
LocalDateTime now = LocalDateTime.now(clock);
@@ -82,6 +91,57 @@ public class QueryHistoryService {
log.debug("Message query recorded: clusterId={} type={} topic={}",
clusterId, queryType, topic);
}
+ /**
+ * Builds a JSON snapshot of query results, excluding message body and
properties to save storage.
+ */
+ public String buildResultSnapshot(List<MessageRecordVO> results) {
+ if (results == null || results.isEmpty()) {
+ return null;
+ }
+ try {
+ List<Map<String, Object>> snapshots = results.stream().map(r -> {
+ Map<String, Object> m = new LinkedHashMap<>();
+ m.put("msgId", r.getMsgId() == null ? "" : r.getMsgId());
+ m.put("topic", r.getTopic() == null ? "" : r.getTopic());
+ m.put("tag", r.getTag() == null ? "" : r.getTag());
+ m.put("key", r.getKey() == null ? "" : r.getKey());
+ m.put("brokerName", r.getBrokerName() == null ? "" :
r.getBrokerName());
+ m.put("queueId", r.getQueueId() == null ? 0 : r.getQueueId());
+ m.put("queueOffset", r.getQueueOffset() == null ? 0L :
r.getQueueOffset());
+ m.put("storeTime", r.getStoreTime());
+ m.put("bornHost", r.getBornHost() == null ? "" :
r.getBornHost());
+ m.put("storeHost", r.getStoreHost() == null ? "" :
r.getStoreHost());
+ m.put("size", r.getSize());
+ return m;
+ }).toList();
+ return objectMapper.writeValueAsString(snapshots);
+ } catch (JsonProcessingException e) {
+ log.warn("Failed to serialize result snapshot: {}",
e.getMessage());
+ return null;
+ }
+ }
+
+ /**
+ * Retrieves the stored result snapshot for a given history record.
+ */
+ public List<MessageRecordVO> getMessageQueryResults(long id) {
+ RmqMessageQuery query = messageQueryMapper.selectById(id);
+ if (query == null) {
+ throw new
org.apache.rocketmq.studio.common.exception.BusinessException(404, "Query
history record not found");
+ }
+ String snapshot = query.getResultSnapshot();
+ if (!StringUtils.hasText(snapshot)) {
+ return List.of();
+ }
+ try {
+ return objectMapper.readValue(snapshot,
+
objectMapper.getTypeFactory().constructCollectionType(List.class,
MessageRecordVO.class));
+ } catch (JsonProcessingException e) {
+ log.warn("Failed to deserialize result snapshot for id={}: {}",
id, e.getMessage());
+ return List.of();
+ }
+ }
+
public void recordTraceQuery(String clusterId, String msgId, String topic,
int nodeCount, int consumerCount) {
RmqTraceQuery query = new RmqTraceQuery();
query.setMsgId(msgId);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertSchemaMigration.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertSchemaMigration.java
index ac2b18b90..b27eb1255 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertSchemaMigration.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertSchemaMigration.java
@@ -100,7 +100,8 @@ public class AlertSchemaMigration implements
ApplicationRunner {
new Column("rmq_system_alert", "labels_json", "TEXT"),
new Column("rmq_alert_notification_outbox", "sending_started_at",
"DATETIME"),
new Column("rmq_alert_notification_outbox", "claim_token",
"VARCHAR(64)"),
- new Column("rmq_alert_notification_outbox", "message_content",
"TEXT"));
+ new Column("rmq_alert_notification_outbox", "message_content",
"TEXT"),
+ new Column("rmq_instance_message", "result_snapshot",
"MEDIUMTEXT"));
private static final List<Index> INDEXES = List.of(
new Index("rmq_metric_snapshot", "idx_metric_snapshot_lookup",
"instance_id, metric_key, collected_at"),
new Index("rmq_metric_snapshot", "idx_metric_snapshot_retention",
"collected_at"),
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqMessageQuery.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqMessageQuery.java
index 27838aea4..195e6d324 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqMessageQuery.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqMessageQuery.java
@@ -46,6 +46,8 @@ public class RmqMessageQuery {
private Integer resultCount;
+ private String resultSnapshot;
+
private String clusterId;
private String queriedBy;
diff --git a/server/src/main/resources/db/schema.sql
b/server/src/main/resources/db/schema.sql
index fc058cd61..4cd62aa9c 100644
--- a/server/src/main/resources/db/schema.sql
+++ b/server/src/main/resources/db/schema.sql
@@ -145,6 +145,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_message (
start_time BIGINT,
end_time BIGINT,
result_count INT DEFAULT 0,
+ result_snapshot MEDIUMTEXT COMMENT '查询结果快照（不含消息体）JSON',
cluster_id VARCHAR(255),
queried_by VARCHAR(128),
PRIMARY KEY (`id`),
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
index 6dfc4dce3..1da40e26c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
@@ -124,7 +124,7 @@ class MessageServiceTest {
service.queryMessages("cloud-instance", "orders", null, null,
"ORDER-1", null, null);
verify(history).recordMessageQuery("cloud-instance", "KEY", "orders",
null, null,
- "ORDER-1", null, null, 1);
+ "ORDER-1", null, null, 1, null);
verifyNoInteractions(fallback);
}
@@ -205,7 +205,7 @@ class MessageServiceTest {
verify(provider, org.mockito.Mockito.times(2))
.queryMessages("instance-a", "TopicA", null, null, null,
1000L, 2000L);
verify(history, org.mockito.Mockito.times(1)).recordMessageQuery(
- "instance-a", "TOPIC", "TopicA", null, null, null, 1000L,
2000L, 1);
+ "instance-a", "TOPIC", "TopicA", null, null, null, 1000L,
2000L, 1, null);
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
index 31c9d6537..b369c9172 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
@@ -50,7 +50,7 @@ class QueryHistoryServiceIntegrationTest {
try {
AuthenticatedUserContext.setUser(longUsername, true);
queryHistoryService.recordMessageQuery("qh-cluster", "TOPIC",
"qh-topic",
- null, null, null, null, null, 3);
+ null, null, null, null, null, 3, null);
queryHistoryService.recordTraceQuery("qh-cluster", "qh-msg-id",
"qh-topic", 2, 1);
PageResult<MessageQueryHistoryVO> messageHistory =
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
index 4260c2702..5ed0445ee 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.instance.message;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.auth.AuthenticatedUserContext;
import org.apache.rocketmq.studio.persistence.entity.RmqMessageQuery;
@@ -47,7 +48,7 @@ class QueryHistoryServiceTest {
private final QueryHistoryProperties properties = new
QueryHistoryProperties();
private final Clock clock =
Clock.fixed(Instant.parse("2026-08-05T12:00:00Z"), ZoneOffset.UTC);
private final QueryHistoryService service = new QueryHistoryService(
- messageQueryMapper, traceQueryMapper, properties, clock);
+ messageQueryMapper, traceQueryMapper, properties, clock, new
ObjectMapper());
@AfterEach
void clearUserContext() {
@@ -59,7 +60,7 @@ class QueryHistoryServiceTest {
AuthenticatedUserContext.setUsername("alice");
service.recordMessageQuery("cluster-a", "TOPIC", "orders", null,
"tag-a", "key-a",
- 1L, 2L, 3);
+ 1L, 2L, 3, null);
ArgumentCaptor<RmqMessageQuery> captor =
ArgumentCaptor.forClass(RmqMessageQuery.class);
verify(messageQueryMapper).insert(captor.capture());
diff --git a/web/src/api/messageHistory.ts b/web/src/api/messageHistory.ts
index a48abe5e8..538052a86 100644
--- a/web/src/api/messageHistory.ts
+++ b/web/src/api/messageHistory.ts
@@ -78,3 +78,24 @@ export async function getQueryHistorySummary(clusterId?:
string) {
});
return response.data.data;
}
+
+export interface MessageResultSnapshot {
+ msgId: string;
+ topic: string;
+ tag: string;
+ key: string;
+ brokerName: string;
+ queueId: number;
+ queueOffset: number;
+ storeTime: number;
+ bornHost: string;
+ storeHost: string;
+ size: number;
+}
+
+export async function getMessageQueryResults(id: number) {
+ const response = await client.get<{ data: MessageResultSnapshot[] }>(
+ `/query-history/messages/${id}/results`,
+ );
+ return response.data.data;
+}
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 9f3a2470b..9d760867e 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -931,33 +931,4 @@ describe('Consumer page', () => {
const dialog = await screen.findByRole('dialog', { name: /unknown-lag-cg/
});
expect(within(dialog).getByText('不可用')).toBeInTheDocument();
});
-
- it('sorts groups with an unknown lag after known backlogs in lag order',
async () => {
- const user = userEvent.setup();
- 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/ });
-
- await user.click(screen.getByText('名称升序'));
- await user.click(await screen.findByText('堆积量降序'));
- await waitFor(() => {
- const rows = Array.from(document.querySelectorAll('tbody tr'));
- const order = rows
- .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']);
- });
- });
});
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index 588ed3523..69e5264ff 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -159,31 +159,13 @@ const GROUP_EXPORT_COLUMNS: CsvColumn<ConsumerGroup>[] = [
const buildConsumerGroupCsv = (groups: ConsumerGroup[]) =>
buildCsv(GROUP_EXPORT_COLUMNS, groups);
-const visibleConsumerGroups = (
- groups: ConsumerGroup[],
- modeFilter: string,
- sortKey: string,
-): ConsumerGroup[] => {
+const visibleConsumerGroups = (groups: ConsumerGroup[], modeFilter: string):
ConsumerGroup[] => {
let data = groups;
if (modeFilter !== 'ALL') {
data = data.filter((group) => group.subscriptionMode === modeFilter);
}
- if (sortKey === 'lag_desc') {
- // An unknown lag (-1) is not a measurable backlog, so it sorts after
- // every group with a known backlog instead of first.
- data = [...data].sort((left, right) => {
- const leftKnown = isLagAvailable(left.totalLag);
- const rightKnown = isLagAvailable(right.totalLag);
- if (leftKnown !== rightKnown) return leftKnown ? -1 : 1;
- if (!leftKnown) return 0;
- return right.totalLag - left.totalLag;
- });
- } else if (sortKey === 'name_asc') {
- data = [...data].sort((left, right) =>
left.name.localeCompare(right.name));
- }
-
return data;
};
@@ -233,7 +215,6 @@ const ConsumerPageContent = ({
const [page, setPage] = useState(1);
const [pageSize, setPageSize] = useState(20);
const [modeFilter, setModeFilter] = useState<string>('ALL');
- const [sortKey, setSortKey] = useState<string>('name_asc');
const [modalOpen, setModalOpen] = useState(false);
const [selectedGroup, setSelectedGroup] = useState<ConsumerGroup |
null>(null);
const [settingsGroup, setSettingsGroup] = useState<ConsumerGroup |
null>(null);
@@ -395,8 +376,8 @@ const ConsumerPageContent = ({
/* ─── Filtered & sorted data ─── */
const filtered = useMemo(() => {
- return visibleConsumerGroups(groups, modeFilter, sortKey);
- }, [groups, modeFilter, sortKey]);
+ return visibleConsumerGroups(groups, modeFilter);
+ }, [groups, modeFilter]);
const handleExport = async () => {
setExporting(true);
@@ -405,7 +386,7 @@ const ConsumerPageContent = ({
instanceId: selectedInstanceId || undefined,
search: search.trim() || undefined,
});
- const exportGroups = visibleConsumerGroups(allGroups, modeFilter,
sortKey);
+ const exportGroups = visibleConsumerGroups(allGroups, modeFilter);
downloadCsv(
`rocketmq-consumer-groups-${new Date().toISOString().slice(0,
10)}.csv`,
buildConsumerGroupCsv(exportGroups),
@@ -1099,15 +1080,6 @@ const ConsumerPageContent = ({
{ value: 'Pop', label: 'Pop' },
]}
/>
- <Select
- value={sortKey}
- onChange={setSortKey}
- style={{ width: 160 }}
- options={[
- { value: 'lag_desc', label: '堆积量降序' },
- { value: 'name_asc', label: '名称升序' },
- ]}
- />
</Space>
<Space>
{selectedRowKeys.length > 0 && (
diff --git a/web/src/pages/instance/message.tsx
b/web/src/pages/instance/message.tsx
index 7358085e2..4a089f76d 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -57,6 +57,7 @@ import {
QueueBrowserResults,
} from '../../components/QueueBrowser';
import type { MessageQueryHistory, TraceQueryHistory } from
'../../api/messageHistory';
+import { getMessageQueryResults } from '../../api/messageHistory';
import { useLang } from '../../i18n/LangContext';
import type { MessageQuery, MessageRecord, TraceRecord } from
'../../api/message';
import {
@@ -365,16 +366,9 @@ const MessagePageContent = ({
await executeQuery(queryMode, currentQueryParams);
};
- const replayHistoryRecord = (record: MessageQueryHistory) => {
+ const replayHistoryRecord = async (record: MessageQueryHistory) => {
const modeMap: Record<string, QueryMode> = { TOPIC: 'topic', KEY: 'key',
MSG_ID: 'msgid' };
const mode = modeMap[record.queryType] || 'topic';
- const params: MessageQuery = {
- topic: record.topic,
- msgId: record.msgId || undefined,
- key: record.messageKey || undefined,
- startTime: record.startTime,
- endTime: record.endTime,
- };
setQueryMode(mode);
setSelectedTopic(record.topic);
setKeyInput(record.messageKey || '');
@@ -383,7 +377,35 @@ const MessagePageContent = ({
setDateRange([dayjs(record.startTime), dayjs(record.endTime)]);
}
setHistoryDrawerOpen(false);
- void executeQuery(mode, params);
+ setQueryLoading(true);
+ setQueryError(null);
+ try {
+ const results = await getMessageQueryResults(record.id);
+ const mapped: MessageRecord[] = results.map((r) => ({
+ msgId: r.msgId,
+ topic: r.topic,
+ tag: r.tag || null,
+ key: r.key || null,
+ brokerName: r.brokerName || null,
+ queueId: r.queueId,
+ queueOffset: r.queueOffset,
+ body: '',
+ storeTime: r.storeTime,
+ bornHost: r.bornHost,
+ storeHost: r.storeHost,
+ properties: {},
+ size: r.size,
+ }));
+ setMessages(mapped);
+ setMessageTotal(mapped.length);
+ setMessagePage(1);
+ setResultMayBeTruncated(false);
+ message.success(`已加载历史查询结果，共 ${mapped.length} 条`);
+ } catch (error) {
+ setQueryError(getErrorMessage(error, '加载历史结果失败'));
+ } finally {
+ setQueryLoading(false);
+ }
};
const replayTraceRecord = (record: TraceQueryHistory) => {
diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx
index d380015dc..58d4961fc 100644
--- a/web/src/pages/instance/topic.tsx
+++ b/web/src/pages/instance/topic.tsx
@@ -44,8 +44,6 @@ import {
import type { TableColumnsType } from 'antd';
import {
PlusOutlined,
- UnorderedListOutlined,
- AppstoreOutlined,
SendOutlined,
DeleteOutlined,
EyeOutlined,
@@ -316,7 +314,6 @@ const TopicPage = () => {
const [typeFilter, setTypeFilter] = useState('');
const [tablePage, setTablePage] = useState(1);
const [tablePageSize, setTablePageSize] = useState(20);
- const [viewMode, setViewMode] = useState<string>('列表');
const [detailModalOpen, setDetailModalOpen] = useState(false);
const [detailLoading, setDetailLoading] = useState(false);
const [rebuilding, setRebuilding] = useState(false);
@@ -787,76 +784,6 @@ const TopicPage = () => {
);
};
- // ─── Card view ────────────────────────────────────────────────
- const renderCardView = () => (
- <Row gutter={[16, 16]}>
- {filteredTopics.map((topic) => {
- const typeInfo = TOPIC_TYPE_MAP[topic.type];
- const cluster = CLUSTER_NAME_MAP[topic.clusterId];
- const clusterType = cluster ? CLUSTER_TYPE_MAP[cluster.type] : null;
-
- return (
- <Col xs={24} sm={12} lg={8} key={topic.name}>
- <Card
- hoverable
- size="small"
- onClick={() => void openDetail(topic)}
- styles={{ body: { padding: '16px 20px' } }}
- style={{ borderRadius: 8, border: '1px solid #f0f0f0' }}
- >
- {/* Header row: name + type badge */}
- <Flex justify="space-between" align="flex-start" style={{
marginBottom: 12 }}>
- <Text strong style={{ fontSize: 15 }}>
- {topic.name}
- </Text>
- <Tag color={typeInfo?.color}>
- {typeInfo?.labelKey ? t(typeInfo.labelKey) : topic.type}
- </Tag>
- </Flex>
-
- {/* Cluster tags */}
- <Space size={4} style={{ marginBottom: 16 }}>
- {clusterType && (
- <Tag color={clusterType.color} style={{ fontSize: 14 }}>
- {t(clusterType.labelKey)}
- </Tag>
- )}
- </Space>
-
- {/* Key stats */}
- <Row gutter={16}>
- <Col span={8}>
- <Text type="secondary" style={{ fontSize: 14, display:
'block' }}>
- 今日消息量
- </Text>
- <Text strong style={{ fontSize: 16, fontVariantNumeric:
'tabular-nums' }}>
- {formatNumber(topic.messageCount)}
- </Text>
- </Col>
- <Col span={8}>
- <Text type="secondary" style={{ fontSize: 14, display:
'block' }}>
- TPS
- </Text>
- <Text strong style={{ fontSize: 16, fontVariantNumeric:
'tabular-nums' }}>
- {formatNumber(topic.tps)}
- </Text>
- </Col>
- <Col span={8}>
- <Text type="secondary" style={{ fontSize: 14, display:
'block' }}>
- 消费者组
- </Text>
- <Text strong style={{ fontSize: 16, fontVariantNumeric:
'tabular-nums' }}>
- {topic.consumerGroupCount}
- </Text>
- </Col>
- </Row>
- </Card>
- </Col>
- );
- })}
- </Row>
- );
-
// ─── Create modal submit ──────────────────────────────────────
const handleCreate = async () => {
if (createInFlightRef.current) return;
@@ -1112,14 +1039,6 @@ const TopicPage = () => {
options={TYPE_OPTIONS}
style={{ width: 140 }}
/>
- <Segmented
- value={viewMode}
- onChange={(v) => setViewMode(v as string)}
- options={[
- { label: '列表', value: '列表', icon: <UnorderedListOutlined /> },
- { label: '卡片', value: '卡片', icon: <AppstoreOutlined /> },
- ]}
- />
</Space>
<Space>
{selectedRowKeys.length > 0 && (
@@ -1209,39 +1128,35 @@ const TopicPage = () => {
</Flex>
{/* ── Content ───────────────────────────────────────────── */}
- {viewMode === '列表' ? (
- <Card styles={{ body: { padding: 0 } }} style={{ borderRadius: 8 }}>
- <Table<Topic>
- columns={columns}
- dataSource={filteredTopics}
- loading={loading}
- rowKey="name"
- rowSelection={{
- selectedRowKeys,
- onChange: (keys) => setSelectedRowKeys(keys),
- }}
- pagination={{
- current: currentTablePage,
- pageSize: tablePageSize,
- total: totalTopics,
- showSizeChanger: true,
- showTotal: (t) => `共 ${t} 条`,
- onChange: (page, pageSize) => {
- setTablePage(page);
- setTablePageSize(pageSize);
- },
- }}
- size="small"
- scroll={{ x: tableScrollX(columns, { selection: true }) }}
- onRow={(record) => ({
- onClick: () => void openDetail(record),
- style: { cursor: 'pointer' },
- })}
- />
- </Card>
- ) : (
- renderCardView()
- )}
+ <Card styles={{ body: { padding: 0 } }} style={{ borderRadius: 8 }}>
+ <Table<Topic>
+ columns={columns}
+ dataSource={filteredTopics}
+ loading={loading}
+ rowKey="name"
+ rowSelection={{
+ selectedRowKeys,
+ onChange: (keys) => setSelectedRowKeys(keys),
+ }}
+ pagination={{
+ current: currentTablePage,
+ pageSize: tablePageSize,
+ total: totalTopics,
+ showSizeChanger: true,
+ showTotal: (t) => `共 ${t} 条`,
+ onChange: (page, pageSize) => {
+ setTablePage(page);
+ setTablePageSize(pageSize);
+ },
+ }}
+ size="small"
+ scroll={{ x: tableScrollX(columns, { selection: true }) }}
+ onRow={(record) => ({
+ onClick: () => void openDetail(record),
+ style: { cursor: 'pointer' },
+ })}
+ />
+ </Card>
{/* ── Detail Modal ──────────────────────────────────────── */}
<Modal
