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 <[email protected]>
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(
