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 1d4bbaf8e fix(message): treat MQClientException NO_MESSAGE key queries
as empty results (#3302)
1d4bbaf8e is described below
commit 1d4bbaf8e35a251101eb14f850ca9d254b6c534f
Author: 烤化の初雪 <[email protected]>
AuthorDate: Mon Sep 7 17:28:54 2026 +0800
fix(message): treat MQClientException NO_MESSAGE key queries as empty
results (#3302)
MQAdminImpl.queryMessage throws MQClientException(ResponseCode.NO_MESSAGE,
"query message by key finished, but no message.") instead of returning an
empty QueryResult when the key matches nothing. queryByKey surfaced that as
a 502 gateway error, so searching a message key that simply has no matches -
a normal outcome - failed. The trace lookups hit the same client path:
getMessageTrace returned 502 for any message without trace data (trace
disabled on the producer or expired records), contradicting the provider's
own exception-grading convention that only grades TOPIC_NOT_EXIST.
Grade NO_MESSAGE alongside TOPIC_NOT_EXIST via a shared response-code check:
key queries return an empty list and both trace paths return an empty trace,
while genuine broker failures still surface as 502. This matches the
expected behavior recorded in issues #1161/#1275: a completed query with no
matching records is an empty result, not an error.
Co-authored-by: unbridled-41
<[email protected]>
---
.../studio/common/util/MqResponseCodes.java | 64 ++++++++++++++++++++
.../provider/apache/RocketMQMessageProvider.java | 36 +++++------
.../studio/common/util/MqResponseCodesTest.java | 69 ++++++++++++++++++++++
.../apache/RocketMQMessageProviderTest.java | 40 +++++++++++++
4 files changed, 191 insertions(+), 18 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/common/util/MqResponseCodes.java
b/server/src/main/java/org/apache/rocketmq/studio/common/util/MqResponseCodes.java
new file mode 100644
index 000000000..7270182be
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/common/util/MqResponseCodes.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.common.util;
+
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+
+/**
+ * Shared classifier for the RocketMQ response codes carried on broker/client
exceptions.
+ * Consolidates the per-provider copies that walk an exception cause chain
looking for a
+ * specific {@code ResponseCode} so the "RPC succeeded but there is no
business data" grading
+ * stays consistent. {@link MQClientException} and {@link MQBrokerException}
both expose
+ * {@code getResponseCode()} but share no common supertype, so both are
checked.
+ */
+public final class MqResponseCodes {
+
+ private MqResponseCodes() {
+ }
+
+ /**
+ * Returns {@code true} when any exception in the cause chain is an {@link
MQClientException}
+ * or {@link MQBrokerException} carrying one of the given response codes.
A self-referential
+ * cause terminates the walk instead of looping forever.
+ */
+ public static boolean hasResponseCode(Throwable error, int...
responseCodes) {
+ Throwable cause = error;
+ while (cause != null) {
+ Integer code = responseCodeOf(cause);
+ if (code != null) {
+ for (int responseCode : responseCodes) {
+ if (code == responseCode) {
+ return true;
+ }
+ }
+ }
+ cause = cause.getCause() == cause ? null : cause.getCause();
+ }
+ return false;
+ }
+
+ private static Integer responseCodeOf(Throwable throwable) {
+ if (throwable instanceof MQClientException clientException) {
+ return clientException.getResponseCode();
+ }
+ if (throwable instanceof MQBrokerException brokerException) {
+ return brokerException.getResponseCode();
+ }
+ return null;
+ }
+}
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 2191c6ea4..f133dd7ce 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
@@ -17,7 +17,6 @@
package org.apache.rocketmq.studio.provider.apache;
import org.apache.rocketmq.client.QueryResult;
-import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
@@ -31,6 +30,7 @@ import org.apache.rocketmq.remoting.protocol.route.QueueData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.util.MqResponseCodes;
import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
import org.apache.rocketmq.studio.instance.message.ConsumerStatusVO;
import org.apache.rocketmq.studio.instance.message.MessageProvider;
@@ -189,6 +189,13 @@ public class RocketMQMessageProvider implements
MessageProvider {
}
return result;
} catch (Exception e) {
+ if (MqResponseCodes.hasResponseCode(e, ResponseCode.NO_MESSAGE)) {
+ // MQAdminImpl.queryMessage throws
MQClientException(NO_MESSAGE) instead of
+ // returning an empty QueryResult when the key matches
nothing: the query
+ // completed, so the correct response is an empty list, not a
gateway error.
+ log.info("queryMessage(topic={}, key={}) matched nothing",
topic, key);
+ return Collections.emptyList();
+ }
log.warn("queryMessage(topic={}, key={}) failed: {}", topic, key,
e.getMessage());
throw new BusinessException(502, "Failed to query messages by key:
" + e.getMessage());
}
@@ -438,11 +445,12 @@ public class RocketMQMessageProvider implements
MessageProvider {
} catch (BusinessException e) {
throw e;
} catch (Exception e) {
- if (isTraceTopicAbsent(e)) {
- // The cluster has no trace topic route (trace dispatch
disabled): the RPC
- // succeeded but there is no business data, so return an empty
trace instead
- // of surfacing an error (exception-grading convention).
- log.info("Trace topic not available on this cluster
(msgId={}), returning empty trace", msgId);
+ if (MqResponseCodes.hasResponseCode(e,
ResponseCode.TOPIC_NOT_EXIST, ResponseCode.NO_MESSAGE)) {
+ // The cluster has no trace topic route (trace dispatch
disabled) or the
+ // message simply has no trace records: the RPC succeeded but
there is no
+ // business data, so return an empty trace instead of
surfacing an error
+ // (exception-grading convention).
+ log.info("No trace data available for msgId={} ({}), returning
empty trace", msgId, e.getMessage());
return emptyTrace();
}
log.warn("Trace query for msgId={} failed: {}", msgId,
e.getMessage());
@@ -455,18 +463,6 @@ public class RocketMQMessageProvider implements
MessageProvider {
.build();
}
- private static boolean isTraceTopicAbsent(Throwable error) {
- Throwable cause = error;
- while (cause != null) {
- if (cause instanceof MQClientException clientException
- && clientException.getResponseCode() ==
ResponseCode.TOPIC_NOT_EXIST) {
- return true;
- }
- cause = cause.getCause() == cause ? null : cause.getCause();
- }
- return false;
- }
-
/**
* Trace lookup by business key. The key query already scopes the returned
trace messages to
* the requested message, so the body parser does not filter on a message
id. The original
@@ -496,6 +492,10 @@ public class RocketMQMessageProvider implements
MessageProvider {
} catch (BusinessException e) {
throw e;
} catch (Exception e) {
+ if (MqResponseCodes.hasResponseCode(e,
ResponseCode.TOPIC_NOT_EXIST, ResponseCode.NO_MESSAGE)) {
+ log.info("No trace data available for key={} ({}), returning
empty trace", key, e.getMessage());
+ return emptyTrace();
+ }
log.warn("Trace query by key={} failed: {}", key, e.getMessage());
throw new BusinessException(502, "Failed to query message trace by
key: " + e.getMessage());
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/common/util/MqResponseCodesTest.java
b/server/src/test/java/org/apache/rocketmq/studio/common/util/MqResponseCodesTest.java
new file mode 100644
index 000000000..cfb229fa1
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/common/util/MqResponseCodesTest.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.common.util;
+
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MqResponseCodesTest {
+
+ @Test
+ void matchesClientExceptionResponseCodeTest() {
+ MQClientException error = new
MQClientException(ResponseCode.NO_MESSAGE, "no message");
+ assertThat(MqResponseCodes.hasResponseCode(error,
+ ResponseCode.TOPIC_NOT_EXIST,
ResponseCode.NO_MESSAGE)).isTrue();
+ }
+
+ @Test
+ void matchesBrokerExceptionResponseCodeTest() {
+ MQBrokerException error = new
MQBrokerException(ResponseCode.CONSUMER_NOT_ONLINE, "not online");
+ assertThat(MqResponseCodes.hasResponseCode(error,
ResponseCode.CONSUMER_NOT_ONLINE)).isTrue();
+ }
+
+ @Test
+ void matchesResponseCodeOnNestedCauseTest() {
+ MQClientException cause = new
MQClientException(ResponseCode.TOPIC_NOT_EXIST, "no route");
+ RuntimeException wrapper = new RuntimeException("wrapper", cause);
+ assertThat(MqResponseCodes.hasResponseCode(wrapper,
ResponseCode.TOPIC_NOT_EXIST)).isTrue();
+ }
+
+ @Test
+ void returnsFalseWhenResponseCodeAbsentTest() {
+ MQClientException error = new
MQClientException(ResponseCode.TOPIC_NOT_EXIST, "no route");
+ assertThat(MqResponseCodes.hasResponseCode(error,
ResponseCode.NO_MESSAGE)).isFalse();
+ }
+
+ @Test
+ void returnsFalseForNullExceptionTest() {
+ assertThat(MqResponseCodes.hasResponseCode(null,
ResponseCode.NO_MESSAGE)).isFalse();
+ }
+
+ @Test
+ void terminatesOnSelfReferentialCauseTest() {
+ Throwable selfCaused = new Throwable("self") {
+ @Override
+ public synchronized Throwable getCause() {
+ return this;
+ }
+ };
+ assertThat(MqResponseCodes.hasResponseCode(selfCaused,
ResponseCode.NO_MESSAGE)).isFalse();
+ }
+}
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 e38d1524e..7812c46f2 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
@@ -20,6 +20,7 @@ import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
+import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.MQClientAPIImpl;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.trace.TraceConstants;
@@ -27,6 +28,7 @@ import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.CMResult;
@@ -213,6 +215,18 @@ class RocketMQMessageProviderTest {
.satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(502));
}
+ @Test
+ void queryByKeyReturnsEmptyListWhenClientReportsNoMessage() throws
Exception {
+ // MQAdminImpl.queryMessage throws MQClientException(NO_MESSAGE)
instead of
+ // returning an empty QueryResult when the key matches nothing.
+ when(adminExt.queryMessage("TopicA", "order-1", 64, 100L, 200L))
+ .thenThrow(new MQClientException(ResponseCode.NO_MESSAGE,
+ "query message by key finished, but no message."));
+
+ assertThat(provider.queryMessages(
+ "instance-a", "TopicA", null, null, "order-1", 100L,
200L)).isEmpty();
+ }
+
@Test
void queryByMsgIdUsesDecodedPhysicalOffsetForFallback() throws Exception {
String msgId = "AC1E0A6400002A9F0000000001A3F2B1";
@@ -631,6 +645,32 @@ class RocketMQMessageProviderTest {
}
+ @Test
+ void getMessageTraceReturnsEmptyTraceWhenClientReportsNoMessage() throws
Exception {
+ // A message without trace data (trace disabled on the producer or
expired) is a
+ // completed query with no records, not a remote failure.
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenThrow(new MQClientException(ResponseCode.NO_MESSAGE,
+ "query message by key finished, but no message."));
+
+ TraceRecordVO record = provider.getMessageTrace("instance-a",
"msg-123", "orders");
+
+ assertThat(record.getNodes()).isEmpty();
+ assertThat(record.getConsumerStatus()).isEmpty();
+ }
+
+ @Test
+ void getMessageTraceByKeyReturnsEmptyTraceWhenClientReportsNoMessage()
throws Exception {
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenThrow(new MQClientException(ResponseCode.NO_MESSAGE,
+ "query message by key finished, but no message."));
+
+ TraceRecordVO record = provider.getMessageTraceByKey("instance-a",
"key-1", "orders", null);
+
+ assertThat(record.getNodes()).isEmpty();
+ assertThat(record.getConsumerStatus()).isEmpty();
+ }
+
@Test
void getMessageTraceQueriesCustomTraceTopicWhenProvided() throws Exception
{
String pub = traceContext("Pub", "1000", "cn", "prod-group", "TopicA",
"msg-custom",
