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 816dcb641 fix(tencent): page the message query by consumed rows, not
collected records (#6518)
816dcb641 is described below
commit 816dcb6416f2d7fb7f56db61dd746dd77d7655ff
Author: Apulupie <[email protected]>
AuthorDate: Thu Oct 15 18:40:02 2026 +0800
fix(tencent): page the message query by consumed rows, not collected
records (#6518)
`queryMessagesDetailed` derived the next page offset from the records it
had collected
(`request.setOffset((long) result.size())`) instead of the rows the API had
already
returned. The SDK maps a JSON null array element to a null element and the
loop skips it,
so every skipped entry made the next offset one row too small: the API
re-returned the
tail of the page, the result contained duplicate msgIds, and
`isLastPage(..., result.size())`
made the completion check drift by the same amount. The Aliyun sibling
counts raw
provider rows in its paging bookkeeping, and this file already guards null
array elements
in three places.
A `fetchedRows` counter now grows by `returned` on every page and feeds
both the next
offset and the last-page check; `result.size()` keeps serving only the
result budget. The
duplicated short-page comment above `isLastPage` is removed while editing
those lines.
---
.../provider/tencent/TencentInstanceProvider.java | 17 ++++++----
.../tencent/TencentInstanceProviderTest.java | 36 ++++++++++++++++++++++
2 files changed, 47 insertions(+), 6 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
index cd363ace1..c3d0d7632 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
@@ -642,6 +642,10 @@ public class TencentInstanceProvider implements
InstanceProvider {
// is not server-paginated (pagination=false), so page through the
whole result set here.
String taskRequestId = UUID.randomUUID().toString();
List<MessageRecordVO> result = new ArrayList<>();
+ // Rows the API has already returned, which is what the next offset
must be derived from:
+ // a page can carry entries the provider skips (the SDK maps a JSON
null array element to a
+ // null element), so the number of collected records is not the number
of consumed rows.
+ long fetchedRows = 0L;
boolean mayBeTruncated = false;
for (int page = 0; page < MAX_PAGES && result.size() <
MESSAGE_QUERY_HARD_LIMIT; page++) {
DescribeMessageListRequest request = new
DescribeMessageListRequest();
@@ -658,7 +662,7 @@ public class TencentInstanceProvider implements
InstanceProvider {
if (StringUtils.hasText(tag)) {
request.setTag(tag);
}
- request.setOffset((long) result.size());
+ request.setOffset(fetchedRows);
request.setLimit((long) MESSAGE_LIMIT);
DescribeMessageListResponse response =
clientFactory.call(context.credentialId(), context.regionId(),
client -> client.DescribeMessageList(request));
@@ -682,8 +686,9 @@ public class TencentInstanceProvider implements
InstanceProvider {
mayBeTruncated = true;
break;
}
+ fetchedRows += returned;
// Missing totals retain the existing short-page termination
fallback.
- if (isLastPage(returned, total, result.size())) {
+ if (isLastPage(returned, total, fetchedRows)) {
break;
}
if (result.size() >= MESSAGE_QUERY_HARD_LIMIT) {
@@ -899,12 +904,12 @@ public class TencentInstanceProvider implements
InstanceProvider {
}
}
- private static boolean isLastPage(int returned, long total, int collected)
{
+ private static boolean isLastPage(int returned, long total, long
fetchedRows) {
// The last page is one that returned no rows or fewer rows than
requested. When the API
- // reports a TotalCount, also stop once all expected results have been
collected.
+ // reports a TotalCount, also stop once all expected rows have been
fetched.
boolean shortPage = returned == 0 || returned < MESSAGE_LIMIT;
- boolean allCollected = total > 0 && collected >= total;
- return shortPage || allCollected;
+ boolean allFetched = total > 0 && fetchedRows >= total;
+ return shortPage || allFetched;
}
private static long parseTraceTime(String value) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
index 02598f59f..5ef54fb12 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
@@ -1210,6 +1210,42 @@ class TencentInstanceProviderTest {
.DescribeMessageList(any());
}
+ @Test
+ void
queryMessagesShouldAdvanceByConsumedRowsWhenAPageSkipsAnAbsentEntryTest()
throws Exception {
+ // The SDK maps a JSON null array element to a null element and the
provider skips it, but
+ // the next page must still start after every row the API already
returned.
+ when(client.DescribeMessageList(any())).thenAnswer(invocation -> {
+ DescribeMessageListRequest request = invocation.getArgument(0);
+ int start = request.getOffset().intValue();
+ int size = Math.min(request.getLimit().intValue(), Math.max(150 -
start, 0));
+ MessageItem[] items = new MessageItem[size];
+ for (int i = 0; i < size; i++) {
+ MessageItem item = new MessageItem();
+ item.setMsgId("MSG-" + (start + i + 1));
+ item.setProduceTime("2024-09-12 14:06:55,591");
+ items[i] = item;
+ }
+ if (start == 0) {
+ items[50] = null;
+ }
+ DescribeMessageListResponse response = new
DescribeMessageListResponse();
+ response.setData(items);
+ response.setTotalCount(150L);
+ return response;
+ });
+
+ List<MessageRecordVO> messages =
provider.queryMessages(STUDIO_INSTANCE_ID, "orders", null,
+ null, null, 1600000000000L, 1600001000000L);
+
+ ArgumentCaptor<DescribeMessageListRequest> captor =
ArgumentCaptor.forClass(DescribeMessageListRequest.class);
+ verify(client,
org.mockito.Mockito.times(2)).DescribeMessageList(captor.capture());
+ assertThat(captor.getAllValues().get(0).getOffset()).isZero();
+ assertThat(captor.getAllValues().get(1).getOffset())
+ .isEqualTo((long) TencentInstanceProvider.MESSAGE_LIMIT);
+
assertThat(messages).extracting(MessageRecordVO::getMsgId).doesNotHaveDuplicates();
+ assertThat(messages).hasSize(149);
+ }
+
@Test
void queryMessagesShouldReportCompleteWhenTencentReturnsAShortPage()
throws Exception {
MessageItem one = new MessageItem();
