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 ccdb01599 fix(message): grade an offline direct-consume client as not
found (#4605)
ccdb01599 is described below
commit ccdb01599b1c2102ba158843c0b0b3565ebb124d
Author: 烤化の初雪 <[email protected]>
AuthorDate: Sat Sep 26 18:46:16 2026 +0800
fix(message): grade an offline direct-consume client as not found (#4605)
The broker answers SYSTEM_ERROR with "The Consumer <group> <client> not
online"
when the typed client id is not connected, and the raw MQClientException was
graded as a gateway failure, so POST /api/messages/direct-consume answered
502
with the raw broker text. The consumer stack endpoint already treats the
same
answer as the business state it is; direct consume now matches it and keeps
other broker failures as 502.
---
.../provider/apache/RocketMQMessageProvider.java | 26 +++++++++++++++++--
.../apache/RocketMQMessageProviderTest.java | 29 ++++++++++++++++++++++
2 files changed, 53 insertions(+), 2 deletions(-)
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 1f3e276eb..66e052932 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
@@ -573,8 +573,21 @@ public class RocketMQMessageProvider implements
MessageProvider {
||
!target.masters().contains(host.getAddress().getHostAddress() + ":" +
host.getPort())) {
throw new BusinessException(409, "Message's store
broker is not in the target master set");
}
- var result = ((DefaultMQAdminExt)
admin).consumeMessageDirectly(group, client,
- topic, message.getMsgId());
+
org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult result;
+ try {
+ result = ((DefaultMQAdminExt)
admin).consumeMessageDirectly(group, client,
+ topic, message.getMsgId());
+ } catch (Exception exception) {
+ String rootMessage = rootMessage(exception);
+ // The broker answers SYSTEM_ERROR with "The Consumer
<group> <client> not online"
+ // when the client is not connected. That is a
business state the operator can act
+ // on, not a gateway failure, so it is graded like the
consumer stack endpoint does.
+ if (rootMessage != null && rootMessage.contains("not
online")) {
+ throw new BusinessException(404, "Consumer client
is not online: " + client
+ + " (group " + group + ")");
+ }
+ throw new BusinessException(502, "Failed to consume
message directly: " + rootMessage);
+ }
return DirectConsumeMessageResultVO.builder()
.consumeResult(result.getConsumeResult() == null ?
"UNKNOWN" : result.getConsumeResult().name())
.remark(result.getRemark()).spentTimeMillis(result.getSpentTimeMills())
@@ -944,6 +957,15 @@ public class RocketMQMessageProvider implements
MessageProvider {
return index < fields.length ? fields[index] : "";
}
+ private static String rootMessage(Throwable error) {
+ Throwable current = error;
+ while (current.getCause() != null && current.getCause() != current) {
+ current = current.getCause();
+ }
+ String message = current.getMessage();
+ return message == null || message.isBlank() ?
current.getClass().getSimpleName() : message;
+ }
+
private static long parseLong(String value) {
try {
return Long.parseLong(value.trim());
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 c7961e5f6..87494e274 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
@@ -246,6 +246,35 @@ class RocketMQMessageProviderTest {
return request;
}
+ @Test
+ void directlyConsumesMessageReportsAnOfflineClientAsNotFoundTest() throws
Exception {
+ // The broker answers SYSTEM_ERROR with "The Consumer <group> <client>
not online" when
+ // the typed client id is not connected, which is the normal state for
a stale dialog.
+ MessageExt message = prepareDirectConsumption();
+ when(adminExt.consumeMessageDirectly("billing", "client-offline",
"orders", message.getMsgId()))
+ .thenThrow(new MQClientException(ResponseCode.SYSTEM_ERROR,
+ "The Consumer billing client-offline not online"));
+ DirectConsumeMessageDTO request = directRequest();
+ request.setClientId("client-offline");
+
+ assertThatThrownBy(() -> provider.consumeMessageDirectly(request))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("Consumer client is not online:
client-offline")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(404));
+ }
+
+ @Test
+ void directlyConsumesMessageKeepsOtherBrokerFailuresAsBadGatewayTest()
throws Exception {
+ MessageExt message = prepareDirectConsumption();
+ when(adminExt.consumeMessageDirectly("billing", "client-a", "orders",
message.getMsgId()))
+ .thenThrow(new MQClientException(ResponseCode.SYSTEM_ERROR,
"broker rejected the request"));
+
+ assertThatThrownBy(() ->
provider.consumeMessageDirectly(directRequest()))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("Failed to consume message directly")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(502));
+ }
+
@Test
void queryMessagesShouldRejectInvertedTimeRangeBeforeAdminLookup() throws
Exception {
assertThatThrownBy(() -> provider.queryMessages(
