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 4ee173ad7 fix(studio): route message operations by instance provider
(#4909)
4ee173ad7 is described below
commit 4ee173ad71ae777f01111394a88358958b9abad6
Author: Yexi Xiang <[email protected]>
AuthorDate: Sat Sep 26 18:49:47 2026 +0800
fix(studio): route message operations by instance provider (#4909)
test(studio): adapt message service tests to ownership guard
---
.../studio/instance/message/MessageService.java | 14 +++++--
.../rocketmq/studio/provider/InstanceProvider.java | 24 +++++++++++
.../provider/apache/ApacheInstanceProvider.java | 18 ++++++++
.../instance/message/MessageServiceTest.java | 48 ++++++++++++++++++++++
.../apache/ApacheInstanceProviderTest.java | 23 +++++++++++
5 files changed, 124 insertions(+), 3 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 693353e24..7cae094e3 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
@@ -79,7 +79,10 @@ public class MessageService {
}
validateProvidedTimeWindow(startTime, endTime);
log.info("Querying message by unique key: topic={}, uniqueKey={}",
topic, uniqueKey);
- return messageProvider.queryMessageByUniqueKey(instanceId, topic,
uniqueKey, startTime, endTime);
+ return providerRegistry.byInstanceId(instanceId)
+ .map(provider -> provider.queryMessageByUniqueKey(instanceId,
topic, uniqueKey, startTime, endTime))
+ .orElseGet(() -> messageProvider.queryMessageByUniqueKey(
+ instanceId, topic, uniqueKey, startTime, endTime));
}
public MessageQueryPageVO queryMessagesPage(String instanceId, String
topic, String msgId, String tag,
@@ -121,7 +124,9 @@ public class MessageService {
if (!StringUtils.hasText(topic)) {
throw new BusinessException(400, "topic is required");
}
- return messageProvider.getQueueOffsets(instanceId, topic);
+ return providerRegistry.byInstanceId(instanceId)
+ .map(provider -> provider.getQueueOffsets(instanceId, topic))
+ .orElseGet(() -> messageProvider.getQueueOffsets(instanceId,
topic));
}
public MessageRecordVO pullMessageAtOffset(String instanceId, String
topic, String brokerName,
@@ -138,7 +143,10 @@ public class MessageService {
if (offset < 0) {
throw new BusinessException(400, "offset must not be negative");
}
- return messageProvider.pullMessageAtOffset(instanceId, topic,
brokerName, queueId, offset);
+ return providerRegistry.byInstanceId(instanceId)
+ .map(provider -> provider.pullMessageAtOffset(instanceId,
topic, brokerName, queueId, offset))
+ .orElseGet(() -> messageProvider.pullMessageAtOffset(
+ instanceId, topic, brokerName, queueId, offset));
}
public DirectConsumeMessageResultVO
consumeMessageDirectly(DirectConsumeMessageDTO request) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
index eb596c75b..b62b03bb9 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
@@ -29,6 +29,7 @@ import
org.apache.rocketmq.studio.instance.message.MessageRecordVO;
import org.apache.rocketmq.studio.instance.message.DirectConsumeMessageDTO;
import
org.apache.rocketmq.studio.instance.message.DirectConsumeMessageResultVO;
import org.apache.rocketmq.studio.instance.message.MessageQueryResult;
+import org.apache.rocketmq.studio.instance.message.QueueOffsetVO;
import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
@@ -205,6 +206,29 @@ public interface InstanceProvider {
TraceRecordVO getMessageTrace(String instanceId, String msgId, String
topic);
+ /**
+ * Lookup by the client-generated unique key (UNIQ_KEY index).
+ */
+ default List<MessageRecordVO> queryMessageByUniqueKey(String instanceId,
String topic, String uniqueKey,
+ Long startTime, Long
endTime) {
+ throw new UnsupportedOperationException("Unique-key message lookup is
not supported");
+ }
+
+ /**
+ * Lists the read queues available for the topic.
+ */
+ default List<QueueOffsetVO> getQueueOffsets(String instanceId, String
topic) {
+ throw new UnsupportedOperationException("Queue browsing is not
supported");
+ }
+
+ /**
+ * Pulls one message from a broker queue at an exact offset.
+ */
+ default MessageRecordVO pullMessageAtOffset(String instanceId, String
topic, String brokerName,
+ int queueId, long offset) {
+ throw new UnsupportedOperationException("Offset message browsing is
not supported");
+ }
+
default SendMessageVO sendMessage(SendMessageDTO request) {
throw new UnsupportedOperationException("Message sending is not
supported");
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
index 02e834a57..9651992d0 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
@@ -30,6 +30,7 @@ import
org.apache.rocketmq.studio.instance.message.MessageQueryResult;
import org.apache.rocketmq.studio.instance.message.DirectConsumeMessageDTO;
import
org.apache.rocketmq.studio.instance.message.DirectConsumeMessageResultVO;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
+import org.apache.rocketmq.studio.instance.message.QueueOffsetVO;
import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
@@ -222,6 +223,23 @@ public class ApacheInstanceProvider implements
InstanceProvider {
return messageProvider.getMessageTrace(instanceId, msgId, topic);
}
+ @Override
+ public List<MessageRecordVO> queryMessageByUniqueKey(String instanceId,
String topic, String uniqueKey,
+ Long startTime, Long
endTime) {
+ return messageProvider.queryMessageByUniqueKey(instanceId, topic,
uniqueKey, startTime, endTime);
+ }
+
+ @Override
+ public List<QueueOffsetVO> getQueueOffsets(String instanceId, String
topic) {
+ return messageProvider.getQueueOffsets(instanceId, topic);
+ }
+
+ @Override
+ public MessageRecordVO pullMessageAtOffset(String instanceId, String
topic, String brokerName,
+ int queueId, long offset) {
+ return messageProvider.pullMessageAtOffset(instanceId, topic,
brokerName, queueId, offset);
+ }
+
@Override
public SendMessageVO sendMessage(SendMessageDTO request) {
return adminClient.sendMessage(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 55f0ecd17..42674f661 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
@@ -602,4 +602,52 @@ class MessageServiceTest {
verify(provider).queryMessageByUniqueKey("instance-a", "TopicA",
"uniq-1", 100L, 200L);
}
+
+ @Test
+ void
instanceScopedMessageOperationsResolveTheSelectedProviderBeforeFallbackTest() {
+ MessageProvider fallback = mock(MessageProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(fallback, registry,
mock(QueryHistoryService.class),
+ mock(OperationAuditService.class), ownershipGuard());
+ BusinessException unsupported = new BusinessException(501,
+ "Selected instance does not support this message operation");
+ when(registry.byInstanceId("cloud-instance")).thenThrow(unsupported);
+
+ assertThatThrownBy(() -> service.queryMessageByUniqueKey(
+ "cloud-instance", "TopicA", "uniq-1", null, null))
+ .isSameAs(unsupported);
+ assertThatThrownBy(() -> service.getQueueOffsets("cloud-instance",
"TopicA"))
+ .isSameAs(unsupported);
+ assertThatThrownBy(() -> service.pullMessageAtOffset(
+ "cloud-instance", "TopicA", "broker-a", 0, 0))
+ .isSameAs(unsupported);
+
+ verify(registry,
org.mockito.Mockito.times(3)).byInstanceId("cloud-instance");
+ verifyNoInteractions(fallback);
+ }
+
+ @Test
+ void instanceScopedMessageOperationsUseTheSelectedProviderResultsTest() {
+ 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 record =
MessageRecordVO.builder().msgId("msg-1").build();
+ QueueOffsetVO queue =
QueueOffsetVO.builder().brokerName("broker-a").queueId(0).build();
+
when(registry.byInstanceId("cloud-instance")).thenReturn(Optional.of(provider));
+ when(provider.queryMessageByUniqueKey("cloud-instance", "TopicA",
"uniq-1", 100L, 200L))
+ .thenReturn(List.of(record));
+ when(provider.getQueueOffsets("cloud-instance",
"TopicA")).thenReturn(List.of(queue));
+ when(provider.pullMessageAtOffset("cloud-instance", "TopicA",
"broker-a", 0, 7L))
+ .thenReturn(record);
+
+ assertThat(service.queryMessageByUniqueKey("cloud-instance", "TopicA",
"uniq-1", 100L, 200L))
+ .containsExactly(record);
+ assertThat(service.getQueueOffsets("cloud-instance",
"TopicA")).containsExactly(queue);
+ assertThat(service.pullMessageAtOffset("cloud-instance", "TopicA",
"broker-a", 0, 7L))
+ .isSameAs(record);
+
+ verifyNoInteractions(fallback);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
index 51bb098ef..a7d13d631 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
@@ -24,6 +24,8 @@ import
org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.topic.TopicVO;
import org.apache.rocketmq.studio.instance.message.MessageProvider;
import org.apache.rocketmq.studio.instance.message.MessageQueryResult;
+import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
+import org.apache.rocketmq.studio.instance.message.QueueOffsetVO;
import org.apache.rocketmq.studio.provider.InstanceCapability;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -135,6 +137,27 @@ class ApacheInstanceProviderTest {
"inst-1", "TopicA", null, null, "order-1", 100L, 200L);
}
+ @Test
+ void nativeMessageOperationsShouldDelegateToMessageProviderTest() {
+ MessageRecordVO record =
MessageRecordVO.builder().msgId("msg-1").build();
+ QueueOffsetVO queue =
QueueOffsetVO.builder().brokerName("broker-a").queueId(0).build();
+ when(messageProvider.queryMessageByUniqueKey("inst-1", "TopicA",
"uniq-1", 100L, 200L))
+ .thenReturn(java.util.List.of(record));
+ when(messageProvider.getQueueOffsets("inst-1",
"TopicA")).thenReturn(java.util.List.of(queue));
+ when(messageProvider.pullMessageAtOffset("inst-1", "TopicA",
"broker-a", 0, 7L))
+ .thenReturn(record);
+
+ assertThat(provider.queryMessageByUniqueKey("inst-1", "TopicA",
"uniq-1", 100L, 200L))
+ .containsExactly(record);
+ assertThat(provider.getQueueOffsets("inst-1",
"TopicA")).containsExactly(queue);
+ assertThat(provider.pullMessageAtOffset("inst-1", "TopicA",
"broker-a", 0, 7L))
+ .isSameAs(record);
+
+ verify(messageProvider).queryMessageByUniqueKey("inst-1", "TopicA",
"uniq-1", 100L, 200L);
+ verify(messageProvider).getQueueOffsets("inst-1", "TopicA");
+ verify(messageProvider).pullMessageAtOffset("inst-1", "TopicA",
"broker-a", 0, 7L);
+ }
+
@Test
void countTopicsShouldDelegateToRepositoryTest() {
InstanceVO instance = InstanceVO.builder().name("inst-1").build();
