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 9994f0305 feat(message): add queue-based message browsing (#2545)
9994f0305 is described below
commit 9994f030593123609bba08656d07f61ec1342edf
Author: lizhimins <[email protected]>
AuthorDate: Mon Aug 24 15:00:36 2026 +0800
feat(message): add queue-based message browsing (#2545)
Add a new 'Browse by Queue' tab to the message query page, allowing
users to inspect messages by selecting a specific broker queue and
pulling from a given offset. Backend exposes queue offset listing and
offset-based message pull endpoints. Frontend introduces QueueBrowser
component with queue table and per-queue message pull panel.
Also fixes the scan-limit warning incorrectly showing in queue mode.
---
.../studio/instance/message/MessageController.java | 15 +
.../studio/instance/message/MessageProvider.java | 4 +
.../instance/message/MessageProviderStub.java | 13 +
.../studio/instance/message/MessageRecordVO.java | 3 +
.../studio/instance/message/MessageService.java | 22 +-
.../{MessageProvider.java => QueueOffsetVO.java} | 20 +-
web/src/api/message.ts | 29 ++
web/src/components/MessageQueryHistoryDrawer.tsx | 26 +-
web/src/components/QueueBrowser.tsx | 364 +++++++++++++++++
.../pages/instance/__tests__/MessagePage.test.tsx | 3 +
.../__tests__/MessagePageAsyncState.test.tsx | 3 +
web/src/pages/instance/message.tsx | 451 ++++++++-------------
12 files changed, 664 insertions(+), 289 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageController.java
index 586944ff8..314692cc8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageController.java
@@ -62,4 +62,19 @@ public class MessageController {
@RequestParam(required =
false) String topic) {
return Result.ok(messageService.getMessageTrace(instanceId, msgId,
topic));
}
+
+ @GetMapping("/queues")
+ public Result<List<QueueOffsetVO>> getQueueOffsets(@RequestParam String
instanceId,
+ @RequestParam String
topic) {
+ return Result.ok(messageService.getQueueOffsets(instanceId, topic));
+ }
+
+ @GetMapping("/queue-message")
+ public Result<MessageRecordVO> pullMessageAtOffset(@RequestParam String
instanceId,
+ @RequestParam String
topic,
+ @RequestParam String
brokerName,
+ @RequestParam int
queueId,
+ @RequestParam long
offset) {
+ return Result.ok(messageService.pullMessageAtOffset(instanceId, topic,
brokerName, queueId, offset));
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
index c4cc365a3..09242f9fd 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
@@ -24,4 +24,8 @@ public interface MessageProvider {
Long endTime);
TraceRecordVO getMessageTrace(String instanceId, String msgId, String
topic);
+
+ List<QueueOffsetVO> getQueueOffsets(String instanceId, String topic);
+
+ MessageRecordVO pullMessageAtOffset(String instanceId, String topic,
String brokerName, int queueId, long offset);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProviderStub.java
index 7b24291ba..a29a98ef7 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProviderStub.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProviderStub.java
@@ -41,6 +41,19 @@ public class MessageProviderStub implements MessageProvider {
throw unsupported();
}
+ @Override
+ public List<QueueOffsetVO> getQueueOffsets(String instanceId, String
topic) {
+ log.warn("MessageProviderStub.getQueueOffsets called but no real
message provider is configured");
+ throw unsupported();
+ }
+
+ @Override
+ public MessageRecordVO pullMessageAtOffset(String instanceId, String
topic, String brokerName,
+ int queueId, long offset) {
+ log.warn("MessageProviderStub.pullMessageAtOffset called but no real
message provider is configured");
+ throw unsupported();
+ }
+
private BusinessException unsupported() {
return new BusinessException(501, "Message query provider is not
configured");
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageRecordVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageRecordVO.java
index 6908f916e..2dc53898e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageRecordVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageRecordVO.java
@@ -32,6 +32,9 @@ public class MessageRecordVO {
private String topic;
private String tag;
private String key;
+ private String brokerName;
+ private Integer queueId;
+ private Long queueOffset;
private String body;
private String bodyEncoding;
private boolean bodyTruncated;
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 fbeab1ab5..e707d7dd6 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
@@ -31,7 +31,7 @@ import java.util.List;
public class MessageService {
private static final long MAX_TOPIC_QUERY_WINDOW_MILLIS = 7L * 24 * 60 *
60 * 1000;
- private static final int MAX_PAGE_SIZE = 100;
+ private static final int MAX_PAGE_SIZE = 200;
private static final int TOPIC_QUERY_RESULT_LIMIT = 200;
private final MessageProvider messageProvider;
@@ -52,7 +52,7 @@ public class MessageService {
public MessageQueryPageVO queryMessagesPage(String instanceId, String
topic, String msgId, String tag,
String key, Long startTime,
Long endTime, int page, int pageSize) {
if (page < 1 || pageSize < 1 || pageSize > MAX_PAGE_SIZE) {
- throw new BusinessException(400, "page must be positive and
pageSize must be between 1 and 100");
+ throw new BusinessException(400, "page must be positive and
pageSize must be between 1 and 200");
}
List<MessageRecordVO> result = queryMessages(instanceId, topic, msgId,
tag, key, startTime, endTime);
long offset = (long) (page - 1) * pageSize;
@@ -76,6 +76,24 @@ public class MessageService {
return result;
}
+ public List<QueueOffsetVO> getQueueOffsets(String instanceId, String
topic) {
+ if (!StringUtils.hasText(topic)) {
+ throw new BusinessException(400, "topic is required");
+ }
+ return messageProvider.getQueueOffsets(instanceId, topic);
+ }
+
+ public MessageRecordVO pullMessageAtOffset(String instanceId, String
topic, String brokerName,
+ int queueId, long offset) {
+ if (!StringUtils.hasText(topic)) {
+ throw new BusinessException(400, "topic is required");
+ }
+ if (!StringUtils.hasText(brokerName)) {
+ throw new BusinessException(400, "brokerName is required");
+ }
+ return messageProvider.pullMessageAtOffset(instanceId, topic,
brokerName, queueId, offset);
+ }
+
private void recordMessageQuery(String instanceId, String topic, String
msgId, String tag,
String key, Long startTime, Long endTime,
int resultCount) {
String queryType = StringUtils.hasText(msgId) ? "MSG_ID" :
StringUtils.hasText(key) ? "KEY" : "TOPIC";
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueueOffsetVO.java
similarity index 72%
copy from
server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
copy to
server/src/main/java/org/apache/rocketmq/studio/instance/message/QueueOffsetVO.java
index c4cc365a3..c470b6236 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueueOffsetVO.java
@@ -16,12 +16,18 @@
*/
package org.apache.rocketmq.studio.instance.message;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
-import java.util.List;
-
-public interface MessageProvider {
- List<MessageRecordVO> queryMessages(String instanceId, String topic,
String msgId, String tag, String key, Long startTime,
- Long endTime);
-
- TraceRecordVO getMessageTrace(String instanceId, String msgId, String
topic);
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class QueueOffsetVO {
+ private String brokerName;
+ private int queueId;
+ private long minOffset;
+ private long maxOffset;
}
diff --git a/web/src/api/message.ts b/web/src/api/message.ts
index 9762e886d..27fa8b10e 100644
--- a/web/src/api/message.ts
+++ b/web/src/api/message.ts
@@ -6,6 +6,9 @@ export interface MessageRecord {
topic: string;
tag: string | null;
key: string | null;
+ brokerName: string | null;
+ queueId: number | null;
+ queueOffset: number | null;
body: string;
storeTime: number | string;
bornHost: string;
@@ -142,3 +145,29 @@ export async function exportDLQMessages(params: {
const res = await client.get<Blob>('/dlq/export', { params, responseType:
'blob' });
return res.data;
}
+
+// ─── Queue Browser ─────────────────────────────────────────────────
+export interface QueueOffset {
+ brokerName: string;
+ queueId: number;
+ minOffset: number;
+ maxOffset: number;
+}
+
+export async function getQueueOffsets(params: { instanceId: string; topic:
string }) {
+ const res = await client.get<{ data: QueueOffset[] }>('/messages/queues', {
params });
+ return res.data.data;
+}
+
+export async function pullMessageAtOffset(params: {
+ instanceId: string;
+ topic: string;
+ brokerName: string;
+ queueId: number;
+ offset: number;
+}) {
+ const res = await client.get<{ data: MessageRecord | null
}>('/messages/queue-message', {
+ params,
+ });
+ return res.data.data;
+}
diff --git a/web/src/components/MessageQueryHistoryDrawer.tsx
b/web/src/components/MessageQueryHistoryDrawer.tsx
index 846f243de..6914b28ce 100644
--- a/web/src/components/MessageQueryHistoryDrawer.tsx
+++ b/web/src/components/MessageQueryHistoryDrawer.tsx
@@ -20,6 +20,8 @@ interface Props {
open: boolean;
clusterId?: string;
onClose: () => void;
+ onSelectMessage?: (record: MessageQueryHistory) => void;
+ onSelectTrace?: (record: TraceQueryHistory) => void;
}
const PAGE_SIZE = 20;
@@ -29,7 +31,13 @@ const formatTime = (value?: string) => {
return Number.isNaN(timestamp.getTime()) ? '-' : timestamp.toLocaleString();
};
-const MessageQueryHistoryDrawer = ({ open, clusterId, onClose }: Props) => {
+const MessageQueryHistoryDrawer = ({
+ open,
+ clusterId,
+ onClose,
+ onSelectMessage,
+ onSelectTrace,
+}: Props) => {
const [tab, setTab] = useState<'messages' | 'traces'>('messages');
const [search, setSearch] = useState('');
const [page, setPage] = useState(1);
@@ -154,6 +162,14 @@ const MessageQueryHistoryDrawer = ({ open, clusterId,
onClose }: Props) => {
columns={messageColumns}
dataSource={messageRows}
pagination={{ current: page, pageSize: PAGE_SIZE, total,
onChange: setPage }}
+ onRow={
+ onSelectMessage
+ ? (record) => ({
+ onClick: () => onSelectMessage(record),
+ style: { cursor: 'pointer' },
+ })
+ : undefined
+ }
/>
),
},
@@ -167,6 +183,14 @@ const MessageQueryHistoryDrawer = ({ open, clusterId,
onClose }: Props) => {
columns={traceColumns}
dataSource={traceRows}
pagination={{ current: page, pageSize: PAGE_SIZE, total,
onChange: setPage }}
+ onRow={
+ onSelectTrace
+ ? (record) => ({
+ onClick: () => onSelectTrace(record),
+ style: { cursor: 'pointer' },
+ })
+ : undefined
+ }
/>
),
},
diff --git a/web/src/components/QueueBrowser.tsx
b/web/src/components/QueueBrowser.tsx
new file mode 100644
index 000000000..367817670
--- /dev/null
+++ b/web/src/components/QueueBrowser.tsx
@@ -0,0 +1,364 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { useCallback, useState } from 'react';
+import {
+ Button,
+ Card,
+ Descriptions,
+ Empty,
+ Flex,
+ Select,
+ Slider,
+ Space,
+ Spin,
+ Table,
+ Tag,
+ Tooltip,
+ Typography,
+ message,
+} from 'antd';
+import { CloseOutlined, SearchOutlined } from '@ant-design/icons';
+import type { MessageRecord, QueueOffset } from '../api/message';
+import { getQueueOffsets, pullMessageAtOffset } from '../api/message';
+
+const { Text, Paragraph } = Typography;
+
+export interface TopicOption {
+ label: string;
+ value: string;
+}
+
+const formatTimeMs = (value: number | string) => {
+ const ts = typeof value === 'string' ? new Date(value).getTime() : value;
+ if (!ts || Number.isNaN(ts)) return '-';
+ return new Date(ts).toLocaleString('zh-CN', { hour12: false });
+};
+
+export interface PulledEntry {
+ key: string;
+ offset: number;
+ message: MessageRecord | null;
+}
+
+export const useQueueBrowser = (instanceId?: string) => {
+ const [topic, setTopic] = useState<string | undefined>();
+ const [queues, setQueues] = useState<QueueOffset[]>([]);
+ const [loading, setLoading] = useState(false);
+ const [offsets, setOffsets] = useState<Record<string, number>>({});
+ const [pulling, setPulling] = useState<string | null>(null);
+ const [entries, setEntries] = useState<PulledEntry[]>([]);
+
+ const loadQueues = useCallback(async () => {
+ if (!instanceId || !topic) return;
+ setLoading(true);
+ setQueues([]);
+ setOffsets({});
+ setEntries([]);
+ try {
+ const result = await getQueueOffsets({ instanceId, topic });
+ setQueues(result);
+ const initial: Record<string, number> = {};
+ for (const q of result) {
+ initial[`${q.brokerName}-${q.queueId}`] =
+ q.maxOffset > q.minOffset ? q.maxOffset - 1 : q.minOffset;
+ }
+ setOffsets(initial);
+ } catch (err) {
+ message.error(err instanceof Error ? err.message : '加载队列信息失败');
+ } finally {
+ setLoading(false);
+ }
+ }, [instanceId, topic]);
+
+ const handlePull = async (queue: QueueOffset) => {
+ if (!instanceId || !topic) return;
+ const key = `${queue.brokerName}-${queue.queueId}`;
+ const offset = offsets[key] ?? queue.minOffset;
+ setPulling(key);
+ try {
+ const msg = await pullMessageAtOffset({
+ instanceId,
+ topic,
+ brokerName: queue.brokerName,
+ queueId: queue.queueId,
+ offset,
+ });
+ setEntries((prev) => [
+ ...prev.filter((entry) => entry.key !== key),
+ { key, offset, message: msg },
+ ]);
+ } catch (err) {
+ message.error(err instanceof Error ? err.message : '拉取消息失败');
+ } finally {
+ setPulling(null);
+ }
+ };
+
+ const closeEntry = (key: string) => {
+ setEntries((prev) => prev.filter((entry) => entry.key !== key));
+ };
+
+ return {
+ topic,
+ setTopic,
+ queues,
+ loading,
+ offsets,
+ setOffsets,
+ pulling,
+ entries,
+ loadQueues,
+ handlePull,
+ closeEntry,
+ };
+};
+
+export type QueueBrowserState = ReturnType<typeof useQueueBrowser>;
+
+interface ControlsProps {
+ instanceId?: string;
+ state: QueueBrowserState;
+ topicOptions: TopicOption[];
+ topicLoading?: boolean;
+}
+
+export const QueueBrowserControls = ({
+ instanceId,
+ state,
+ topicOptions,
+ topicLoading,
+}: ControlsProps) => (
+ <Flex gap={12} align="center">
+ <Select
+ showSearch
+ allowClear
+ placeholder="选择 Topic"
+ value={state.topic}
+ onChange={state.setTopic}
+ options={topicOptions}
+ loading={topicLoading}
+ style={{ width: 280 }}
+ />
+ <Button
+ type="primary"
+ icon={<SearchOutlined />}
+ disabled={!instanceId || !state.topic}
+ loading={state.loading}
+ onClick={() => void state.loadQueues()}
+ >
+ 加载队列
+ </Button>
+ </Flex>
+);
+
+export const QueueBrowserResults = ({ state }: { state: QueueBrowserState })
=> (
+ <Card>
+ {state.loading ? (
+ <Flex justify="center" style={{ padding: 32 }}>
+ <Spin />
+ </Flex>
+ ) : state.queues.length === 0 ? (
+ <Empty
+ image={Empty.PRESENTED_IMAGE_SIMPLE}
+ description="选择 Topic 并点击「加载队列」，按队列浏览消息"
+ style={{ padding: '32px 0' }}
+ />
+ ) : (
+ <Flex gap={16} align="flex-start">
+ {/* 左侧：队列表格 */}
+ <div style={{ width: '50%', flexShrink: 0 }}>
+ <Table<QueueOffset>
+ rowKey={(r) => `${r.brokerName}-${r.queueId}`}
+ dataSource={state.queues}
+ size="small"
+ pagination={false}
+ columns={[
+ {
+ title: 'Broker',
+ dataIndex: 'brokerName',
+ width: 180,
+ ellipsis: { showTitle: false },
+ render: (v: string) => (
+ <Tooltip title={v}>
+ <Text strong style={{ fontSize: 14 }}>
+ {v}
+ </Text>
+ </Tooltip>
+ ),
+ },
+ {
+ title: 'Queue',
+ dataIndex: 'queueId',
+ width: 50,
+ align: 'center',
+ render: (v: number) => <Text style={{ fontSize: 14
}}>{v}</Text>,
+ },
+ {
+ title: 'Offset 范围',
+ key: 'offset',
+ width: 170,
+ render: (_: unknown, record: QueueOffset) => {
+ const key = `${record.brokerName}-${record.queueId}`;
+ const currentOffset = state.offsets[key] ?? record.minOffset;
+ return (
+ <Flex align="center" gap={8}>
+ <Text type="secondary" style={{ fontSize: 14,
flexShrink: 0 }}>
+ {record.minOffset}
+ </Text>
+ <Slider
+ style={{ flex: 1, margin: 0 }}
+ min={record.minOffset}
+ max={
+ record.maxOffset > record.minOffset
+ ? record.maxOffset - 1
+ : record.minOffset
+ }
+ value={currentOffset}
+ onChange={(value) =>
+ state.setOffsets((prev) => ({ ...prev, [key]: value
}))
+ }
+ tooltip={{ formatter: (v) => `offset: ${v}` }}
+ />
+ <Text code style={{ fontSize: 14, flexShrink: 0 }}>
+ {currentOffset}
+ </Text>
+ </Flex>
+ );
+ },
+ },
+ {
+ title: '操作',
+ key: 'action',
+ width: 70,
+ align: 'center',
+ render: (_: unknown, record: QueueOffset) => {
+ const key = `${record.brokerName}-${record.queueId}`;
+ return (
+ <Button
+ size="small"
+ type="primary"
+ loading={state.pulling === key}
+ onClick={() => void state.handlePull(record)}
+ >
+ 查看
+ </Button>
+ );
+ },
+ },
+ ]}
+ />
+ <Text type="secondary" style={{ display: 'block', marginTop: 8,
fontSize: 14 }}>
+ 共 {state.queues.length} 个队列，总消息量{' '}
+ {state.queues.reduce((sum, q) => sum + (q.maxOffset -
q.minOffset), 0)} 条
+ </Text>
+ </div>
+
+ {/* 右侧：消息详情（2 列，可多条并存） */}
+ <div style={{ flex: 1, minWidth: 0 }}>
+ {state.entries.length === 0 ? (
+ <Empty
+ image={Empty.PRESENTED_IMAGE_SIMPLE}
+ description="点击左侧「查看」，消息详情将显示在这里"
+ style={{ padding: '32px 0' }}
+ />
+ ) : (
+ <div style={{ display: 'grid', gridTemplateColumns: '1fr 1fr',
gap: 12 }}>
+ {state.entries.map((entry) => (
+ <Card
+ key={entry.key}
+ size="small"
+ style={{ borderRadius: 8 }}
+ title={
+ <Space size={4}>
+ <Tag color="blue" style={{ marginInlineEnd: 0 }}>
+ {entry.key}
+ </Tag>
+ <Text type="secondary" style={{ fontSize: 14 }}>
+ @ {entry.offset}
+ </Text>
+ </Space>
+ }
+ extra={
+ <Button
+ type="text"
+ size="small"
+ icon={<CloseOutlined />}
+ onClick={() => state.closeEntry(entry.key)}
+ />
+ }
+ >
+ {entry.message ? (
+ <>
+ <Descriptions column={1} size="small">
+ <Descriptions.Item label="Message ID">
+ <Paragraph
+ copyable
+ style={{ marginBottom: 0, fontFamily: 'monospace',
fontSize: 14 }}
+ >
+ {entry.message.msgId}
+ </Paragraph>
+ </Descriptions.Item>
+ <Descriptions.Item label="Tag">
+ <Tag>{entry.message.tag || '-'}</Tag>
+ </Descriptions.Item>
+ <Descriptions.Item label="Key">
+ <span style={{ fontFamily: 'monospace', fontSize: 14
}}>
+ {entry.message.key || '-'}
+ </span>
+ </Descriptions.Item>
+ <Descriptions.Item label="存储时间">
+ <span style={{ fontFamily: 'monospace', fontSize: 14
}}>
+ {formatTimeMs(entry.message.storeTime)}
+ </span>
+ </Descriptions.Item>
+ <Descriptions.Item label="大小">
+ {entry.message.size} bytes
+ </Descriptions.Item>
+ <Descriptions.Item label="Born Host">
+ {entry.message.bornHost || '-'}
+ </Descriptions.Item>
+ </Descriptions>
+ {entry.message.body && (
+ <Card size="small" title="Body" style={{ marginTop: 12
}}>
+ <pre
+ style={{
+ maxHeight: 160,
+ overflow: 'auto',
+ fontSize: 14,
+ fontFamily: 'monospace',
+ whiteSpace: 'pre-wrap',
+ wordBreak: 'break-all',
+ margin: 0,
+ }}
+ >
+ {entry.message.body}
+ </pre>
+ </Card>
+ )}
+ </>
+ ) : (
+ <Text type="secondary">该 offset 处无消息</Text>
+ )}
+ </Card>
+ ))}
+ </div>
+ )}
+ </div>
+ </Flex>
+ )}
+ </Card>
+);
diff --git a/web/src/pages/instance/__tests__/MessagePage.test.tsx
b/web/src/pages/instance/__tests__/MessagePage.test.tsx
index f12750ddc..3a0e259eb 100644
--- a/web/src/pages/instance/__tests__/MessagePage.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePage.test.tsx
@@ -62,6 +62,9 @@ const createMessage = (msgId: string): MessageRecord => ({
topic: `topic-${msgId}`,
tag: 'tag',
key: `key-${msgId}`,
+ brokerName: 'broker-a',
+ queueId: 0,
+ queueOffset: 0,
body: '{}',
storeTime: '2026-07-31T00:00:00Z',
bornHost: '127.0.0.1:1000',
diff --git a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
index 3ee9c2bfa..8f7ae5192 100644
--- a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
@@ -83,6 +83,9 @@ const createMessage = (msgId: string): MessageRecord => ({
topic: `topic-${msgId}`,
tag: 'tag',
key: `key-${msgId}`,
+ brokerName: 'broker-a',
+ queueId: 0,
+ queueOffset: 0,
body: '{}',
storeTime: '2026-07-31T00:00:00Z',
bornHost: '127.0.0.1:1000',
diff --git a/web/src/pages/instance/message.tsx
b/web/src/pages/instance/message.tsx
index 9d91efc31..d74cb2292 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -33,10 +33,8 @@ import {
Input,
Space,
Flex,
- Dropdown,
message,
} from 'antd';
-import type { MenuProps } from 'antd';
import {
SearchOutlined,
ReloadOutlined,
@@ -46,7 +44,6 @@ import {
CheckCircleOutlined,
DownloadOutlined,
HistoryOutlined,
- DeleteOutlined,
} from '@ant-design/icons';
import type { ColumnsType } from 'antd/es/table';
import dayjs from 'dayjs';
@@ -54,6 +51,12 @@ import type { Dayjs } from 'dayjs';
import PageHeader from '../../components/PageHeader';
import { InstanceSelect } from '../../components/InstanceSelect';
import MessageQueryHistoryDrawer from
'../../components/MessageQueryHistoryDrawer';
+import {
+ useQueueBrowser,
+ QueueBrowserControls,
+ QueueBrowserResults,
+} from '../../components/QueueBrowser';
+import type { MessageQueryHistory, TraceQueryHistory } from
'../../api/messageHistory';
import { useLang } from '../../i18n/LangContext';
import type { MessageQuery, MessageRecord, TraceRecord } from
'../../api/message';
import { getMessageTrace, queryMessagePage } from
'../../services/messageService';
@@ -69,12 +72,7 @@ const DEFAULT_TRACE_ERROR = '消息轨迹加载失败，请稍后重试';
/* ─── Constants ─── */
-type QueryMode = 'topic' | 'key' | 'msgid';
-
-type RecentQuery = {
- mode: QueryMode;
- params: MessageQuery;
-};
+type QueryMode = 'topic' | 'key' | 'msgid' | 'queue';
type ApiErrorLike = {
message?: unknown;
@@ -85,14 +83,13 @@ type ApiErrorLike = {
};
};
-const QUERY_HISTORY_STORAGE_KEY = 'rocketmq-studio-message-query-history';
-const MAX_QUERY_HISTORY = 5;
const RESEND_UNAVAILABLE_MESSAGE = '当前版本尚未接入普通消息重新发送接口';
const QUERY_OPTIONS = [
{ value: 'topic' as const, label: '按 Topic 查询' },
{ value: 'key' as const, label: '按 Message Key' },
{ value: 'msgid' as const, label: '按 Message ID' },
+ { value: 'queue' as const, label: '按队列浏览' },
];
const DELIVERY_STATUS_MAP: Record<string, { label: string; color: string }> = {
@@ -135,28 +132,6 @@ const formatBody = (body: string): string => {
}
};
-const isQueryMode = (value: unknown): value is QueryMode =>
- value === 'topic' || value === 'key' || value === 'msgid';
-
-const isOptionalString = (value: unknown): value is string | undefined =>
- value === undefined || typeof value === 'string';
-
-const isOptionalTimestamp = (value: unknown): value is number | undefined =>
- value === undefined || (typeof value === 'number' && Number.isFinite(value));
-
-const isMessageQuery = (value: unknown): value is MessageQuery => {
- if (typeof value !== 'object' || value === null || Array.isArray(value))
return false;
- const params = value as MessageQuery;
- return (
- isOptionalString(params.topic) &&
- isOptionalString(params.tag) &&
- isOptionalString(params.key) &&
- isOptionalString(params.msgId) &&
- isOptionalTimestamp(params.startTime) &&
- isOptionalTimestamp(params.endTime)
- );
-};
-
const getQueryValidationError = (mode: QueryMode, params: MessageQuery):
string | null => {
if (!params.topic?.trim()) return '请选择 Topic';
if (mode === 'key' && !params.key?.trim()) return '请输入 Message Key';
@@ -191,42 +166,6 @@ const normalizeMessageQuery = (mode: QueryMode, params:
MessageQuery): MessageQu
return commonParams;
};
-const isRecentQuery = (value: unknown): value is RecentQuery => {
- if (typeof value !== 'object' || value === null) return false;
- const query = value as RecentQuery;
- return (
- isQueryMode(query.mode) &&
- isMessageQuery(query.params) &&
- getQueryValidationError(query.mode, normalizeMessageQuery(query.mode,
query.params)) === null
- );
-};
-
-const loadRecentQueries = (): RecentQuery[] => {
- try {
- const stored = localStorage.getItem(QUERY_HISTORY_STORAGE_KEY);
- if (!stored) return [];
- const parsed: unknown = JSON.parse(stored);
- if (!Array.isArray(parsed)) return [];
- return parsed
- .filter(isRecentQuery)
- .map(({ mode, params }) => ({ mode, params: normalizeMessageQuery(mode,
params) }))
- .slice(0, MAX_QUERY_HISTORY);
- } catch {
- return [];
- }
-};
-
-const querySignature = (query: RecentQuery): string => JSON.stringify(query);
-
-const queryLabel = ({ mode, params }: RecentQuery): string => {
- if (mode === 'msgid')
- return `Message ID: ${params.msgId || '全部'} · Topic: ${params.topic ||
'全部'}`;
- if (mode === 'key') {
- return `Key: ${params.key || '全部'}${params.topic ? ` · Topic:
${params.topic}` : ''}`;
- }
- return `Topic: ${params.topic || '全部'}`;
-};
-
const getErrorMessage = (error: unknown, fallback: string): string => {
const apiError = error as ApiErrorLike;
const responseMessage = apiError.response?.data?.message;
@@ -307,6 +246,7 @@ const MessagePageContent = ({
};
}, [loadTopicOptions]);
const [queryMode, setQueryMode] = useState<QueryMode>('topic');
+ const queueBrowser = useQueueBrowser(selectedInstanceId);
const [selectedTopic, setSelectedTopic] = useState<string | undefined>();
const [dateRange, setDateRange] = useState<[Dayjs, Dayjs]>(getDefaultRange);
const [keyInput, setKeyInput] = useState('');
@@ -324,7 +264,6 @@ const MessagePageContent = ({
const [traceLoading, setTraceLoading] = useState(false);
const [queryError, setQueryError] = useState<string | null>(null);
const [traceError, setTraceError] = useState<string | null>(null);
- const [recentQueries, setRecentQueries] =
useState<RecentQuery[]>(loadRecentQueries);
const [historyDrawerOpen, setHistoryDrawerOpen] = useState(false);
const queryGenerationRef = useRef(0);
const traceGenerationRef = useRef(0);
@@ -364,29 +303,11 @@ const MessagePageContent = ({
setQueryLoading(false);
};
- const saveRecentQuery = (mode: QueryMode, params: MessageQuery) => {
- const nextQuery = { mode, params };
- const signature = querySignature(nextQuery);
- setRecentQueries((current) => {
- const next = [
- nextQuery,
- ...current.filter((item) => querySignature(item) !== signature),
- ].slice(0, MAX_QUERY_HISTORY);
- try {
- localStorage.setItem(QUERY_HISTORY_STORAGE_KEY, JSON.stringify(next));
- } catch {
- // Query history remains available for the current session when
storage is unavailable.
- }
- return next;
- });
- };
-
const executeQuery = async (
mode: QueryMode,
params: MessageQuery,
page = 1,
pageSize = messagePageSize,
- saveHistory = true,
) => {
const requestGeneration = queryGenerationRef.current + 1;
queryGenerationRef.current = requestGeneration;
@@ -418,10 +339,7 @@ const MessagePageContent = ({
setMessagePageSize(result.size);
setResultMayBeTruncated(result.resultMayBeTruncated);
setQueryError(null);
- if (saveHistory) {
- saveRecentQuery(mode, normalizedParams);
- message.success(`查询完成，共 ${result.total} 条`);
- }
+ message.success(`查询完成，共 ${result.total} 条`);
} catch (error) {
if (queryGenerationRef.current === requestGeneration) {
setQueryError(getErrorMessage(error, DEFAULT_QUERY_ERROR));
@@ -437,66 +355,33 @@ const MessagePageContent = ({
await executeQuery(queryMode, currentQueryParams);
};
- const replayRecentQuery = (recentQuery: RecentQuery) => {
- const { mode, params } = recentQuery;
+ const replayHistoryRecord = (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(params.topic);
- setKeyInput(params.key || '');
- setMsgIdInput(params.msgId || '');
- if (mode === 'topic' && params.startTime !== undefined && params.endTime
!== undefined) {
- setDateRange([dayjs(params.startTime), dayjs(params.endTime)]);
+ setSelectedTopic(record.topic);
+ setKeyInput(record.messageKey || '');
+ setMsgIdInput(record.msgId || '');
+ if (mode === 'topic' && record.startTime !== undefined && record.endTime
!== undefined) {
+ setDateRange([dayjs(record.startTime), dayjs(record.endTime)]);
}
+ setHistoryDrawerOpen(false);
void executeQuery(mode, params);
};
- const clearRecentQueries = () => {
- setRecentQueries([]);
- try {
- localStorage.removeItem(QUERY_HISTORY_STORAGE_KEY);
- } catch {
- // Ignore storage failures after clearing the in-memory history.
- }
- };
-
- const recentQueryMenuItems: MenuProps['items'] = [
- ...recentQueries.map((recentQuery, index) => {
- const label = queryLabel(recentQuery);
- return {
- key: String(index),
- label: (
- <Text ellipsis={{ tooltip: label }} style={{ maxWidth: 360 }}>
- {label}
- </Text>
- ),
- };
- }),
- ...(recentQueries.length > 0
- ? [
- { type: 'divider' as const },
- {
- key: 'clear',
- danger: true,
- icon: <DeleteOutlined />,
- label: '清空历史',
- },
- ]
- : []),
- ];
-
- const handleRecentQueryMenuClick: MenuProps['onClick'] = ({ key }) => {
- if (key === 'clear') {
- Modal.confirm({
- title: '清空查询历史',
- content: '确定要清空全部查询历史吗？此操作不可恢复。',
- okText: '清空',
- okType: 'danger',
- cancelText: '取消',
- onOk: clearRecentQueries,
- });
- return;
- }
- const recentQuery = recentQueries[Number(key)];
- if (recentQuery) replayRecentQuery(recentQuery);
+ const replayTraceRecord = (record: TraceQueryHistory) => {
+ setQueryMode('msgid');
+ setSelectedTopic(record.topic);
+ setMsgIdInput(record.msgId);
+ setHistoryDrawerOpen(false);
+ void executeQuery('msgid', { topic: record.topic, msgId: record.msgId });
};
const handleVerifyConsume = () => {
@@ -547,6 +432,7 @@ const MessagePageContent = ({
dataIndex: 'topic',
key: 'topic',
width: 170,
+ ellipsis: true,
sorter: (a, b) => a.topic.localeCompare(b.topic),
render: (topic: string) => (
<Text strong style={{ fontSize: 14 }}>
@@ -559,7 +445,6 @@ const MessagePageContent = ({
dataIndex: 'tag',
key: 'tag',
width: 80,
- sorter: (a, b) => (a.tag ?? '').localeCompare(b.tag ?? ''),
render: (tag: string | null) => <Tag>{tag || '-'}</Tag>,
},
{
@@ -567,7 +452,7 @@ const MessagePageContent = ({
dataIndex: 'key',
key: 'key',
width: 120,
- sorter: (a, b) => (a.key ?? '').localeCompare(b.key ?? ''),
+ ellipsis: true,
render: (key: string | null) => (
<span style={{ fontFamily: 'monospace', fontSize: 14 }}>{key ||
'-'}</span>
),
@@ -576,11 +461,15 @@ const MessagePageContent = ({
title: 'Message ID',
dataIndex: 'msgId',
key: 'msgId',
- sorter: (a, b) => a.msgId.localeCompare(b.msgId),
+ width: 260,
render: (id: string) => (
- <Paragraph copyable style={{ fontSize: 14, marginBottom: 0,
fontFamily: 'monospace' }}>
+ <Text
+ copyable={{ text: id }}
+ ellipsis={{ tooltip: id }}
+ style={{ fontSize: 14, fontFamily: 'monospace', width: '100%',
display: 'block' }}
+ >
{id}
- </Paragraph>
+ </Text>
),
},
{
@@ -601,7 +490,6 @@ const MessagePageContent = ({
key: 'size',
width: 80,
align: 'right',
- sorter: (a, b) => a.size - b.size,
render: (size: number) => formatSize(size),
},
{
@@ -818,119 +706,120 @@ const MessagePageContent = ({
/>
</Space>
- <Space wrap size={12}>
- {queryMode === 'topic' && (
- <>
- <Select
- placeholder="选择 Topic"
- style={{ width: 360 }}
- value={selectedTopic}
- onChange={setSelectedTopic}
- allowClear
- showSearch
- loading={topicLoading}
- disabled={topicLoading || Boolean(topicError)}
- options={topicOptions.map((t) => ({
- value: t,
- label: t,
- }))}
- />
- <RangePicker
- showTime
- style={{ width: 400 }}
- value={dateRange}
- onChange={(vals) => {
- if (vals && vals[0] && vals[1]) {
- setDateRange([vals[0], vals[1]]);
- }
- }}
- />
- </>
- )}
-
- {queryMode === 'key' && (
- <>
- <Select
- placeholder="选择 Topic"
- style={{ width: 360 }}
- value={selectedTopic}
- onChange={setSelectedTopic}
- allowClear
- showSearch
- loading={topicLoading}
- disabled={topicLoading || Boolean(topicError)}
- options={topicOptions.map((t) => ({
- value: t,
- label: t,
- }))}
- />
- <Input
- placeholder="输入 Message Key"
- style={{ width: 240 }}
- value={keyInput}
- onChange={(e) => setKeyInput(e.target.value)}
- />
- </>
- )}
-
- {queryMode === 'msgid' && (
- <>
- <Select
- placeholder="选择 Topic"
- style={{ width: 360 }}
- value={selectedTopic}
- onChange={setSelectedTopic}
- allowClear
- showSearch
- loading={topicLoading}
- disabled={topicLoading || Boolean(topicError)}
- options={topicOptions.map((t) => ({
- value: t,
- label: t,
- }))}
- />
- <Input
- placeholder="输入 Message ID"
- style={{ width: 400 }}
- value={msgIdInput}
- onChange={(e) => setMsgIdInput(e.target.value)}
- />
- </>
- )}
+ {queryMode !== 'queue' && (
+ <Space wrap size={12}>
+ {queryMode === 'topic' && (
+ <>
+ <Select
+ placeholder="选择 Topic"
+ style={{ width: 360 }}
+ value={selectedTopic}
+ onChange={setSelectedTopic}
+ allowClear
+ showSearch
+ loading={topicLoading}
+ disabled={topicLoading || Boolean(topicError)}
+ options={topicOptions.map((t) => ({
+ value: t,
+ label: t,
+ }))}
+ />
+ <RangePicker
+ showTime
+ style={{ width: 400 }}
+ value={dateRange}
+ onChange={(vals) => {
+ if (vals && vals[0] && vals[1]) {
+ setDateRange([vals[0], vals[1]]);
+ }
+ }}
+ />
+ </>
+ )}
+
+ {queryMode === 'key' && (
+ <>
+ <Select
+ placeholder="选择 Topic"
+ style={{ width: 360 }}
+ value={selectedTopic}
+ onChange={setSelectedTopic}
+ allowClear
+ showSearch
+ loading={topicLoading}
+ disabled={topicLoading || Boolean(topicError)}
+ options={topicOptions.map((t) => ({
+ value: t,
+ label: t,
+ }))}
+ />
+ <Input
+ placeholder="输入 Message Key"
+ style={{ width: 240 }}
+ value={keyInput}
+ onChange={(e) => setKeyInput(e.target.value)}
+ />
+ </>
+ )}
+
+ {queryMode === 'msgid' && (
+ <>
+ <Select
+ placeholder="选择 Topic"
+ style={{ width: 360 }}
+ value={selectedTopic}
+ onChange={setSelectedTopic}
+ allowClear
+ showSearch
+ loading={topicLoading}
+ disabled={topicLoading || Boolean(topicError)}
+ options={topicOptions.map((t) => ({
+ value: t,
+ label: t,
+ }))}
+ />
+ <Input
+ placeholder="输入 Message ID"
+ style={{ width: 400 }}
+ value={msgIdInput}
+ onChange={(e) => setMsgIdInput(e.target.value)}
+ />
+ </>
+ )}
- <Button
- type="primary"
- icon={<SearchOutlined />}
- disabled={Boolean(queryDisabledReason)}
- title={queryDisabledReason || undefined}
- onClick={() => {
- void handleQuery();
- }}
- >
- 查询
- </Button>
- <Dropdown
- menu={{ items: recentQueryMenuItems, onClick:
handleRecentQueryMenuClick }}
- trigger={['click']}
- disabled={recentQueries.length === 0 || !selectedInstanceId}
- >
<Button
- icon={<HistoryOutlined />}
- disabled={recentQueries.length === 0 || !selectedInstanceId}
+ type="primary"
+ icon={<SearchOutlined />}
+ disabled={Boolean(queryDisabledReason)}
+ title={queryDisabledReason || undefined}
+ onClick={() => {
+ void handleQuery();
+ }}
>
- 最近查询
+ 查询
</Button>
- </Dropdown>
- <Button icon={<ReloadOutlined />} onClick={handleReset}>
- 重置
- </Button>
- <Button icon={<HistoryOutlined />} onClick={() =>
setHistoryDrawerOpen(true)}>
- 服务端历史
- </Button>
- </Space>
+ <Button icon={<ReloadOutlined />} onClick={handleReset}>
+ 重置
+ </Button>
+ <Button icon={<HistoryOutlined />} onClick={() =>
setHistoryDrawerOpen(true)}>
+ 服务端历史
+ </Button>
+ </Space>
+ )}
+
+ {queryMode === 'queue' && (
+ <QueueBrowserControls
+ instanceId={selectedInstanceId}
+ state={queueBrowser}
+ topicOptions={topicOptions.map((t) => ({ label: t, value: t }))}
+ topicLoading={topicLoading}
+ />
+ )}
</Space>
</Card>
+ {queryMode === 'queue' && <QueueBrowserResults state={queueBrowser} />}
+
{topicError && (
<Alert
showIcon
@@ -949,12 +838,14 @@ const MessagePageContent = ({
open={historyDrawerOpen}
clusterId={selectedInstanceId}
onClose={() => setHistoryDrawerOpen(false)}
+ onSelectMessage={replayHistoryRecord}
+ onSelectTrace={replayTraceRecord}
/>
{queryError && (
<Alert showIcon type="warning" message={queryError} style={{
marginBottom: 16 }} />
)}
- {resultMayBeTruncated && (
+ {queryMode !== 'queue' && resultMayBeTruncated && (
<Alert
showIcon
type="warning"
@@ -964,25 +855,27 @@ const MessagePageContent = ({
)}
{/* ── Results Table ── */}
- <Card styles={{ body: { padding: 0 } }}>
- <Table
- columns={columns}
- dataSource={messages}
- loading={queryLoading}
- rowKey="msgId"
- pagination={{
- current: messagePage,
- pageSize: messagePageSize,
- total: messageTotal,
- showSizeChanger: true,
- showTotal: (total) => `共 ${total} 条消息`,
- onChange: (page, pageSize) =>
- void executeQuery(queryMode, currentQueryParams, page, pageSize,
false),
- }}
- size="small"
- scroll={{ x: tableScrollX(columns) }}
- />
- </Card>
+ {queryMode !== 'queue' && (
+ <Card styles={{ body: { padding: 0 } }}>
+ <Table
+ columns={columns}
+ dataSource={messages}
+ loading={queryLoading}
+ rowKey="msgId"
+ pagination={{
+ current: messagePage,
+ pageSize: messagePageSize,
+ total: messageTotal,
+ showSizeChanger: true,
+ showTotal: (total) => `共 ${total} 条消息`,
+ onChange: (page, pageSize) =>
+ void executeQuery(queryMode, currentQueryParams, page,
pageSize),
+ }}
+ size="small"
+ scroll={{ x: tableScrollX(columns) }}
+ />
+ </Card>
+ )}
{/* ── Message Detail Modal ── */}
<Modal
