(rocketmq-dashboard) branch rocketmq-studio updated: fix(message): preserve empty offset results from registered providers (#6662)

[lizhimin]

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 abe12af0d fix(message): preserve empty offset results from registered 
providers (#6662)
abe12af0d is described below

commit abe12af0d5bf7017fa38fb02c815611df78c1fe0
Author: 健康宝 <chenna...@gmail.com>
AuthorDate: Thu Oct 15 18:17:52 2026 +0800

    fix(message): preserve empty offset results from registered providers 
(#6662)
---
 .../studio/instance/message/MessageService.java    | 14 +++++---
 .../instance/message/MessageServiceTest.java       | 37 ++++++++++++++++++++++
 2 files changed, 47 insertions(+), 4 deletions(-)

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 f997a32e8..557363601 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
@@ -149,10 +149,16 @@ public class MessageService {
         if (offset < 0) {
             throw new BusinessException(400, "offset must not be negative");
         }
-        return providerRegistry.byInstanceId(instanceId)
-                .map(provider -> provider.pullMessageAtOffset(instanceId, 
topic, brokerName, queueId, offset))
-                .orElseGet(() -> messageProvider.pullMessageAtOffset(
-                        instanceId, topic, brokerName, queueId, offset));
+        // The registered provider's null is a legitimate "no message at this
+        // offset" answer; Optional.map would turn it into an empty Optional
+        // and the fallback below would issue a second, identical pull whose
+        // failure would surface as an error on a query that already succeeded
+        // once. The fallback is only for instances without a provider.
+        InstanceProvider provider = 
providerRegistry.byInstanceId(instanceId).orElse(null);
+        if (provider != null) {
+            return provider.pullMessageAtOffset(instanceId, topic, brokerName, 
queueId, offset);
+        }
+        return messageProvider.pullMessageAtOffset(instanceId, topic, 
brokerName, queueId, offset);
     }
 
     public DirectConsumeMessageResultVO 
consumeMessageDirectly(DirectConsumeMessageDTO request) {
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 55a0021b0..04c961dee 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
@@ -697,4 +697,41 @@ class MessageServiceTest {
 
         verifyNoInteractions(fallback);
     }
+
+    @Test
+    void pullMessageAtOffsetKeepsTheRegisteredProviderNullResultTest() {
+        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());
+        
when(registry.byInstanceId("cloud-instance")).thenReturn(Optional.of(provider));
+        when(provider.pullMessageAtOffset("cloud-instance", "TopicA", 
"broker-a", 0, 7L))
+                .thenReturn(null);
+
+        // A registered provider's null is the documented "no message at this
+        // offset" answer: it must not trigger the fallback (a second,
+        // identical pull whose failure would surface as an error here).
+        assertThat(service.pullMessageAtOffset("cloud-instance", "TopicA", 
"broker-a", 0, 7L))
+                .isNull();
+
+        verify(provider, org.mockito.Mockito.times(1))
+                .pullMessageAtOffset("cloud-instance", "TopicA", "broker-a", 
0, 7L);
+        verifyNoInteractions(fallback);
+    }
+
+    @Test
+    void pullMessageAtOffsetStillFallsBackWithoutAProviderTest() {
+        MessageProvider fallback = mock(MessageProvider.class);
+        MessageRecordVO record = 
MessageRecordVO.builder().msgId("msg-1").build();
+        InstanceProviderRegistry registry = 
mock(InstanceProviderRegistry.class);
+        MessageService service = new MessageService(fallback, registry, 
mock(QueryHistoryService.class),
+                mock(OperationAuditService.class), ownershipGuard());
+        
when(registry.byInstanceId("apache-instance")).thenReturn(Optional.empty());
+        when(fallback.pullMessageAtOffset("apache-instance", "TopicA", 
"broker-a", 0, 7L))
+                .thenReturn(record);
+
+        assertThat(service.pullMessageAtOffset("apache-instance", "TopicA", 
"broker-a", 0, 7L))
+                .isSameAs(record);
+    }
 }