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 11b10ba2d fix(message): continue scans after unmatched pull batches
(#6180)
11b10ba2d is described below
commit 11b10ba2db1e0f9e0f98b287ac4a71ff905d80ad
Author: zmuxuny <[email protected]>
AuthorDate: Mon Oct 12 00:16:29 2026 -0700
fix(message): continue scans after unmatched pull batches (#6180)
Both queue scan loops treated every non-`FOUND` pull status as the end of
the queue: `RocketMQMessageProvider.java:421-424` and
`RocketMQDLQProvider.java:500-503` break out of the scan as soon as the status
is anything other than `FOUND`. A broker answers `NO_MATCHED_MSG` with an
advanced `nextBeginOffset` when every message in the batch was filtered or
already deleted, so a single unmatched batch silently discarded the remainder
of the queue — for time-window topic queries and, throu [...]
Eight new tests cover both providers. Reverting either branch loses the
messages after the gap in `queryByTopicContinuesAfterUnmatchedBatchTest` and
`resendMessagesContinuesAfterUnmatchedBatchTest`; reverting the counter reset
breaks `queryByTopicResetsIllegalOffsetCounterAfterUnmatchedBatchTest` and its
DLQ twin, because interleaved `OFFSET_ILLEGAL` results then trip
`MAX_CONSECUTIVE_OFFSET_ILLEGAL`.
`listMessagesStopsUnmatchedBatchesAtWindowEndTest`,
`queryByTopicBoundsRepeatedUnmat [...]
---
.../provider/apache/RocketMQDLQProvider.java | 6 +
.../provider/apache/RocketMQMessageProvider.java | 6 +
.../provider/apache/RocketMQDLQProviderTest.java | 121 +++++++++++++++++++++
.../apache/RocketMQMessageProviderTest.java | 83 ++++++++++++++
4 files changed, 216 insertions(+)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
index 1375fa826..2e8159c78 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
@@ -497,6 +497,12 @@ public class RocketMQDLQProvider implements DLQProvider {
queue, dlqTopic, offset);
continue;
}
+ if (pullResult.getPullStatus() ==
PullStatus.NO_MATCHED_MSG) {
+ // The broker can skip removed messages even for a
wildcard pull.
+ // An advancing empty batch is not the end of the
queue.
+ consecutiveIllegalOffsets = 0;
+ continue;
+ }
if (pullResult.getPullStatus() != PullStatus.FOUND
|| pullResult.getMsgFoundList() == null) {
break;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
index 5ba5abd8a..c8a7ea26a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
@@ -418,6 +418,12 @@ public class RocketMQMessageProvider implements
MessageProvider {
queue, topic, offset);
continue;
}
+ if (pullResult.getPullStatus() ==
PullStatus.NO_MATCHED_MSG) {
+ // The broker can skip removed messages even for a
wildcard pull.
+ // An advancing empty batch is not the end of the
queue.
+ consecutiveIllegalOffsets = 0;
+ continue;
+ }
if (pullResult.getPullStatus() != PullStatus.FOUND
|| pullResult.getMsgFoundList() == null) {
break;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
index 0b09f6641..27ba07a9f 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
@@ -822,6 +822,127 @@ class RocketMQDLQProviderTest {
eq("SUCCESS"));
}
+ @Test
+ void listMessagesStopsUnmatchedBatchesAtWindowEndTest() throws Exception {
+ String topic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(topic, "broker-a", 0);
+
when(pullConsumer.fetchSubscribeMessageQueues(topic)).thenReturn(Set.of(queue));
+ when(pullConsumer.searchOffset(queue, 100L)).thenReturn(10L);
+ when(pullConsumer.searchOffset(queue, 201L)).thenReturn(13L);
+ when(pullConsumer.pull(eq(queue), eq("*"), anyLong(),
eq(32))).thenAnswer(invocation ->
+ new PullResult(PullStatus.NO_MATCHED_MSG, (Long)
invocation.getArgument(2) + 1,
+ 0L, 13L, null));
+
+ assertThat(provider.listMessages("instance-a", "group-a", 100L, 200L,
1, 20).getTotal()).isZero();
+
+ verify(pullConsumer, times(3)).pull(eq(queue), eq("*"), anyLong(),
eq(32));
+ }
+
+ @Test
+ void listMessagesStopsWhenUnmatchedBatchDoesNotAdvanceTest() throws
Exception {
+ String topic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(topic, "broker-a", 0);
+
when(pullConsumer.fetchSubscribeMessageQueues(topic)).thenReturn(Set.of(queue));
+ when(pullConsumer.searchOffset(queue, 100L)).thenReturn(10L);
+ when(pullConsumer.searchOffset(queue, 201L)).thenReturn(13L);
+ when(pullConsumer.pull(queue, "*", 10L, 32))
+ .thenReturn(new PullResult(PullStatus.NO_MATCHED_MSG, 10L, 0L,
13L, null));
+
+ assertThat(provider.listMessages("instance-a", "group-a", 100L, 200L,
1, 20).getTotal()).isZero();
+
+ verify(pullConsumer, times(1)).pull(eq(queue), eq("*"), anyLong(),
eq(32));
+ }
+
+ @Test
+ void resendMessagesContinuesAfterUnmatchedBatchTest() throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(dlqTopic, "broker-a", 0);
+ MessageExt deadLetter = new MessageExt();
+ deadLetter.setMsgId("dlq-after-gap");
+ deadLetter.setTopic(dlqTopic);
+ deadLetter.setBody(new byte[] {1, 2, 3});
+ deadLetter.setStoreTimestamp(150L);
+ PullResult unmatchedBatch = new PullResult(PullStatus.NO_MATCHED_MSG,
20L, 0L, 50L, null);
+ PullResult foundAfterGap = new PullResult(PullStatus.FOUND, 40L, 20L,
50L, List.of(deadLetter));
+ PullResult endOfQueue = new PullResult(PullStatus.NO_NEW_MSG, 50L,
40L, 50L, List.of());
+ SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.SEND_OK);
+
+
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(pullConsumer.searchOffset(queue, 100L)).thenReturn(10L);
+ when(pullConsumer.searchOffset(queue, 201L)).thenReturn(50L);
+ when(pullConsumer.pull(queue, "*", 10L,
32)).thenReturn(unmatchedBatch);
+ when(pullConsumer.pull(queue, "*", 20L, 32)).thenReturn(foundAfterGap);
+ when(pullConsumer.pull(queue, "*", 40L, 32)).thenReturn(endOfQueue);
+ when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("orders"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
+ assertThat(provider.resendMessages("instance-a", "group-a", 100L,
200L, "orders"))
+ .extracting("matched", "resent", "failed", "outcome")
+ .containsExactly(1, 1, 0, "SUCCESS");
+
+ DefaultMQPullConsumer consumer = pullConsumer;
+ verify(consumer).pull(queue, "*", 10L, 32);
+ verify(consumer).pull(queue, "*", 20L, 32);
+ verify(consumer).pull(queue, "*", 40L, 32);
+ verify(dlqProducer).send(any(Message.class));
+ verify(auditService).record(
+ eq("RESEND_DLQ"),
+ eq("DLQ"),
+ eq("group-a"),
+ isNull(),
+ contains("matched=1, resent=1, failed=0"),
+ eq("SUCCESS"));
+ }
+
+ @Test
+ void resendMessagesResetsIllegalOffsetCounterAfterUnmatchedBatchTest()
throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(dlqTopic, "broker-a", 0);
+ MessageExt deadLetter = new MessageExt();
+ deadLetter.setMsgId("dlq-after-gap");
+ deadLetter.setTopic(dlqTopic);
+ deadLetter.setBody(new byte[] {1, 2, 3});
+ deadLetter.setStoreTimestamp(150L);
+ PullResult unmatchedBatch = new PullResult(PullStatus.NO_MATCHED_MSG,
17L, 0L, 50L, null);
+ PullResult foundAfterGap = new PullResult(PullStatus.FOUND, 40L, 20L,
50L, List.of(deadLetter));
+ PullResult endOfQueue = new PullResult(PullStatus.NO_NEW_MSG, 50L,
40L, 50L, List.of());
+ SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.SEND_OK);
+
+
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(pullConsumer.searchOffset(queue, 100L)).thenReturn(10L);
+ when(pullConsumer.searchOffset(queue, 201L)).thenReturn(50L);
+ for (long offset : new long[] {10L, 11L, 12L, 17L, 18L, 19L}) {
+ when(pullConsumer.pull(queue, "*", offset, 32))
+ .thenReturn(new PullResult(PullStatus.OFFSET_ILLEGAL,
offset + 1, 0L, 50L, null));
+ }
+ when(pullConsumer.pull(queue, "*", 13L,
32)).thenReturn(unmatchedBatch);
+ when(pullConsumer.pull(queue, "*", 20L, 32)).thenReturn(foundAfterGap);
+ when(pullConsumer.pull(queue, "*", 40L, 32)).thenReturn(endOfQueue);
+ when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("orders"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
+ assertThat(provider.resendMessages("instance-a", "group-a", 100L,
200L, "orders"))
+ .extracting("matched", "resent", "failed", "outcome")
+ .containsExactly(1, 1, 0, "SUCCESS");
+
+ DefaultMQPullConsumer consumer = pullConsumer;
+ verify(consumer).pull(queue, "*", 10L, 32);
+ verify(consumer).pull(queue, "*", 20L, 32);
+ verify(consumer).pull(queue, "*", 40L, 32);
+ verify(dlqProducer).send(any(Message.class));
+ verify(auditService).record(
+ eq("RESEND_DLQ"),
+ eq("DLQ"),
+ eq("group-a"),
+ isNull(),
+ contains("matched=1, resent=1, failed=0"),
+ eq("SUCCESS"));
+ }
+
@Test
void resendMessagesCountsNonSendOkResultsAsFailures() throws Exception {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
index d2d470831..405dbee4c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
@@ -704,6 +704,89 @@ class RocketMQMessageProviderTest {
verify(pullConsumer).pull(queue, "*", 40L, 32);
}
+ @Test
+ void queryByTopicBoundsRepeatedUnmatchedBatchesTest() throws Exception {
+ MessageQueue queue = new MessageQueue("TopicA", "broker-a", 0);
+
when(pullConsumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of(queue));
+ mockQueueWindow(pullConsumer, queue, 100L, 200L, 0L, 0L, 100L, 100L);
+ when(pullConsumer.pull(eq(queue), eq("*"), anyLong(),
eq(32))).thenAnswer(invocation ->
+ new PullResult(PullStatus.NO_MATCHED_MSG, (Long)
invocation.getArgument(2) + 1,
+ 0L, 100L, null));
+
+ assertThat(provider.queryMessages("instance-a", "TopicA", null, null,
null, 100L, 200L)).isEmpty();
+
+ verify(pullConsumer, times(35)).pull(eq(queue), eq("*"), anyLong(),
eq(32));
+ }
+
+ @Test
+ void queryByTopicStopsWhenUnmatchedBatchDoesNotAdvanceTest() throws
Exception {
+ MessageQueue queue = new MessageQueue("TopicA", "broker-a", 0);
+
when(pullConsumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of(queue));
+ mockQueueWindow(pullConsumer, queue, 100L, 200L, 10L, 10L, 30L, 30L);
+ when(pullConsumer.pull(queue, "*", 10L, 32))
+ .thenReturn(new PullResult(PullStatus.NO_MATCHED_MSG, 10L, 0L,
30L, null));
+
+ assertThat(provider.queryMessages("instance-a", "TopicA", null, null,
null, 100L, 200L)).isEmpty();
+
+ verify(pullConsumer, times(1)).pull(eq(queue), eq("*"), anyLong(),
eq(32));
+ }
+
+ @Test
+ void queryByTopicContinuesAfterUnmatchedBatchTest() throws Exception {
+ MessageQueue queue = new MessageQueue("TopicA", "broker-a", 0);
+ MessageExt message = new MessageExt();
+ message.setMsgId("msg-after-gap");
+ message.setTopic("TopicA");
+ message.setBody("payload".getBytes(StandardCharsets.UTF_8));
+ message.setStoreTimestamp(150L);
+ PullResult unmatchedBatch = new PullResult(PullStatus.NO_MATCHED_MSG,
20L, 0L, 50L, null);
+ PullResult foundAfterGap = new PullResult(PullStatus.FOUND, 40L, 20L,
50L, List.of(message));
+ PullResult endOfQueue = new PullResult(PullStatus.NO_NEW_MSG, 50L,
40L, 50L, List.of());
+
when(pullConsumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of(queue));
+ mockQueueWindow(pullConsumer, queue, 100L, 200L, 10L, 10L, 50L, 50L);
+ when(pullConsumer.pull(eq(queue), eq("*"), eq(10L),
eq(32))).thenReturn(unmatchedBatch);
+ when(pullConsumer.pull(eq(queue), eq("*"), eq(20L),
eq(32))).thenReturn(foundAfterGap);
+ when(pullConsumer.pull(eq(queue), eq("*"), eq(40L),
eq(32))).thenReturn(endOfQueue);
+
+ List<MessageRecordVO> messages = provider.queryMessages(
+ "instance-a", "TopicA", null, null, null, 100L, 200L);
+
+
assertThat(messages).extracting(MessageRecordVO::getMsgId).containsExactly("msg-after-gap");
+ verify(pullConsumer).pull(queue, "*", 10L, 32);
+ verify(pullConsumer).pull(queue, "*", 20L, 32);
+ verify(pullConsumer).pull(queue, "*", 40L, 32);
+ }
+
+ @Test
+ void queryByTopicResetsIllegalOffsetCounterAfterUnmatchedBatchTest()
throws Exception {
+ MessageQueue queue = new MessageQueue("TopicA", "broker-a", 0);
+ MessageExt message = new MessageExt();
+ message.setMsgId("msg-after-gap");
+ message.setTopic("TopicA");
+ message.setBody("payload".getBytes(StandardCharsets.UTF_8));
+ message.setStoreTimestamp(150L);
+ PullResult unmatchedBatch = new PullResult(PullStatus.NO_MATCHED_MSG,
17L, 0L, 50L, null);
+ PullResult foundAfterGap = new PullResult(PullStatus.FOUND, 40L, 20L,
50L, List.of(message));
+ PullResult endOfQueue = new PullResult(PullStatus.NO_NEW_MSG, 50L,
40L, 50L, List.of());
+
when(pullConsumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of(queue));
+ mockQueueWindow(pullConsumer, queue, 100L, 200L, 10L, 10L, 50L, 50L);
+ for (long offset : new long[] {10L, 11L, 12L, 17L, 18L, 19L}) {
+ when(pullConsumer.pull(queue, "*", offset, 32))
+ .thenReturn(new PullResult(PullStatus.OFFSET_ILLEGAL,
offset + 1, 0L, 50L, null));
+ }
+ when(pullConsumer.pull(queue, "*", 13L,
32)).thenReturn(unmatchedBatch);
+ when(pullConsumer.pull(eq(queue), eq("*"), eq(20L),
eq(32))).thenReturn(foundAfterGap);
+ when(pullConsumer.pull(eq(queue), eq("*"), eq(40L),
eq(32))).thenReturn(endOfQueue);
+
+ List<MessageRecordVO> messages = provider.queryMessages(
+ "instance-a", "TopicA", null, null, null, 100L, 200L);
+
+
assertThat(messages).extracting(MessageRecordVO::getMsgId).containsExactly("msg-after-gap");
+ verify(pullConsumer).pull(queue, "*", 10L, 32);
+ verify(pullConsumer).pull(queue, "*", 20L, 32);
+ verify(pullConsumer).pull(queue, "*", 40L, 32);
+ }
+
@Test
void queryByTopicKeepsNewestMessagesWhenEarlierQueuesFillTheDefaultLimit()
throws Exception {
MessageQueue olderQueue = new MessageQueue("TopicA", "broker-a", 0);
