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 720ed854f fix(message): stabilize query results before pagination
(#5126)
720ed854f is described below
commit 720ed854f2e1030b4cd772a2c3b738a3b2ca9cf9
Author: coder999o <[email protected]>
AuthorDate: Thu Oct 1 17:30:38 2026 +0800
fix(message): stabilize query results before pagination (#5126)
---
.../studio/instance/message/MessageService.java | 8 +++-
.../instance/message/MessageServiceTest.java | 47 ++++++++++++++++++++++
2 files changed, 54 insertions(+), 1 deletion(-)
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 7cae094e3..f997a32e8 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
@@ -30,6 +30,7 @@ import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
+import java.util.Comparator;
import java.util.List;
@Service
@@ -40,6 +41,10 @@ public class MessageService {
private static final long MAX_TOPIC_QUERY_WINDOW_MILLIS = 7L * 24 * 60 *
60 * 1000;
private static final int MAX_PAGE_SIZE = 200;
private static final int TOPIC_QUERY_RESULT_LIMIT = 200;
+ private static final Comparator<MessageRecordVO> PAGE_ORDER = Comparator
+ .comparingLong(MessageRecordVO::getStoreTime)
+ .thenComparing(MessageRecordVO::getMsgId,
Comparator.nullsFirst(String::compareTo))
+ .reversed();
private final MessageProvider messageProvider;
private final InstanceProviderRegistry providerRegistry;
@@ -97,7 +102,8 @@ public class MessageService {
// tell the user when deeper pages may be empty.
MessageQueryResult queryResult = queryMessagesDetailed(
instanceId, topic, msgId, tag, key, startTime, endTime, page
== 1);
- List<MessageRecordVO> result = queryResult.messages();
+ // Each page re-queries the provider; broker response order must not
move rows across page boundaries.
+ List<MessageRecordVO> result =
queryResult.messages().stream().sorted(PAGE_ORDER).toList();
long offset = (long) (page - 1) * pageSize;
int from = (int) Math.min(offset, result.size());
int to = Math.min(from + pageSize, result.size());
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 42674f661..55a0021b0 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
@@ -25,6 +25,8 @@ import org.apache.rocketmq.studio.provider.InstanceProvider;
import org.apache.rocketmq.studio.provider.InstanceCapability;
import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.List;
import java.util.Optional;
@@ -379,6 +381,51 @@ class MessageServiceTest {
assertThat(page.isResultMayBeTruncated()).isTrue();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void pageQueryKeepsStableOrderAcrossChangingProviderResultsTest(boolean
registeredProvider) {
+ MessageProvider fallback = mock(MessageProvider.class);
+ InstanceProvider provider = mock(InstanceProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(fallback, registry,
mock(QueryHistoryService.class),
+ mock(OperationAuditService.class), ownershipGuard());
+ MessageRecordVO newest =
MessageRecordVO.builder().msgId("newest").storeTime(3000L).build();
+ MessageRecordVO sameTimeA =
MessageRecordVO.builder().msgId("a").storeTime(2000L).build();
+ MessageRecordVO sameTimeB =
MessageRecordVO.builder().msgId("b").storeTime(2000L).build();
+ MessageRecordVO missingId =
MessageRecordVO.builder().storeTime(2000L).build();
+ MessageRecordVO oldest =
MessageRecordVO.builder().msgId("oldest").storeTime(1000L).build();
+ MessageQueryResult firstResult = MessageQueryResult.truncated(
+ List.of(oldest, sameTimeA, missingId, newest, sameTimeB));
+ MessageQueryResult secondResult = MessageQueryResult.truncated(
+ List.of(sameTimeB, newest, oldest, missingId, sameTimeA));
+ when(registry.byInstanceId("instance-a")).thenReturn(
+ registeredProvider ? Optional.of(provider) : Optional.empty());
+ if (registeredProvider) {
+ when(provider.queryMessagesDetailed("instance-a", "TopicA", null,
null, "key-1", null, null))
+ .thenReturn(firstResult, secondResult, firstResult);
+ } else {
+ when(fallback.queryMessagesDetailed("instance-a", "TopicA", null,
null, "key-1", null, null))
+ .thenReturn(firstResult, secondResult, firstResult);
+ }
+
+ MessageQueryPageVO firstPage = service.queryMessagesPage(
+ "instance-a", "TopicA", null, null, "key-1", null, null, 1, 2);
+ MessageQueryPageVO secondPage = service.queryMessagesPage(
+ "instance-a", "TopicA", null, null, "key-1", null, null, 2, 2);
+ MessageQueryPageVO thirdPage = service.queryMessagesPage(
+ "instance-a", "TopicA", null, null, "key-1", null, null, 3, 2);
+
+ assertThat(firstPage.getItems()).containsExactly(newest, sameTimeB);
+ assertThat(secondPage.getItems()).containsExactly(sameTimeA,
missingId);
+ assertThat(thirdPage.getItems()).containsExactly(oldest);
+ assertThat(List.of(firstPage, secondPage, thirdPage)).allSatisfy(page
-> {
+ assertThat(page.getTotal()).isEqualTo(5);
+ assertThat(page.isResultMayBeTruncated()).isTrue();
+ });
+ assertThat(firstResult.messages()).containsExactly(oldest, sameTimeA,
missingId, newest, sameTimeB);
+ assertThat(secondResult.messages()).containsExactly(sameTimeB, newest,
oldest, missingId, sameTimeA);
+ }
+
@Test
void pageQueryPropagatesProviderTruncationBelowTheGenericTopicThreshold() {
MessageProvider provider = mock(MessageProvider.class);
