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 245699f92 feat(studio): support cloud direct message consumption
(#4455)
245699f92 is described below
commit 245699f92dcc3df40c67b34a39863d88b13e3b3d
Author: aias00 <[email protected]>
AuthorDate: Sat Sep 26 03:43:35 2026 -0700
feat(studio): support cloud direct message consumption (#4455)
fix(studio): address cloud direct consume review
Signed-off-by: liuhy <[email protected]>
---
docs/api-spec.md | 10 +-
.../studio/instance/message/MessageService.java | 20 +++-
.../studio/provider/InstanceCapability.java | 1 +
.../provider/alibaba/AliyunInstanceProvider.java | 46 ++++++++
.../provider/apache/ApacheInstanceProvider.java | 1 +
.../provider/tencent/TencentInstanceProvider.java | 36 ++++++
.../instance/message/MessageServiceTest.java | 35 ++++++
.../alibaba/AliyunInstanceProviderTest.java | 86 +++++++++++++++
.../apache/ApacheInstanceProviderTest.java | 1 +
.../tencent/TencentInstanceProviderTest.java | 51 +++++++++
web/src/api/instance.ts | 1 +
web/src/i18n/translations.ts | 16 +++
web/src/mock/instances.ts | 31 +++++-
.../pages/instance/__tests__/MessagePage.test.tsx | 10 ++
.../__tests__/MessagePageAsyncState.test.tsx | 121 +++++++++++++++++++++
web/src/pages/instance/message.tsx | 73 +++++++++++--
web/src/services/instanceService.test.ts | 35 +++++-
web/src/services/instanceService.ts | 21 +---
18 files changed, 558 insertions(+), 37 deletions(-)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index 0f8dbc2f9..4763e67bd 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -1503,11 +1503,11 @@ POST /api/messages/direct-consume
| 字段 | 类型 | 说明 |
|------|------|------|
-| `consumeResult` | `string` | Broker 返回的直接消费结果 |
-| `remark` | `string` | Broker 返回的说明 |
-| `spentTimeMillis` | `number` | Broker 执行耗时（毫秒） |
-| `order` | `boolean` | 是否为顺序消费 |
-| `autoCommit` | `boolean` | 客户端是否启用自动提交 |
+| `consumeResult` | `string` | 直消费结果；Apache 路径为 Broker 的
`ConsumeMessageDirectlyResult` 名（`CR_SUCCESS` 等），Aliyun 为
`CR_SUCCESS`/`CR_FAILED`，Tencent 因 OpenAPI 不回传消费结果固定为 `REQUEST_ACCEPTED`，实际语义见
`remark` |
+| `remark` | `string` | 结果说明；Apache 路径取自 Broker，云厂商路径取自云 OpenAPI 应答（含
`requestId`） |
+| `spentTimeMillis` | `number` | 执行耗时（毫秒）；云厂商路径为 Studio 侧测得的 OpenAPI 调用耗时 |
+| `order` | `boolean` | 是否为顺序消费（云厂商路径固定 `false`） |
+| `autoCommit` | `boolean` | 客户端是否启用自动提交（云厂商路径固定 `false`） |
---
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 cb85a5528..ead1d11cc 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
@@ -23,6 +23,8 @@ import
org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource;
import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind;
+import org.apache.rocketmq.studio.provider.InstanceCapability;
+import org.apache.rocketmq.studio.provider.InstanceProvider;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.springframework.stereotype.Service;
@@ -162,10 +164,10 @@ public class MessageService {
DirectConsumeMessageResultVO result = apache
? ownershipGuard.withOwned(instance, resources, () ->
providerRegistry.byInstanceId(request.getInstanceId())
- .map(provider ->
provider.consumeMessageDirectly(request))
+ .map(provider ->
consumeDirectlyThrough(provider, request))
.orElseGet(() ->
messageProvider.consumeMessageDirectly(request)))
: providerRegistry.byInstanceId(request.getInstanceId())
- .map(provider ->
provider.consumeMessageDirectly(request))
+ .map(provider -> consumeDirectlyThrough(provider,
request))
.orElseGet(() ->
messageProvider.consumeMessageDirectly(request));
recordDirectConsumeAudit(request, detail + ", result=" +
result.getConsumeResult(),
auditResult(result.getConsumeResult()), null);
@@ -176,6 +178,20 @@ public class MessageService {
}
}
+ /**
+ * Gate the registered-provider path on the advertised capability so a
vendor without a
+ * direct-consume API answers 501 (see {@code GlobalExceptionHandler})
instead of letting the
+ * call fall through to an implementation that cannot honour it.
+ */
+ private static DirectConsumeMessageResultVO
consumeDirectlyThrough(InstanceProvider provider,
+
DirectConsumeMessageDTO request) {
+ if
(!provider.capabilities().contains(InstanceCapability.DIRECT_MESSAGE_CONSUME)) {
+ throw new UnsupportedOperationException(
+ "Direct message consumption is not supported by this
instance");
+ }
+ return provider.consumeMessageDirectly(request);
+ }
+
/** consumeResult mirrors the broker-side CMResult enum, where only
CR_SUCCESS means consumed. */
private static String auditResult(String consumeResult) {
return "CR_SUCCESS".equals(consumeResult) ? "SUCCESS" : "FAILED";
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceCapability.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceCapability.java
index a98a5433d..12fa74694 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceCapability.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceCapability.java
@@ -25,6 +25,7 @@ public enum InstanceCapability {
MESSAGE_QUERY,
MESSAGE_TRACE,
MESSAGE_SEND,
+ DIRECT_MESSAGE_CONSUME,
ACL_MANAGEMENT,
DLQ_MANAGEMENT
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
index da5a502ff..d1c10c0c3 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
@@ -46,6 +46,9 @@ import
com.aliyun.sdk.service.rocketmq20220801.models.UpdateTopicRequest;
import com.aliyun.sdk.service.rocketmq20220801.models.VerifySendMessageRequest;
import
com.aliyun.sdk.service.rocketmq20220801.models.VerifySendMessageResponse;
import
com.aliyun.sdk.service.rocketmq20220801.models.VerifySendMessageResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.VerifyConsumeMessageRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.VerifyConsumeMessageResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.VerifyConsumeMessageResponseBody;
import org.springframework.util.StringUtils;
import org.apache.rocketmq.studio.common.domain.PageResult;
@@ -59,6 +62,8 @@ import
org.apache.rocketmq.studio.instance.group.QueueProgressVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.apache.rocketmq.studio.instance.message.MessageQueryResult;
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.TraceRecordVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
@@ -74,6 +79,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
/**
* Aliyun RocketMQ 5.x implementation of the instance-scoped operations SPI,
backed by the
@@ -109,6 +115,7 @@ public class AliyunInstanceProvider implements
InstanceProvider {
InstanceCapability.MESSAGE_QUERY,
InstanceCapability.MESSAGE_TRACE,
InstanceCapability.MESSAGE_SEND,
+ InstanceCapability.DIRECT_MESSAGE_CONSUME,
InstanceCapability.ACL_MANAGEMENT);
}
@@ -556,6 +563,35 @@ public class AliyunInstanceProvider implements
InstanceProvider {
return mayBeTruncated ? MessageQueryResult.truncated(records) :
MessageQueryResult.complete(records);
}
+ @Override
+ public DirectConsumeMessageResultVO
consumeMessageDirectly(DirectConsumeMessageDTO request) {
+ Context ctx = resolve(request.getInstanceId());
+ VerifyConsumeMessageRequest verifyRequest =
VerifyConsumeMessageRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .topicName(request.getTopic())
+ .messageId(request.getMsgId())
+ .consumerGroupId(request.getConsumerGroup())
+ .clientId(request.getClientId())
+ .build();
+ long startedAt = System.nanoTime();
+ VerifyConsumeMessageResponse response =
clientFactory.call(ctx.credentialId(), ctx.regionId(),
+ client -> client.verifyConsumeMessage(verifyRequest));
+ long spentTimeMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime()
- startedAt);
+ VerifyConsumeMessageResponseBody body = response == null ? null :
response.getBody();
+ if (body == null) {
+ throw new BusinessException(502, "Aliyun direct consume returned
an empty response");
+ }
+ boolean success = Boolean.TRUE.equals(body.getSuccess()) &&
Boolean.TRUE.equals(body.getData());
+ String message = StringUtils.hasText(body.getMessage()) ?
body.getMessage() : body.getDynamicMessage();
+ return DirectConsumeMessageResultVO.builder()
+ .consumeResult(success ? "CR_SUCCESS" : "CR_FAILED")
+ .remark(formatCloudRemark(message, body.getRequestId()))
+ .spentTimeMillis(spentTimeMillis)
+ .order(false)
+ .autoCommit(false)
+ .build();
+ }
+
@Override
public TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic) {
Context ctx = resolve(instanceId);
@@ -581,6 +617,16 @@ public class AliyunInstanceProvider implements
InstanceProvider {
.build();
}
+ private static String formatCloudRemark(String message, String requestId) {
+ if (StringUtils.hasText(message) && StringUtils.hasText(requestId)) {
+ return message + " (requestId=" + requestId + ")";
+ }
+ if (StringUtils.hasText(message)) {
+ return message;
+ }
+ return StringUtils.hasText(requestId) ? "requestId=" + requestId :
null;
+ }
+
private Context resolve(String instanceId) {
if (!StringUtils.hasText(instanceId)) {
throw new BusinessException(400, "instanceId is required");
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 f2c5572d6..02e834a57 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
@@ -70,6 +70,7 @@ public class ApacheInstanceProvider implements
InstanceProvider {
InstanceCapability.MESSAGE_QUERY,
InstanceCapability.MESSAGE_TRACE,
InstanceCapability.MESSAGE_SEND,
+ InstanceCapability.DIRECT_MESSAGE_CONSUME,
InstanceCapability.ACL_MANAGEMENT,
InstanceCapability.DLQ_MANAGEMENT);
}
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 93a6f0752..e17d26b12 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
@@ -46,6 +46,8 @@ import
com.tencentcloudapi.trocket.v20230308.models.SendMessageRequest;
import com.tencentcloudapi.trocket.v20230308.models.SendMessageResponse;
import com.tencentcloudapi.trocket.v20230308.models.SubscriptionData;
import com.tencentcloudapi.trocket.v20230308.models.TopicItem;
+import
com.tencentcloudapi.trocket.v20230308.models.VerifyMessageConsumptionRequest;
+import
com.tencentcloudapi.trocket.v20230308.models.VerifyMessageConsumptionResponse;
import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
@@ -62,6 +64,8 @@ import
org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.apache.rocketmq.studio.instance.message.ConsumerStatusVO;
+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.MessageQueryResult;
import org.apache.rocketmq.studio.instance.message.TraceNodeVO;
@@ -94,6 +98,7 @@ import java.util.Locale;
import java.util.Set;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
/**
* Tencent Cloud TDMQ RocketMQ 5.x topic operations backed by Trocket
v20230308 OpenAPI.
@@ -150,6 +155,7 @@ public class TencentInstanceProvider implements
InstanceProvider {
InstanceCapability.MESSAGE_QUERY,
InstanceCapability.MESSAGE_TRACE,
InstanceCapability.MESSAGE_SEND,
+ InstanceCapability.DIRECT_MESSAGE_CONSUME,
InstanceCapability.ACL_MANAGEMENT);
}
@@ -686,6 +692,36 @@ public class TencentInstanceProvider implements
InstanceProvider {
return mayBeTruncated ? MessageQueryResult.truncated(result) :
MessageQueryResult.complete(result);
}
+ @Override
+ public DirectConsumeMessageResultVO
consumeMessageDirectly(DirectConsumeMessageDTO request) {
+ Context context = resolve(request.getInstanceId());
+ VerifyMessageConsumptionRequest verifyRequest = new
VerifyMessageConsumptionRequest();
+ verifyRequest.setInstanceId(context.cloudInstanceId());
+ verifyRequest.setTopic(request.getTopic());
+ verifyRequest.setMsgId(request.getMsgId());
+ verifyRequest.setConsumerGroup(request.getConsumerGroup());
+ verifyRequest.setClientId(request.getClientId());
+ long startedAt = System.nanoTime();
+ VerifyMessageConsumptionResponse response = clientFactory.call(
+ context.credentialId(), context.regionId(), client ->
client.VerifyMessageConsumption(verifyRequest));
+ long spentTimeMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime()
- startedAt);
+ if (response == null) {
+ throw new BusinessException(502, "Tencent direct consume returned
an empty response");
+ }
+ String requestId = response.getRequestId();
+ String acknowledgement = "Tencent accepted the verification request; "
+ + "the consumption outcome is not returned";
+ return DirectConsumeMessageResultVO.builder()
+ .consumeResult("REQUEST_ACCEPTED")
+ .remark(StringUtils.hasText(requestId)
+ ? acknowledgement + " (requestId=" + requestId + ")"
+ : acknowledgement)
+ .spentTimeMillis(spentTimeMillis)
+ .order(false)
+ .autoCommit(false)
+ .build();
+ }
+
@Override
public TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic) {
Context context = resolve(instanceId);
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 260123acd..b5921d1dd 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
@@ -22,17 +22,20 @@ import
org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
import static org.mockito.ArgumentMatchers.any;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
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 java.util.List;
import java.util.Optional;
+import java.util.Set;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.when;
class MessageServiceTest {
@@ -175,6 +178,7 @@ class MessageServiceTest {
DirectConsumeMessageResultVO expected =
DirectConsumeMessageResultVO.builder()
.consumeResult("CR_SUCCESS").build();
when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
+
when(provider.capabilities()).thenReturn(Set.of(InstanceCapability.DIRECT_MESSAGE_CONSUME));
when(provider.consumeMessageDirectly(request)).thenReturn(expected);
MessageService service = new MessageService(fallback, registry,
mock(QueryHistoryService.class), audit, ownershipGuard());
@@ -199,6 +203,7 @@ class MessageServiceTest {
request.setConsumerGroup("billing");
request.setClientId("client-a");
when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
+
when(provider.capabilities()).thenReturn(Set.of(InstanceCapability.DIRECT_MESSAGE_CONSUME));
when(provider.consumeMessageDirectly(request)).thenReturn(DirectConsumeMessageResultVO.builder()
.consumeResult("CR_ROLLBACK").build());
MessageService service = new MessageService(fallback, registry,
mock(QueryHistoryService.class), audit, ownershipGuard());
@@ -224,6 +229,7 @@ class MessageServiceTest {
request.setConsumerGroup("billing");
request.setClientId("client-a");
when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
+
when(provider.capabilities()).thenReturn(Set.of(InstanceCapability.DIRECT_MESSAGE_CONSUME));
when(provider.consumeMessageDirectly(request)).thenThrow(new
IllegalStateException("client offline"));
MessageService service = new MessageService(fallback, registry,
mock(QueryHistoryService.class), audit, ownershipGuard());
@@ -238,6 +244,35 @@ class MessageServiceTest {
verifyNoInteractions(fallback);
}
+ @Test
+ void
rejectsDirectConsumptionWhenSelectedProviderDoesNotAdvertiseCapabilityTest() {
+ MessageProvider fallback = mock(MessageProvider.class);
+ InstanceProvider provider = mock(InstanceProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ OperationAuditService audit = mock(OperationAuditService.class);
+ DirectConsumeMessageDTO request = new DirectConsumeMessageDTO();
+ request.setInstanceId("cloud-instance");
+ request.setTopic("orders");
+ request.setMsgId("msg-1");
+ request.setConsumerGroup("billing");
+ request.setClientId("client-a");
+
when(registry.byInstanceId("cloud-instance")).thenReturn(Optional.of(provider));
+
when(provider.capabilities()).thenReturn(Set.of(InstanceCapability.MESSAGE_QUERY));
+ MessageService service = new MessageService(
+ fallback, registry, mock(QueryHistoryService.class), audit,
ownershipGuard());
+
+ assertThatThrownBy(() -> service.consumeMessageDirectly(request))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Direct message consumption is not supported by
this instance");
+
+ verify(provider, never()).consumeMessageDirectly(request);
+ verifyNoInteractions(fallback);
+
verify(audit).record(org.mockito.ArgumentMatchers.eq("DIRECT_CONSUME_MESSAGE"),
+ org.mockito.ArgumentMatchers.eq("MESSAGE"),
org.mockito.ArgumentMatchers.eq("msg-1"),
+ org.mockito.ArgumentMatchers.eq("cloud-instance"),
org.mockito.ArgumentMatchers.contains("billing"),
+ org.mockito.ArgumentMatchers.eq("FAILED"),
org.mockito.ArgumentMatchers.contains("not supported"));
+ }
+
@Test
void rejectsOverflowingTopicQueryWindowBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
index a3a167f92..875512d48 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
@@ -40,6 +40,9 @@ import
com.aliyun.sdk.service.rocketmq20220801.models.ResetConsumeOffsetResponse
import com.aliyun.sdk.service.rocketmq20220801.models.VerifySendMessageRequest;
import
com.aliyun.sdk.service.rocketmq20220801.models.VerifySendMessageResponse;
import
com.aliyun.sdk.service.rocketmq20220801.models.VerifySendMessageResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.VerifyConsumeMessageRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.VerifyConsumeMessageResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.VerifyConsumeMessageResponseBody;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
@@ -53,6 +56,8 @@ import
org.apache.rocketmq.studio.instance.group.QueueProgressVO;
import org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
import org.apache.rocketmq.studio.instance.message.MessageQueryResult;
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.TraceNodeVO;
import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
@@ -114,6 +119,7 @@ class AliyunInstanceProviderTest {
.contains(InstanceCapability.TOPIC_MANAGEMENT,
InstanceCapability.MESSAGE_QUERY,
InstanceCapability.MESSAGE_SEND,
+ InstanceCapability.DIRECT_MESSAGE_CONSUME,
InstanceCapability.ACL_MANAGEMENT)
.doesNotContain(InstanceCapability.DLQ_MANAGEMENT);
}
@@ -163,6 +169,76 @@ class AliyunInstanceProviderTest {
.hasMessageContaining("send denied");
}
+ @Test
+ void consumeMessageDirectlyShouldCallAliyunVerifyApiTest() {
+ stubInstance();
+ stubCallThrough();
+ VerifyConsumeMessageResponse response =
VerifyConsumeMessageResponse.create().toBuilder()
+ .statusCode(200)
+ .body(VerifyConsumeMessageResponseBody.builder()
+ .success(true)
+ .data(true)
+ .message("accepted")
+ .requestId("aliyun-request-1")
+ .build())
+ .build();
+ when(asyncClient.verifyConsumeMessage(any()))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ DirectConsumeMessageResultVO result =
provider.consumeMessageDirectly(directConsumeRequest());
+
+ ArgumentCaptor<VerifyConsumeMessageRequest> captor =
+ ArgumentCaptor.forClass(VerifyConsumeMessageRequest.class);
+ verify(asyncClient).verifyConsumeMessage(captor.capture());
+ VerifyConsumeMessageRequest request = captor.getValue();
+ assertThat(request.getInstanceId()).isEqualTo(CLOUD_INSTANCE_ID);
+ assertThat(request.getTopicName()).isEqualTo("orders");
+ assertThat(request.getMessageId()).isEqualTo("msg-1");
+ assertThat(request.getConsumerGroupId()).isEqualTo("billing");
+ assertThat(request.getClientId()).isEqualTo("client-a");
+ assertThat(result.getConsumeResult()).isEqualTo("CR_SUCCESS");
+ assertThat(result.getRemark()).contains("accepted",
"aliyun-request-1");
+ assertThat(result.getSpentTimeMillis()).isNotNegative();
+ assertThat(result.isOrder()).isFalse();
+ assertThat(result.isAutoCommit()).isFalse();
+ }
+
+ @Test
+ void consumeMessageDirectlyShouldPreserveAliyunBusinessFailureTest() {
+ stubInstance();
+ stubCallThrough();
+ VerifyConsumeMessageResponse response =
VerifyConsumeMessageResponse.create().toBuilder()
+ .statusCode(200)
+ .body(VerifyConsumeMessageResponseBody.builder()
+ .success(true)
+ .data(false)
+ .message("client offline")
+ .requestId("aliyun-request-2")
+ .build())
+ .build();
+ when(asyncClient.verifyConsumeMessage(any()))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ DirectConsumeMessageResultVO result =
provider.consumeMessageDirectly(directConsumeRequest());
+
+ assertThat(result.getConsumeResult()).isEqualTo("CR_FAILED");
+ assertThat(result.getRemark()).contains("client offline",
"aliyun-request-2");
+ }
+
+ @Test
+ void consumeMessageDirectlyShouldRejectEmptyAliyunResponseTest() {
+ stubInstance();
+ stubCallThrough();
+ when(asyncClient.verifyConsumeMessage(any()))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ assertThatThrownBy(() ->
provider.consumeMessageDirectly(directConsumeRequest()))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Aliyun direct consume returned an empty response")
+ .extracting("code")
+ .isEqualTo(502);
+ }
+
@Test
void listTopicsShouldMapMessageTypeAndFilterTest() {
stubInstance();
@@ -888,6 +964,16 @@ class AliyunInstanceProviderTest {
});
}
+ private DirectConsumeMessageDTO directConsumeRequest() {
+ DirectConsumeMessageDTO request = new DirectConsumeMessageDTO();
+ request.setInstanceId(STUDIO_INSTANCE_ID);
+ request.setTopic("orders");
+ request.setMsgId("msg-1");
+ request.setConsumerGroup("billing");
+ request.setClientId("client-a");
+ return request;
+ }
+
private static ListTopicsResponse
topicsResponse(ListTopicsResponseBody.List... rows) {
return topicsResponse((long) rows.length, 1L, 100L, rows);
}
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 2c782966c..51bb098ef 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
@@ -111,6 +111,7 @@ class ApacheInstanceProviderTest {
InstanceCapability.MESSAGE_QUERY,
InstanceCapability.MESSAGE_TRACE,
InstanceCapability.MESSAGE_SEND,
+ InstanceCapability.DIRECT_MESSAGE_CONSUME,
InstanceCapability.ACL_MANAGEMENT,
InstanceCapability.DLQ_MANAGEMENT);
}
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 91393f953..4ee5af80b 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
@@ -44,6 +44,8 @@ import
com.tencentcloudapi.trocket.v20230308.models.SendMessageRequest;
import com.tencentcloudapi.trocket.v20230308.models.SendMessageResponse;
import com.tencentcloudapi.trocket.v20230308.models.SubscriptionData;
import com.tencentcloudapi.trocket.v20230308.models.TopicItem;
+import
com.tencentcloudapi.trocket.v20230308.models.VerifyMessageConsumptionRequest;
+import
com.tencentcloudapi.trocket.v20230308.models.VerifyMessageConsumptionResponse;
import com.tencentcloudapi.trocket.v20230308.TrocketClient;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
@@ -60,6 +62,8 @@ import
org.apache.rocketmq.studio.instance.group.ResetConsumerOffsetPreviewVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
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.TraceNodeVO;
import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
import org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
@@ -131,6 +135,7 @@ class TencentInstanceProviderTest {
.contains(InstanceCapability.TOPIC_MANAGEMENT,
InstanceCapability.MESSAGE_QUERY,
InstanceCapability.MESSAGE_SEND,
+ InstanceCapability.DIRECT_MESSAGE_CONSUME,
InstanceCapability.ACL_MANAGEMENT)
.doesNotContain(InstanceCapability.DLQ_MANAGEMENT);
}
@@ -174,6 +179,42 @@ class TencentInstanceProviderTest {
.hasMessageContaining("no message id");
}
+ @Test
+ void consumeMessageDirectlyShouldCallTencentVerifyApiTest() throws
Exception {
+ VerifyMessageConsumptionResponse response = new
VerifyMessageConsumptionResponse();
+ response.setRequestId("tencent-request-1");
+ when(client.VerifyMessageConsumption(any())).thenReturn(response);
+
+ DirectConsumeMessageResultVO result =
provider.consumeMessageDirectly(directConsumeRequest());
+
+ ArgumentCaptor<VerifyMessageConsumptionRequest> captor =
+ ArgumentCaptor.forClass(VerifyMessageConsumptionRequest.class);
+ verify(client).VerifyMessageConsumption(captor.capture());
+ VerifyMessageConsumptionRequest request = captor.getValue();
+ assertThat(request.getInstanceId()).isEqualTo(CLOUD_INSTANCE_ID);
+ assertThat(request.getTopic()).isEqualTo("orders");
+ assertThat(request.getMsgId()).isEqualTo("msg-1");
+ assertThat(request.getConsumerGroup()).isEqualTo("billing");
+ assertThat(request.getClientId()).isEqualTo("client-a");
+ assertThat(result.getConsumeResult()).isEqualTo("REQUEST_ACCEPTED");
+ assertThat(result.getRemark())
+ .contains("verification request", "outcome is not returned",
"tencent-request-1");
+ assertThat(result.getSpentTimeMillis()).isNotNegative();
+ assertThat(result.isOrder()).isFalse();
+ assertThat(result.isAutoCommit()).isFalse();
+ }
+
+ @Test
+ void consumeMessageDirectlyShouldRejectEmptyTencentResponseTest() throws
Exception {
+ when(client.VerifyMessageConsumption(any())).thenReturn(null);
+
+ assertThatThrownBy(() ->
provider.consumeMessageDirectly(directConsumeRequest()))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Tencent direct consume returned an empty
response")
+ .extracting("code")
+ .isEqualTo(502);
+ }
+
@Test
void countTopicsShouldClampOversizedTotals() throws Exception {
DescribeTopicListResponse response = new DescribeTopicListResponse();
@@ -1194,4 +1235,14 @@ class TencentInstanceProviderTest {
assertThat(trace.getConsumerStatus().get(0).getDeliveryStatus())
.isEqualTo(DeliveryStatus.pending);
}
+
+ private DirectConsumeMessageDTO directConsumeRequest() {
+ DirectConsumeMessageDTO request = new DirectConsumeMessageDTO();
+ request.setInstanceId(STUDIO_INSTANCE_ID);
+ request.setTopic("orders");
+ request.setMsgId("msg-1");
+ request.setConsumerGroup("billing");
+ request.setClientId("client-a");
+ return request;
+ }
}
diff --git a/web/src/api/instance.ts b/web/src/api/instance.ts
index eab47239b..9bf6f31db 100644
--- a/web/src/api/instance.ts
+++ b/web/src/api/instance.ts
@@ -26,6 +26,7 @@ export type InstanceCapability =
| 'MESSAGE_QUERY'
| 'MESSAGE_TRACE'
| 'MESSAGE_SEND'
+ | 'DIRECT_MESSAGE_CONSUME'
| 'ACL_MANAGEMENT'
| 'DLQ_MANAGEMENT';
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index b47668d20..2c695edae 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -3109,6 +3109,22 @@ const translations: Record<string, Record<Lang, string>>
= {
'message.batchResend': { zh: '批量重发', en: 'Batch Resend' },
'message.batchExport': { zh: '批量导出', en: 'Batch Export' },
'message.noMatchResult': { zh: '没有查到符合条件的结果', en: 'No matching results' },
+ 'message.directConsumeSelectInstance': {
+ zh: '请先选择实例',
+ en: 'Select an instance first',
+ },
+ 'message.directConsumeCapabilityLoading': {
+ zh: '正在确认当前实例是否支持直接消费',
+ en: 'Checking whether this instance supports direct consumption',
+ },
+ 'message.directConsumeUnsupported': {
+ zh: '当前实例不支持直接消费',
+ en: 'This instance does not support direct consumption',
+ },
+ 'message.directConsumeCapabilityUnavailable': {
+ zh: '无法获取实例能力，直接消费暂不可用',
+ en: 'Instance capabilities are unavailable; direct consumption is
disabled',
+ },
// ─── DLQ (detailed) ───
'dlq.subtitle': {
diff --git a/web/src/mock/instances.ts b/web/src/mock/instances.ts
index caf315d4f..e1b1ea7fa 100644
--- a/web/src/mock/instances.ts
+++ b/web/src/mock/instances.ts
@@ -15,7 +15,28 @@
* limitations under the License.
*/
-import type { Instance } from '../api/instance';
+import type { Instance, InstanceCapability } from '../api/instance';
+
+const MOCK_APACHE_PROVIDER_CAPABILITIES = [
+ 'TOPIC_MANAGEMENT',
+ 'CONSUMER_GROUP_MANAGEMENT',
+ 'MESSAGE_QUERY',
+ 'MESSAGE_TRACE',
+ 'MESSAGE_SEND',
+ 'DIRECT_MESSAGE_CONSUME',
+ 'ACL_MANAGEMENT',
+ 'DLQ_MANAGEMENT',
+] as const satisfies readonly InstanceCapability[];
+
+const MOCK_CLOUD_PROVIDER_CAPABILITIES = [
+ 'TOPIC_MANAGEMENT',
+ 'CONSUMER_GROUP_MANAGEMENT',
+ 'MESSAGE_QUERY',
+ 'MESSAGE_TRACE',
+ 'MESSAGE_SEND',
+ 'DIRECT_MESSAGE_CONSUME',
+ 'ACL_MANAGEMENT',
+] as const satisfies readonly InstanceCapability[];
export const mockInstances: Instance[] = [
{
@@ -74,3 +95,11 @@ export const mockInstances: Instance[] = [
gmtModified: '2026-08-03 10:00:00',
},
];
+
+export function getMockInstanceCapabilities(instance: Instance):
InstanceCapability[] {
+ const capabilities =
+ instance.vendor === 'ALIYUN' || instance.vendor === 'TENCENT'
+ ? MOCK_CLOUD_PROVIDER_CAPABILITIES
+ : MOCK_APACHE_PROVIDER_CAPABILITIES;
+ return [...capabilities];
+}
diff --git a/web/src/pages/instance/__tests__/MessagePage.test.tsx
b/web/src/pages/instance/__tests__/MessagePage.test.tsx
index 8aa82669b..ddbcc8f30 100644
--- a/web/src/pages/instance/__tests__/MessagePage.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePage.test.tsx
@@ -37,6 +37,9 @@ const topicServiceMocks = vi.hoisted(() => ({
const instanceFilterMocks = vi.hoisted(() => ({
useInstanceFilter: vi.fn(),
}));
+const instanceServiceMocks = vi.hoisted(() => ({
+ getInstanceCapabilities: vi.fn(),
+}));
vi.mock('../../../services/messageService', () => ({
...messageServiceMocks,
@@ -52,6 +55,7 @@ vi.mock('../../../services/messageService', () => ({
vi.mock('../../../hooks/useInstanceFilter', () => instanceFilterMocks);
vi.mock('../../../services/instanceService', () => ({
+ getInstanceCapabilities: instanceServiceMocks.getInstanceCapabilities,
listInstances: vi.fn().mockResolvedValue([]),
}));
vi.mock('../../../services/topicService', () => topicServiceMocks);
@@ -122,6 +126,12 @@ describe('Message page query history', () => {
selectInstance: vi.fn(),
instanceOptions: [{ value: 1, label: 'Instance A' }],
});
+
instanceServiceMocks.getInstanceCapabilities.mockReset().mockResolvedValue({
+ instanceId: '1',
+ vendor: 'APACHE',
+ accessType: 'DIRECT',
+ capabilities: ['DIRECT_MESSAGE_CONSUME'],
+ });
});
afterEach(() => {
diff --git a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
index 25edbcb94..64d7139ab 100644
--- a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
@@ -33,6 +33,9 @@ const serviceMocks = vi.hoisted(() => ({
const instanceFilterMocks = vi.hoisted(() => ({
useInstanceFilter: vi.fn(),
}));
+const instanceServiceMocks = vi.hoisted(() => ({
+ getInstanceCapabilities: vi.fn(),
+}));
const historyMocks = vi.hoisted(() => ({
getQueryHistorySummary: vi.fn(),
listMessageQueryHistory: vi.fn(),
@@ -59,6 +62,7 @@ vi.mock('../../../hooks/useInstanceFilter', () =>
instanceFilterMocks);
vi.mock('../../../api/messageHistory', () => historyMocks);
vi.mock('../../../services/instanceService', () => ({
+ getInstanceCapabilities: instanceServiceMocks.getInstanceCapabilities,
listInstances: vi.fn().mockResolvedValue([]),
}));
vi.mock('../../../services/topicService', () => ({
@@ -165,6 +169,12 @@ describe('MessagePage async request ownership', () => {
selectInstance: vi.fn(),
instanceOptions: [{ value: 1, label: 'Instance A' }],
});
+ instanceServiceMocks.getInstanceCapabilities.mockResolvedValue({
+ instanceId: '1',
+ vendor: 'APACHE',
+ accessType: 'DIRECT',
+ capabilities: ['DIRECT_MESSAGE_CONSUME'],
+ });
vi.spyOn(message, 'success').mockImplementation(vi.fn());
});
@@ -370,6 +380,117 @@ describe('MessagePage async request ownership', () => {
expect(serviceMocks.consumeMessageDirectly).not.toHaveBeenCalled();
});
+
it('disablesDirectConsumeWithReasonWhenTheSelectedInstanceDoesNotAdvertiseItTest',
async () => {
+ instanceServiceMocks.getInstanceCapabilities.mockResolvedValue({
+ instanceId: '1',
+ vendor: 'ALIYUN',
+ accessType: 'CLOUD',
+ capabilities: ['MESSAGE_QUERY'],
+ });
+ serviceMocks.queryMessages.mockResolvedValue([createMessage('message-a')]);
+ const user = userEvent.setup();
+ renderPage();
+ await selectTopic(user);
+
+ await user.click(screen.getByRole('button', { name: /^search查询$/ }));
+ const row = await screen.findByRole('row', { name: /message-a/ });
+ await user.click(within(row).getByRole('button', { name: /详情/ }));
+
+ const dialog = await screen.findByRole('dialog', { name: '消息详情' });
+ await waitFor(() =>
+
expect(instanceServiceMocks.getInstanceCapabilities).toHaveBeenCalledWith(1),
+ );
+ const directConsumeButton = within(dialog).getByRole('button', { name:
/直接消费/ });
+ expect(directConsumeButton).toBeDisabled();
+ await user.hover(directConsumeButton.parentElement as HTMLElement);
+ expect(await screen.findByText('当前实例不支持直接消费')).toBeInTheDocument();
+ });
+
+ it('disablesDirectConsumeWithReasonWhenCapabilityLoadingFailsTest', async ()
=> {
+ instanceServiceMocks.getInstanceCapabilities.mockRejectedValue(
+ new Error('capability unavailable'),
+ );
+ serviceMocks.queryMessages.mockResolvedValue([createMessage('message-a')]);
+ const user = userEvent.setup();
+ renderPage();
+ await selectTopic(user);
+
+ await user.click(screen.getByRole('button', { name: /^search查询$/ }));
+ const row = await screen.findByRole('row', { name: /message-a/ });
+ await user.click(within(row).getByRole('button', { name: /详情/ }));
+
+ const dialog = await screen.findByRole('dialog', { name: '消息详情' });
+ await waitFor(() =>
+
expect(instanceServiceMocks.getInstanceCapabilities).toHaveBeenCalledWith(1),
+ );
+ const directConsumeButton = within(dialog).getByRole('button', { name:
/直接消费/ });
+ expect(directConsumeButton).toBeDisabled();
+ await user.hover(directConsumeButton.parentElement as HTMLElement);
+ expect(await screen.findByText('无法获取实例能力，直接消费暂不可用')).toBeInTheDocument();
+ });
+
+ it('ignoresStaleDirectConsumeCapabilityAfterInstanceChangeTest', async () =>
{
+ const first = createDeferred<{
+ instanceId: string;
+ vendor: string;
+ accessType: string;
+ capabilities: string[];
+ }>();
+ const second = createDeferred<{
+ instanceId: string;
+ vendor: string;
+ accessType: string;
+ capabilities: string[];
+ }>();
+ instanceServiceMocks.getInstanceCapabilities
+ .mockReturnValueOnce(first.promise)
+ .mockReturnValueOnce(second.promise);
+ let selectedInstanceId = 'instance-a';
+ instanceFilterMocks.useInstanceFilter.mockImplementation(() => ({
+ selectedInstanceId,
+ selectInstance: vi.fn(),
+ instanceOptions: [
+ { value: 'instance-a', label: 'Instance A' },
+ { value: 'instance-b', label: 'Instance B' },
+ ],
+ }));
+ const view = renderPage();
+ await waitFor(() =>
+
expect(instanceServiceMocks.getInstanceCapabilities).toHaveBeenCalledWith('instance-a'),
+ );
+
+ selectedInstanceId = 'instance-b';
+ view.rerender(<MessagePageWithProviders />);
+ await waitFor(() =>
+
expect(instanceServiceMocks.getInstanceCapabilities).toHaveBeenCalledWith('instance-b'),
+ );
+ await act(async () =>
+ second.resolve({
+ instanceId: 'instance-b',
+ vendor: 'ALIYUN',
+ accessType: 'CLOUD',
+ capabilities: ['MESSAGE_QUERY'],
+ }),
+ );
+ await act(async () =>
+ first.resolve({
+ instanceId: 'instance-a',
+ vendor: 'APACHE',
+ accessType: 'DIRECT',
+ capabilities: ['DIRECT_MESSAGE_CONSUME'],
+ }),
+ );
+ serviceMocks.queryMessages.mockResolvedValue([createMessage('message-b')]);
+ const user = userEvent.setup();
+ await selectTopic(user);
+ await user.click(screen.getByRole('button', { name: /^search查询$/ }));
+ const row = await screen.findByRole('row', { name: /message-b/ });
+ await user.click(within(row).getByRole('button', { name: /详情/ }));
+
+ const dialog = await screen.findByRole('dialog', { name: '消息详情' });
+ expect(within(dialog).getByRole('button', { name: /直接消费/
})).toBeDisabled();
+ });
+
it('queries trace by key with a custom trace topic from the trace tab',
async () => {
serviceMocks.queryMessages.mockResolvedValue([createMessage('message-a')]);
serviceMocks.getMessageTraceByKey.mockResolvedValue(createTrace('key-trace'));
diff --git a/web/src/pages/instance/message.tsx
b/web/src/pages/instance/message.tsx
index c57c09bae..197d403da 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -35,6 +35,7 @@ import {
Flex,
Progress,
Statistic,
+ Tooltip,
message,
} from 'antd';
import {
@@ -68,6 +69,7 @@ import {
queryMessagePage,
} from '../../services/messageService';
import { listTopics } from '../../services/topicService';
+import { getInstanceCapabilities } from '../../services/instanceService';
import { useInstanceFilter } from '../../hooks/useInstanceFilter';
import { downloadBlob } from '../../utils/download';
import { describeThrownMessage } from '../../utils/apiError';
@@ -89,6 +91,7 @@ const { RangePicker } = DatePicker;
/* ─── Constants ─── */
type QueryMode = 'topic' | 'key' | 'msgid' | 'queue';
+type DirectConsumeCapabilityStatus = 'loading' | 'supported' | 'unsupported' |
'unavailable';
const QUERY_OPTIONS = [
{ value: 'topic' as const },
@@ -400,6 +403,10 @@ const MessagePageContent = ({
const [directConsumeGroup, setDirectConsumeGroup] = useState('');
const [directConsumeClientId, setDirectConsumeClientId] = useState('');
const [directConsumeSubmitting, setDirectConsumeSubmitting] =
useState(false);
+ const [directConsumeCapability, setDirectConsumeCapability] = useState<{
+ instanceId: string;
+ status: DirectConsumeCapabilityStatus;
+ } | null>(null);
const queryGenerationRef = useRef(0);
// The query whose results the table currently shows. Pagination must re-run
this
// committed query, not whatever the form inputs hold at the moment a page
is clicked.
@@ -416,6 +423,49 @@ const MessagePageContent = ({
[],
);
+ useEffect(() => {
+ const instanceId = selectedInstanceId;
+ if (!instanceId) return;
+
+ let active = true;
+ void getInstanceCapabilities(instanceId)
+ .then((result) => {
+ if (active) {
+ setDirectConsumeOpen(false);
+ setDirectConsumeCapability({
+ instanceId,
+ status: result.capabilities.includes('DIRECT_MESSAGE_CONSUME')
+ ? 'supported'
+ : 'unsupported',
+ });
+ }
+ })
+ .catch(() => {
+ if (active) {
+ setDirectConsumeOpen(false);
+ setDirectConsumeCapability({ instanceId, status: 'unavailable' });
+ }
+ });
+ return () => {
+ active = false;
+ };
+ }, [selectedInstanceId]);
+
+ const directConsumeCapabilityStatus =
+ directConsumeCapability !== null && directConsumeCapability.instanceId ===
selectedInstanceId
+ ? directConsumeCapability.status
+ : 'loading';
+ const directConsumeSupported = directConsumeCapabilityStatus === 'supported';
+ const directConsumeDisabledReason = !selectedInstanceId
+ ? t('message.directConsumeSelectInstance')
+ : directConsumeCapabilityStatus === 'unavailable'
+ ? t('message.directConsumeCapabilityUnavailable')
+ : directConsumeCapabilityStatus === 'unsupported'
+ ? t('message.directConsumeUnsupported')
+ : directConsumeCapabilityStatus === 'loading'
+ ? t('message.directConsumeCapabilityLoading')
+ : undefined;
+
useEffect(() => {
writeMessageTraceTopic(selectedInstanceId, customTraceTopic);
}, [customTraceTopic, selectedInstanceId]);
@@ -668,6 +718,7 @@ const MessagePageContent = ({
};
const openDirectConsume = () => {
+ if (!directConsumeSupported) return;
setDirectConsumeGroup('');
setDirectConsumeClientId('');
setDirectConsumeOpen(true);
@@ -1260,14 +1311,18 @@ const MessagePageContent = ({
footer={
<Flex justify="flex-end" gap={8}>
<Button onClick={closeDetail}>{t('common.close')}</Button>
- <Button
- type="primary"
- icon={<SendOutlined />}
- disabled={!selectedInstanceId || !selectedMsg}
- onClick={openDirectConsume}
- >
- {t('messagePage.directConsume')}
- </Button>
+ <Tooltip title={directConsumeDisabledReason}>
+ <span>
+ <Button
+ type="primary"
+ icon={<SendOutlined />}
+ disabled={!directConsumeSupported || !selectedInstanceId ||
!selectedMsg}
+ onClick={openDirectConsume}
+ >
+ {t('messagePage.directConsume')}
+ </Button>
+ </span>
+ </Tooltip>
</Flex>
}
>
@@ -1276,7 +1331,7 @@ const MessagePageContent = ({
<Modal
title={t('messagePage.directConsumeTitle')}
- open={directConsumeOpen}
+ open={directConsumeOpen && directConsumeSupported}
onCancel={() => setDirectConsumeOpen(false)}
onOk={() => void handleDirectConsume()}
confirmLoading={directConsumeSubmitting}
diff --git a/web/src/services/instanceService.test.ts
b/web/src/services/instanceService.test.ts
index d19c3f29d..50b78ca92 100644
--- a/web/src/services/instanceService.test.ts
+++ b/web/src/services/instanceService.test.ts
@@ -123,11 +123,44 @@ describe('instanceService mock instances', () => {
const second = await getInstanceCapabilities('instance-direct-1');
- expect(second.capabilities).toContain('DLQ_MANAGEMENT');
+ expect(second.capabilities).toEqual(
+ expect.arrayContaining(['MESSAGE_SEND', 'DIRECT_MESSAGE_CONSUME',
'DLQ_MANAGEMENT']),
+ );
await expect(getInstanceCapabilities('missing-instance')).rejects.toThrow(
'Instance not found: missing-instance',
);
});
+
+ it('derives provider capabilities for newly created mock instances', async
() => {
+ const apacheName = 'mock-capability-apache';
+ const tencentName = 'mock-capability-tencent';
+ try {
+ await createInstance({
+ name: apacheName,
+ type: 'DIRECT',
+ endpoint: 'apache:9876',
+ vendor: 'APACHE',
+ });
+ await createInstance({
+ name: tencentName,
+ type: 'CLOUD',
+ endpoint: 'tencent:8080',
+ vendor: 'TENCENT',
+ });
+
+ await expect(getInstanceCapabilities(apacheName)).resolves.toEqual(
+ expect.objectContaining({
+ capabilities: expect.arrayContaining(['DIRECT_MESSAGE_CONSUME',
'DLQ_MANAGEMENT']),
+ }),
+ );
+ const tencent = await getInstanceCapabilities(tencentName);
+ expect(tencent.capabilities).toContain('DIRECT_MESSAGE_CONSUME');
+ expect(tencent.capabilities).not.toContain('DLQ_MANAGEMENT');
+ } finally {
+ await deleteInstance(apacheName);
+ await deleteInstance(tencentName);
+ }
+ });
});
describe('instanceService list request dedupe', () => {
diff --git a/web/src/services/instanceService.ts
b/web/src/services/instanceService.ts
index 7badf7c25..8e66addcb 100644
--- a/web/src/services/instanceService.ts
+++ b/web/src/services/instanceService.ts
@@ -8,7 +8,7 @@ import type {
UpdateInstanceRequest,
InstanceCapabilities,
} from '../api/instance';
-import { mockInstances } from '../mock/instances';
+import { getMockInstanceCapabilities, mockInstances } from '../mock/instances';
// Compile-time switch: mock or real API
@@ -21,23 +21,6 @@ function matchesType(instance: Instance, type?:
Instance['type']) {
return instance.type === type;
}
-const APACHE_CAPABILITIES: InstanceCapabilities['capabilities'] = [
- 'TOPIC_MANAGEMENT',
- 'CONSUMER_GROUP_MANAGEMENT',
- 'MESSAGE_QUERY',
- 'MESSAGE_TRACE',
- 'ACL_MANAGEMENT',
- 'DLQ_MANAGEMENT',
-];
-
-const CLOUD_CAPABILITIES: InstanceCapabilities['capabilities'] = [
- 'TOPIC_MANAGEMENT',
- 'CONSUMER_GROUP_MANAGEMENT',
- 'MESSAGE_QUERY',
- 'MESSAGE_TRACE',
- 'ACL_MANAGEMENT',
-];
-
const inflightListRequests = new Map<string, Promise<Instance[]>>();
// A mutation that just completed must not be shadowed by a list request that
@@ -87,7 +70,7 @@ export async function getInstanceCapabilities(instanceId:
string): Promise<Insta
instanceId: instance.name,
vendor,
accessType: instance.type,
- capabilities: [...(vendor === 'APACHE' ? APACHE_CAPABILITIES :
CLOUD_CAPABILITIES)],
+ capabilities: getMockInstanceCapabilities(instance),
};
}
