You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2026/09/21 09:29:49 UTC
(rocketmq-dashboard) branch rocketmq-studio updated: fix(message): surface message-id lookup failures (#4525)
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 85f839b6c fix(message): surface message-id lookup failures (#4525)
85f839b6c is described below
commit 85f839b6c35b108de0844e71c75bc3abeb73981d
Author: zmuxuny <xi...@gmail.com>
AuthorDate: Mon Sep 21 17:29:43 2026 +0800
fix(message): surface message-id lookup failures (#4525)
queryByMsgId conflated operational lookup failures with an absent message: both the primary viewMessage call and the decoded-offset fallback reduced errors to an empty result. Known absence (TOPIC_NOT_EXIST / NO_MESSAGE / QUERY_NOT_FOUND via MqResponseCodes) still returns an empty list; any other failure now raises BusinessException(502) so the console reports the lookup error instead of "no such message".
Fixes #4524
---
.../provider/apache/RocketMQMessageProvider.java | 69 ++++++++++++++++++----
.../apache/RocketMQMessageProviderTest.java | 33 ++++++++++-
2 files changed, 90 insertions(+), 12 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 4b42fccaf..d17bb88ec 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
@@ -145,17 +145,30 @@ public class RocketMQMessageProvider implements MessageProvider {
private List<MessageRecordVO> queryByMsgId(DefaultMQAdminExt adminExt, String topic, String msgId) {
MessageExt messageExt = null;
+ Exception primaryFailure = null;
if (StringUtils.hasText(topic)) {
+ if (!BrokerTopologyGuards.isWithinKnownBrokerTopology(adminExt, msgId)) {
+ return Collections.emptyList();
+ }
try {
- if (BrokerTopologyGuards.isWithinKnownBrokerTopology(adminExt, msgId)) {
- messageExt = adminExt.viewMessage(topic, msgId);
- }
+ messageExt = adminExt.viewMessage(topic, msgId);
} catch (Exception e) {
+ if (isMessageLookupAbsent(e)) {
+ return Collections.emptyList();
+ }
+ primaryFailure = e;
log.warn("viewMessage(topic={}, msgId={}) failed: {}", topic, msgId, e.getMessage());
}
}
if (messageExt == null) {
- messageExt = viewMessageByOffsetId(adminExt, topic, msgId);
+ OffsetMessageLookup lookup = lookupMessageByOffsetId(adminExt, topic, msgId);
+ messageExt = lookup.message();
+ if (messageExt == null && lookup.failure() != null) {
+ throw messageLookupFailure(lookup.failure());
+ }
+ }
+ if (messageExt == null && primaryFailure != null) {
+ throw messageLookupFailure(primaryFailure);
}
if (messageExt == null) {
return Collections.emptyList();
@@ -168,20 +181,50 @@ public class RocketMQMessageProvider implements MessageProvider {
* msgId, then querying that broker directly.
*/
private MessageExt viewMessageByOffsetId(DefaultMQAdminExt adminExt, String topic, String msgId) {
+ OffsetMessageLookup lookup = lookupMessageByOffsetId(adminExt, topic, msgId);
+ if (lookup.failure() != null) {
+ log.warn("viewMessage by decoded offset id failed for msgId={}: {}",
+ msgId, lookup.failure().getMessage());
+ }
+ return lookup.message();
+ }
+
+ private OffsetMessageLookup lookupMessageByOffsetId(DefaultMQAdminExt adminExt, String topic, String msgId) {
+ MessageId messageId;
+ try {
+ messageId = MessageDecoder.decodeMessageId(msgId);
+ } catch (Exception exception) {
+ return OffsetMessageLookup.empty();
+ }
try {
- MessageId messageId = MessageDecoder.decodeMessageId(msgId);
String brokerAddr = BrokerTopologyGuards.validatedBrokerAddr(adminExt, msgId, messageId);
if (!StringUtils.hasText(brokerAddr)) {
- return null;
+ return OffsetMessageLookup.empty();
}
- return adminExt.getDefaultMQAdminExtImpl()
+ MessageExt message = adminExt.getDefaultMQAdminExtImpl()
.getMqClientInstance()
.getMQClientAPIImpl()
.viewMessage(brokerAddr, topic, messageId.getOffset(), VIEW_MESSAGE_TIMEOUT_MILLIS);
- } catch (Exception e) {
- log.warn("viewMessage by decoded offset id failed for msgId={}: {}", msgId, e.getMessage());
- return null;
+ return new OffsetMessageLookup(message, null);
+ } catch (Exception exception) {
+ if (isMessageLookupAbsent(exception)) {
+ return OffsetMessageLookup.empty();
+ }
+ return new OffsetMessageLookup(null, exception);
+ }
+ }
+
+ private static boolean isMessageLookupAbsent(Throwable throwable) {
+ return MqResponseCodes.hasResponseCode(throwable, ResponseCode.TOPIC_NOT_EXIST,
+ ResponseCode.NO_MESSAGE, ResponseCode.QUERY_NOT_FOUND);
+ }
+
+ private static BusinessException messageLookupFailure(Throwable throwable) {
+ String message = throwable.getMessage();
+ if (message == null || message.isBlank()) {
+ message = throwable.getClass().getSimpleName();
}
+ return new BusinessException(502, "Failed to query message by id: " + message);
}
private MessageQueryResult queryByKey(DefaultMQAdminExt adminExt, String topic, String key,
@@ -863,6 +906,12 @@ public class RocketMQMessageProvider implements MessageProvider {
return (value & 0xC0) == 0x80;
}
+ private record OffsetMessageLookup(MessageExt message, Throwable failure) {
+ private static OffsetMessageLookup empty() {
+ return new OffsetMessageLookup(null, null);
+ }
+ }
+
private record DisplayBody(String value, String encoding, boolean truncated) {
}
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 ba3d0acfa..37654fec4 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
@@ -385,6 +385,22 @@ class RocketMQMessageProviderTest {
verify(clientApi).viewMessage("172.30.10.100:10911", "TopicA", 27521713L, 3000L);
}
+ @Test
+ void queryByMsgIdSurfacesOffsetFallbackFailureTest() throws Exception {
+ String msgId = "AC1E0A6400002A9F0000000001A3F2B1";
+ MQClientAPIImpl clientApi = mockOffsetLookupClient();
+ when(adminExt.viewMessage("TopicA", msgId))
+ .thenThrow(new IllegalStateException("primary lookup failed"));
+ when(clientApi.viewMessage("172.30.10.100:10911", "TopicA", 27521713L, 3000L))
+ .thenThrow(new IllegalStateException("broker unavailable"));
+
+ assertThatThrownBy(() -> provider.queryMessages(
+ "instance-a", "TopicA", msgId, null, null, 100L, 200L))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Failed to query message by id: broker unavailable")
+ .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502));
+ }
+
@Test
void queryByMsgIdIgnoresUnrelatedTimeBounds() throws Exception {
MessageExt message = new MessageExt();
@@ -441,9 +457,10 @@ class RocketMQMessageProviderTest {
}
@Test
- void queryByMsgIdPassesNonOffsetIdsThroughToViewMessage() throws Exception {
+ void queryByMsgIdTreatsKnownNonOffsetAbsenceAsEmptyTest() throws Exception {
when(adminExt.viewMessage("TopicA", "uniq-key-1"))
- .thenThrow(new IllegalStateException("unique key lookup handled by MQAdminImpl"));
+ .thenThrow(new MQClientException(ResponseCode.QUERY_NOT_FOUND,
+ "query message by key finished, but no message"));
List<MessageRecordVO> result = provider.queryMessages(
"instance-a", "TopicA", "uniq-key-1", null, null, 100L, 200L);
@@ -453,6 +470,18 @@ class RocketMQMessageProviderTest {
verify(adminExt, never()).examineBrokerClusterInfo();
}
+ @Test
+ void queryByMsgIdSurfacesNonOffsetLookupFailureTest() throws Exception {
+ when(adminExt.viewMessage("TopicA", "uniq-key-failure"))
+ .thenThrow(new IllegalStateException("nameserver unavailable"));
+
+ assertThatThrownBy(() -> provider.queryMessages(
+ "instance-a", "TopicA", "uniq-key-failure", null, null, 100L, 200L))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Failed to query message by id: nameserver unavailable")
+ .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502));
+ }
+
@Test
void queryByMsgIdStillViewsMessagesInsideKnownTopology() throws Exception {
String msgId = MessageDecoder.createMessageId(