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 746bdc8e6 feat(message): server-side pagination for message query
(#2531)
746bdc8e6 is described below
commit 746bdc8e6185ef4358bfeb1e838e9a0984def4f1
Author: aias00 <[email protected]>
AuthorDate: Sat Aug 22 16:12:16 2026 +0800
feat(message): server-side pagination for message query (#2531)
---
.../studio/instance/message/MessageController.java | 10 ++++
.../instance/message/MessageQueryPageVO.java | 35 +++++++++++++
.../studio/instance/message/MessageService.java | 17 ++++++
.../instance/message/MessageControllerTest.java | 16 ++++++
.../instance/message/MessageServiceTest.java | 20 ++++++++
web/src/api/message.test.ts | 13 ++++-
web/src/api/message.ts | 15 ++++++
.../pages/instance/__tests__/MessagePage.test.tsx | 60 ++++++++++++++++++----
.../__tests__/MessagePageAsyncState.test.tsx | 12 ++++-
web/src/pages/instance/message.tsx | 47 ++++++++++++++---
web/src/services/messageService.ts | 23 ++++++++-
11 files changed, 247 insertions(+), 21 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 3fb6da3c6..586944ff8 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
@@ -45,6 +45,16 @@ public class MessageController {
return Result.ok(messageService.queryMessages(instanceId, topic,
msgId, tag, key, startTime, endTime));
}
+ @GetMapping("/page")
+ public Result<MessageQueryPageVO> queryMessagesPage(@RequestParam String
instanceId,
+ @RequestParam(required = false) String topic,
@RequestParam(required = false) String msgId,
+ @RequestParam(required = false) String tag, @RequestParam(required
= false) String key,
+ @RequestParam(required = false) Long startTime,
@RequestParam(required = false) Long endTime,
+ @RequestParam(defaultValue = "1") int page,
@RequestParam(defaultValue = "50") int pageSize) {
+ return Result.ok(messageService.queryMessagesPage(instanceId, topic,
msgId, tag, key, startTime, endTime,
+ page, pageSize));
+ }
+
@GetMapping("/{msgId}/trace")
public Result<TraceRecordVO> getMessageTrace(@PathVariable String msgId,
@RequestParam String instanceId,
// Optional for the
Apache/Aliyun providers; the Tencent
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageQueryPageVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageQueryPageVO.java
new file mode 100644
index 000000000..99dd893e9
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageQueryPageVO.java
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+import java.util.List;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class MessageQueryPageVO {
+ private List<MessageRecordVO> items;
+ private long total;
+ private int page;
+ private int size;
+ private boolean resultMayBeTruncated;
+}
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 b5319be5f..fbeab1ab5 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,6 +31,8 @@ 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 TOPIC_QUERY_RESULT_LIMIT = 200;
private final MessageProvider messageProvider;
private final InstanceProviderRegistry providerRegistry;
@@ -47,6 +49,21 @@ public class MessageService {
return result;
}
+ 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");
+ }
+ List<MessageRecordVO> result = queryMessages(instanceId, topic, msgId,
tag, key, startTime, endTime);
+ long offset = (long) (page - 1) * pageSize;
+ int from = (int) Math.min(offset, result.size());
+ int to = Math.min(from + pageSize, result.size());
+ boolean topicQuery = StringUtils.hasText(topic) &&
!StringUtils.hasText(msgId)
+ && !StringUtils.hasText(key);
+ return MessageQueryPageVO.builder().items(result.subList(from,
to)).total(result.size()).page(page)
+ .size(pageSize).resultMayBeTruncated(topicQuery &&
result.size() >= TOPIC_QUERY_RESULT_LIMIT).build();
+ }
+
public TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic) {
if (!StringUtils.hasText(msgId)) {
throw new BusinessException(400, "msgId is required");
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageControllerTest.java
index e12071588..5dba50b20 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageControllerTest.java
@@ -71,6 +71,22 @@ class MessageControllerTest {
eq(1784246400000L), eq(1784332800000L));
}
+ @Test
+ void pagedQueryShouldPassPageAndReturnTruncationState() throws Exception {
+ MessageQueryPageVO page =
MessageQueryPageVO.builder().items(List.of()).total(200).page(2).size(50)
+ .resultMayBeTruncated(true).build();
+ when(messageService.queryMessagesPage("instance-a", "orders", null,
null, null, 1000L, 2000L, 2, 50))
+ .thenReturn(page);
+
+ mockMvc.perform(get("/api/messages/page").param("instanceId",
"instance-a").param("topic", "orders")
+ .param("startTime", "1000").param("endTime",
"2000").param("page", "2"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.total").value(200))
+
.andExpect(jsonPath("$.data.resultMayBeTruncated").value(true));
+
+ verify(messageService).queryMessagesPage("instance-a", "orders", null,
null, null, 1000L, 2000L, 2, 50);
+ }
+
@Test
void messageTraceShouldPassInstanceId() throws Exception {
TraceRecordVO trace =
TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
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 d30b43dd5..2c7a9bfc8 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
@@ -19,6 +19,7 @@ import java.util.List;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verify;
@@ -123,4 +124,23 @@ class MessageServiceTest {
verifyNoInteractions(provider, registry);
}
+
+ @Test
+ void pageQuerySlicesProviderResultAndSignalsBoundedTopicResult() {
+ MessageProvider provider = mock(MessageProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ QueryHistoryService history = mock(QueryHistoryService.class);
+ MessageService service = new MessageService(provider, registry,
history);
+ when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
+ when(provider.queryMessages("instance-a", "TopicA", null, null, null,
1000L, 2000L))
+ .thenReturn(java.util.stream.IntStream.range(0, 200)
+ .mapToObj(index ->
MessageRecordVO.builder().msgId("msg-" + index).build()).toList());
+
+ MessageQueryPageVO page = service.queryMessagesPage("instance-a",
"TopicA", null, null, null,
+ 1000L, 2000L, 2, 50);
+
+ assertThat(page.getItems()).hasSize(50);
+ assertThat(page.getTotal()).isEqualTo(200);
+ assertThat(page.isResultMayBeTruncated()).isTrue();
+ }
}
diff --git a/web/src/api/message.test.ts b/web/src/api/message.test.ts
index f43c08854..7c7a651af 100644
--- a/web/src/api/message.test.ts
+++ b/web/src/api/message.test.ts
@@ -18,7 +18,7 @@
import MockAdapter from 'axios-mock-adapter';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import client from './client';
-import { getMessageTrace, queryMessages } from './message';
+import { getMessageTrace, queryMessagePage, queryMessages } from './message';
const mock = new MockAdapter(client);
@@ -87,6 +87,17 @@ describe('message API', () => {
]);
});
+ it('uses the paged query contract and preserves its truncation state', async
() => {
+ const params = { instanceId: 'instance-1', topic: 'orders', page: 2,
pageSize: 50 };
+ const page = { items: [], total: 200, page: 2, size: 50,
resultMayBeTruncated: true };
+ mock.onGet('/messages/page').reply((config) => {
+ expect(config.params).toEqual(params);
+ return [200, { code: 200, data: page }];
+ });
+
+ await expect(queryMessagePage(params)).resolves.toEqual(page);
+ });
+
it('unwraps trace records with numeric timestamps', async () => {
const trace = {
nodes: [
diff --git a/web/src/api/message.ts b/web/src/api/message.ts
index 9bc1cfa12..9762e886d 100644
--- a/web/src/api/message.ts
+++ b/web/src/api/message.ts
@@ -44,6 +44,14 @@ export interface MessageQuery {
endTime?: number;
}
+export interface MessageQueryPage {
+ items: MessageRecord[];
+ total: number;
+ page: number;
+ size: number;
+ resultMayBeTruncated: boolean;
+}
+
const toStoreTimestamp = (storeTime: MessageRecord['storeTime']): number => {
if (typeof storeTime === 'number') return storeTime;
@@ -87,6 +95,13 @@ export async function queryMessages(params: MessageQuery) {
return sortMessagesByStoreTimeDesc(res.data.data);
}
+export async function queryMessagePage(
+ params: MessageQuery & { page?: number; pageSize?: number },
+) {
+ const res = await client.get<{ data: MessageQueryPage }>('/messages/page', {
params });
+ return { ...res.data.data, items:
sortMessagesByStoreTimeDesc(res.data.data.items) };
+}
+
export async function getMessageTrace(msgId: string, instanceId?: string,
topic?: string) {
const params: Record<string, string> = {};
if (instanceId !== undefined) params.instanceId = instanceId;
diff --git a/web/src/pages/instance/__tests__/MessagePage.test.tsx
b/web/src/pages/instance/__tests__/MessagePage.test.tsx
index b692192a3..f12750ddc 100644
--- a/web/src/pages/instance/__tests__/MessagePage.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePage.test.tsx
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import { App, Modal } from 'antd';
+import { App, message, Modal } from 'antd';
import { render, screen, waitFor } from '@testing-library/react';
import userEvent from '@testing-library/user-event';
import type React from 'react';
@@ -37,7 +37,17 @@ const instanceFilterMocks = vi.hoisted(() => ({
const QUERY_HISTORY_STORAGE_KEY = 'rocketmq-studio-message-query-history';
-vi.mock('../../../services/messageService', () => messageServiceMocks);
+vi.mock('../../../services/messageService', () => ({
+ ...messageServiceMocks,
+ queryMessagePage: ({ page = 1, pageSize = 50, ...params }: Record<string,
unknown>) =>
+ Promise.resolve(messageServiceMocks.queryMessages(params)).then((items) =>
({
+ items,
+ total: items.length,
+ page,
+ size: pageSize,
+ resultMayBeTruncated: false,
+ })),
+}));
vi.mock('../../../hooks/useInstanceFilter', () => instanceFilterMocks);
vi.mock('../../../services/instanceService', () => ({
@@ -256,14 +266,10 @@ describe('Message page query history', () => {
await user.click(screen.getByRole('button', { name: /最近查询/ }));
// Clearing requires confirmation: the dialog is commanded imperatively,
so spy on it
// and drive the confirm callback instead of depending on portal rendering
in jsdom.
- const confirmSpy = vi
- .spyOn(Modal, 'confirm')
- .mockImplementation((config) => {
- config.onOk?.();
- return { destroy: vi.fn(), update: vi.fn() } as unknown as ReturnType<
- typeof Modal.confirm
- >;
- });
+ const confirmSpy = vi.spyOn(Modal, 'confirm').mockImplementation((config)
=> {
+ config.onOk?.();
+ return { destroy: vi.fn(), update: vi.fn() } as unknown as
ReturnType<typeof Modal.confirm>;
+ });
await user.click(await screen.findByText('清空历史'));
expect(confirmSpy).toHaveBeenCalled();
confirmSpy.mockRestore();
@@ -343,6 +349,40 @@ describe('Message page query history', () => {
]);
});
+ it('does not rewrite query history or repeat the success toast when
paginating', async () => {
+ const user = userEvent.setup();
+ const setItemSpy = vi.spyOn(Storage.prototype, 'setItem');
+ const historyWrites = () =>
+ setItemSpy.mock.calls.filter(([key]) => key ===
QUERY_HISTORY_STORAGE_KEY);
+ const successSpy = vi.spyOn(message,
'success').mockImplementation(vi.fn());
+ messageServiceMocks.queryMessages.mockResolvedValue(
+ Array.from({ length: 60 }, (_, index) => createMessage(`MID-PAGE-${index
+ 1}`)),
+ );
+ renderWithProviders(<MessagePage />);
+
+ await user.click(screen.getByText('按 Message ID'));
+ await user.click(lastElement(screen.getAllByRole('combobox')));
+ await user.click(lastElement(await screen.findAllByText('order-create')));
+ await user.type(screen.getByPlaceholderText('输入 Message ID'), 'MID-PAGE');
+ await user.click(screen.getByRole('button', { name: /^search查询$/ }));
+
+ await waitFor(() => {
+ expect(messageServiceMocks.queryMessages).toHaveBeenCalledTimes(1);
+ });
+ expect(historyWrites()).toHaveLength(1);
+ expect(successSpy).toHaveBeenCalledTimes(1);
+
+ await user.click(screen.getByTitle('2'));
+ await waitFor(() => {
+ expect(messageServiceMocks.queryMessages).toHaveBeenCalledTimes(2);
+ });
+
+ // Paging only navigates the result set: the explicit query is written to
history once and
+ // the success toast is not re-announced on every page change.
+ expect(historyWrites()).toHaveLength(1);
+ expect(successSpy).toHaveBeenCalledTimes(1);
+ });
+
it('replays topic and key queries with their saved parameters', async () => {
const user = userEvent.setup();
const topicParams = {
diff --git a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
index 59bd3025f..3ee9c2bfa 100644
--- a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
@@ -32,7 +32,17 @@ const instanceFilterMocks = vi.hoisted(() => ({
useInstanceFilter: vi.fn(),
}));
-vi.mock('../../../services/messageService', () => serviceMocks);
+vi.mock('../../../services/messageService', () => ({
+ ...serviceMocks,
+ queryMessagePage: ({ page = 1, pageSize = 50, ...params }: Record<string,
unknown>) =>
+ Promise.resolve(serviceMocks.queryMessages(params)).then((items) => ({
+ items,
+ total: items.length,
+ page,
+ size: pageSize,
+ resultMayBeTruncated: false,
+ })),
+}));
vi.mock('../../../hooks/useInstanceFilter', () => instanceFilterMocks);
vi.mock('../../../services/instanceService', () => ({
diff --git a/web/src/pages/instance/message.tsx
b/web/src/pages/instance/message.tsx
index b36d99e28..9d91efc31 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -56,7 +56,7 @@ import { InstanceSelect } from
'../../components/InstanceSelect';
import MessageQueryHistoryDrawer from
'../../components/MessageQueryHistoryDrawer';
import { useLang } from '../../i18n/LangContext';
import type { MessageQuery, MessageRecord, TraceRecord } from
'../../api/message';
-import { getMessageTrace, queryMessages } from '../../services/messageService';
+import { getMessageTrace, queryMessagePage } from
'../../services/messageService';
import { listTopics } from '../../services/topicService';
import { useInstanceFilter } from '../../hooks/useInstanceFilter';
import { downloadBlob } from '../../utils/download';
@@ -312,6 +312,10 @@ const MessagePageContent = ({
const [keyInput, setKeyInput] = useState('');
const [msgIdInput, setMsgIdInput] = useState('');
const [messages, setMessages] = useState<MessageRecord[]>([]);
+ const [messageTotal, setMessageTotal] = useState(0);
+ const [messagePage, setMessagePage] = useState(1);
+ const [messagePageSize, setMessagePageSize] = useState(50);
+ const [resultMayBeTruncated, setResultMayBeTruncated] = useState(false);
const [queryLoading, setQueryLoading] = useState(false);
const [modalOpen, setModalOpen] = useState(false);
const [modalTab, setModalTab] = useState('content');
@@ -377,7 +381,13 @@ const MessagePageContent = ({
});
};
- const executeQuery = async (mode: QueryMode, params: MessageQuery) => {
+ const executeQuery = async (
+ mode: QueryMode,
+ params: MessageQuery,
+ page = 1,
+ pageSize = messagePageSize,
+ saveHistory = true,
+ ) => {
const requestGeneration = queryGenerationRef.current + 1;
queryGenerationRef.current = requestGeneration;
if (!selectedInstanceId) {
@@ -395,12 +405,23 @@ const MessagePageContent = ({
setQueryLoading(true);
setQueryError(null);
try {
- const result = await queryMessages({ ...normalizedParams, instanceId:
selectedInstanceId });
+ const result = await queryMessagePage({
+ ...normalizedParams,
+ instanceId: selectedInstanceId,
+ page,
+ pageSize,
+ });
if (queryGenerationRef.current !== requestGeneration) return;
- setMessages(result);
+ setMessages(result.items);
+ setMessageTotal(result.total);
+ setMessagePage(result.page);
+ setMessagePageSize(result.size);
+ setResultMayBeTruncated(result.resultMayBeTruncated);
setQueryError(null);
- saveRecentQuery(mode, normalizedParams);
- message.success(`查询完成，共 ${result.length} 条`);
+ if (saveHistory) {
+ saveRecentQuery(mode, normalizedParams);
+ message.success(`查询完成，共 ${result.total} 条`);
+ }
} catch (error) {
if (queryGenerationRef.current === requestGeneration) {
setQueryError(getErrorMessage(error, DEFAULT_QUERY_ERROR));
@@ -933,6 +954,14 @@ const MessagePageContent = ({
{queryError && (
<Alert showIcon type="warning" message={queryError} style={{
marginBottom: 16 }} />
)}
+ {resultMayBeTruncated && (
+ <Alert
+ showIcon
+ type="warning"
+ message="查询结果达到服务端扫描上限，当前总数可能不完整。"
+ style={{ marginBottom: 16 }}
+ />
+ )}
{/* ── Results Table ── */}
<Card styles={{ body: { padding: 0 } }}>
@@ -942,9 +971,13 @@ const MessagePageContent = ({
loading={queryLoading}
rowKey="msgId"
pagination={{
- pageSize: 50,
+ 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) }}
diff --git a/web/src/services/messageService.ts
b/web/src/services/messageService.ts
index a738c1281..ff9e0de50 100644
--- a/web/src/services/messageService.ts
+++ b/web/src/services/messageService.ts
@@ -3,6 +3,7 @@ import * as messageApi from '../api/message';
import { sortMessagesByStoreTimeDesc } from '../api/message';
import type {
MessageQuery,
+ MessageQueryPage,
MessageRecord,
TraceRecord,
DLQGroup,
@@ -49,6 +50,25 @@ export async function queryMessages(params: MessageQuery):
Promise<MessageRecord
return messageApi.queryMessages(params);
}
+export async function queryMessagePage(
+ params: MessageQuery & { page?: number; pageSize?: number },
+): Promise<MessageQueryPage> {
+ if (isMockMode()) {
+ const items = await queryMessages(params);
+ const page = params.page ?? 1;
+ const pageSize = params.pageSize ?? 50;
+ const from = Math.min((page - 1) * pageSize, items.length);
+ return {
+ items: items.slice(from, from + pageSize),
+ total: items.length,
+ page,
+ size: pageSize,
+ resultMayBeTruncated: false,
+ };
+ }
+ return messageApi.queryMessagePage(params);
+}
+
export async function getMessageTrace(
msgId: string,
instanceId?: string,
@@ -69,8 +89,7 @@ export async function listDLQGroups(
): Promise<DLQGroupPage> {
if (isMockMode()) {
const groups = (mockDLQGroups as unknown as DLQGroup[]).filter(
- (group) =>
- !search || group.groupName.includes(search) ||
group.dlqTopic.includes(search),
+ (group) => !search || group.groupName.includes(search) ||
group.dlqTopic.includes(search),
);
const from = Math.min((page - 1) * pageSize, groups.length);
return {
