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 5d98c4907 feat(dlq): show user properties on dead-letter messages
(#3999)
5d98c4907 is described below
commit 5d98c4907deba613b0953320387a8735fa83df2a
Author: 烤化の初雪 <[email protected]>
AuthorDate: Mon Sep 7 18:18:59 2026 +0800
feat(dlq): show user properties on dead-letter messages (#3999)
The Apache DLQ provider builds DLQMessageVO from the full MessageExt it
has already scanned, but dropped the user properties. Carry them (and a
propertiesTruncated flag) through with the same 64-entry / 1024-char
limits the message explorer applies, and render them as an expandable
row in the DLQ message drawer, mirroring the classic dashboard's DLQ
detail dialog.
Related to #3998.
Co-authored-by: unbridled-41
<[email protected]>
---
.../studio/common/util/MessagePropertyDisplay.java | 80 ++++++++++++++++++++++
.../rocketmq/studio/instance/dlq/DLQMessageVO.java | 4 ++
.../provider/apache/RocketMQDLQProvider.java | 8 +++
.../provider/apache/RocketMQMessageProvider.java | 32 +--------
.../common/util/MessagePropertyDisplayTest.java | 67 ++++++++++++++++++
.../provider/apache/RocketMQDLQProviderTest.java | 52 ++++++++++++++
web/src/api/message.ts | 2 +
web/src/pages/instance/__tests__/DLQPage.test.tsx | 36 ++++++++++
web/src/pages/instance/dlq.tsx | 42 ++++++++++++
9 files changed, 294 insertions(+), 29 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplay.java
b/server/src/main/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplay.java
new file mode 100644
index 000000000..e12357581
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplay.java
@@ -0,0 +1,80 @@
+/*
+ * 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.common.message.MessageConst;
+
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * Shared rendering limits for message property maps, used by the message
explorer and the DLQ
+ * drawer so both apply the same {@value #MAX_PROPERTIES}-entry / {@value
#MAX_PROPERTY_VALUE_CHARS}-char
+ * caps instead of duplicating the logic. {@link #userProperties} additionally
drops the broker-set
+ * system keys ({@link MessageConst#STRING_HASH_SET}) so a view labelled "user
properties" is not
+ * crowded out by system entries once the cap and alphabetical ordering are
applied.
+ */
+public final class MessagePropertyDisplay {
+
+ public static final int MAX_PROPERTIES = 64;
+ public static final int MAX_PROPERTY_VALUE_CHARS = 1024;
+
+ private MessagePropertyDisplay() {
+ }
+
+ /** Keeps only user-defined properties, dropping broker-set system keys.
Null-safe. */
+ public static Map<String, String> userProperties(Map<String, String>
properties) {
+ if (properties == null || properties.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ Map<String, String> user = new LinkedHashMap<>();
+ properties.forEach((key, value) -> {
+ if (key != null && !MessageConst.STRING_HASH_SET.contains(key)) {
+ user.put(key, value);
+ }
+ });
+ return user;
+ }
+
+ /** Sorts by key, caps at {@link #MAX_PROPERTIES} entries, and abbreviates
each value. Null-safe. */
+ public static Map<String, String> limitProperties(Map<String, String>
properties) {
+ if (properties == null || properties.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ Map<String, String> limited = new LinkedHashMap<>();
+ properties.entrySet().stream()
+
.sorted(Map.Entry.comparingByKey(Comparator.nullsLast(String::compareTo)))
+ .limit(MAX_PROPERTIES)
+ .forEach(entry -> limited.put(entry.getKey(),
abbreviate(entry.getValue())));
+ return limited;
+ }
+
+ /** True when any value exceeds {@link #MAX_PROPERTY_VALUE_CHARS} and
would be abbreviated. */
+ public static boolean hasOversizedProperty(Map<String, String> properties)
{
+ return properties != null && properties.values().stream()
+ .anyMatch(value -> value != null && value.length() >
MAX_PROPERTY_VALUE_CHARS);
+ }
+
+ private static String abbreviate(String value) {
+ if (value == null || value.length() <= MAX_PROPERTY_VALUE_CHARS) {
+ return value;
+ }
+ return value.substring(0, MAX_PROPERTY_VALUE_CHARS) + "...";
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageVO.java
index a27842460..7986b0455 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageVO.java
@@ -16,6 +16,8 @@
*/
package org.apache.rocketmq.studio.instance.dlq;
+import java.util.Map;
+
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
@@ -40,4 +42,6 @@ public class DLQMessageVO {
private String keys;
private String body;
private String bodyBase64;
+ private Map<String, String> properties;
+ private boolean propertiesTruncated;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
index c0d10c60f..ae6d5b2f9 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
@@ -34,6 +34,7 @@ import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.util.MessagePropertyDisplay;
import org.apache.rocketmq.studio.common.util.Pagination;
import org.apache.rocketmq.studio.common.util.SystemTopicFilter;
import org.apache.rocketmq.studio.instance.dlq.DLQExcelExportResultVO;
@@ -395,6 +396,10 @@ public class RocketMQDLQProvider implements DLQProvider {
}
private DLQMessageVO toExportVO(MessageExt message) {
+ // Show only user-defined properties: broker-set system keys
(REAL_TOPIC, UNIQ_KEY, ...)
+ // would otherwise crowd out real user entries under the entry cap and
alphabetical order.
+ Map<String, String> userProperties =
MessagePropertyDisplay.userProperties(message.getProperties());
+ Map<String, String> displayProperties =
MessagePropertyDisplay.limitProperties(userProperties);
return DLQMessageVO.builder()
.msgId(message.getMsgId())
.topic(message.getTopic())
@@ -405,6 +410,9 @@ public class RocketMQDLQProvider implements DLQProvider {
.body(toUtf8Text(message.getBody()))
.bodyBase64(message.getBody() == null ? null
:
Base64.getEncoder().encodeToString(message.getBody()))
+ .properties(displayProperties)
+ .propertiesTruncated(displayProperties.size() <
userProperties.size()
+ ||
MessagePropertyDisplay.hasOversizedProperty(userProperties))
.build();
}
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 f133dd7ce..c6d6066c6 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
@@ -30,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.MessagePropertyDisplay;
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;
@@ -57,7 +58,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Base64;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
@@ -82,8 +82,6 @@ public class RocketMQMessageProvider implements
MessageProvider {
private static final int TOPIC_PULL_BATCH_SIZE = 32;
private static final int MAX_BODY_DISPLAY_BYTES = 64 * 1024;
private static final int MAX_BINARY_BODY_DISPLAY_BYTES = 48 * 1024;
- private static final int MAX_PROPERTIES = 64;
- private static final int MAX_PROPERTY_VALUE_CHARS = 1024;
private static final long VIEW_MESSAGE_TIMEOUT_MILLIS = 3000L;
private static final long ONE_HOUR_MILLIS = 3600_000L;
private static final long ONE_DAY_MILLIS = 24 * ONE_HOUR_MILLIS;
@@ -703,7 +701,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
byte[] body = messageExt.getBody();
DisplayBody displayBody = displayBody(body);
Map<String, String> properties = messageExt.getProperties();
- Map<String, String> displayProperties = limitProperties(properties);
+ Map<String, String> displayProperties =
MessagePropertyDisplay.limitProperties(properties);
return MessageRecordVO.builder()
.msgId(messageExt.getMsgId())
.topic(messageExt.getTopic())
@@ -720,7 +718,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
.storeHost(String.valueOf(messageExt.getStoreHost()))
.properties(displayProperties)
.propertiesTruncated(properties != null &&
(displayProperties.size() < properties.size()
- || hasOversizedProperty(properties)))
+ ||
MessagePropertyDisplay.hasOversizedProperty(properties)))
.size(messageExt.getStoreSize())
.build();
}
@@ -753,30 +751,6 @@ public class RocketMQMessageProvider implements
MessageProvider {
return (value & 0xC0) == 0x80;
}
- private Map<String, String> limitProperties(Map<String, String>
properties) {
- if (properties == null || properties.isEmpty()) {
- return Collections.emptyMap();
- }
- Map<String, String> limited = new LinkedHashMap<>();
- properties.entrySet().stream()
-
.sorted(Map.Entry.comparingByKey(Comparator.nullsLast(String::compareTo)))
- .limit(MAX_PROPERTIES)
- .forEach(entry -> limited.put(entry.getKey(),
abbreviate(entry.getValue(), MAX_PROPERTY_VALUE_CHARS)));
- return limited;
- }
-
- private boolean hasOversizedProperty(Map<String, String> properties) {
- return properties != null && properties.values().stream()
- .anyMatch(value -> value != null && value.length() >
MAX_PROPERTY_VALUE_CHARS);
- }
-
- private String abbreviate(String value, int maxLength) {
- if (value == null || value.length() <= maxLength) {
- return value;
- }
- return value.substring(0, maxLength) + "...";
- }
-
private record DisplayBody(String value, String encoding, boolean
truncated) {
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplayTest.java
b/server/src/test/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplayTest.java
new file mode 100644
index 000000000..e3edf1bd1
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplayTest.java
@@ -0,0 +1,67 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MessagePropertyDisplayTest {
+
+ @Test
+ void userPropertiesShouldDropBrokerSystemKeysTest() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("REAL_TOPIC", "%RETRY%group");
+ properties.put("KEYS", "k1");
+ properties.put("traceId", "abc-123");
+ assertThat(MessagePropertyDisplay.userProperties(properties))
+ .containsEntry("traceId", "abc-123")
+ .doesNotContainKeys("REAL_TOPIC", "KEYS");
+ }
+
+ @Test
+ void userPropertiesShouldReturnEmptyForNullOrEmptyTest() {
+ assertThat(MessagePropertyDisplay.userProperties(null)).isEmpty();
+ assertThat(MessagePropertyDisplay.userProperties(Map.of())).isEmpty();
+ }
+
+ @Test
+ void limitPropertiesShouldCapEntryCountTest() {
+ Map<String, String> properties = new HashMap<>();
+ for (int i = 0; i < 80; i++) {
+ properties.put(String.format("k%02d", i), "v" + i);
+ }
+ assertThat(MessagePropertyDisplay.limitProperties(properties))
+ .hasSize(MessagePropertyDisplay.MAX_PROPERTIES);
+ }
+
+ @Test
+ void limitPropertiesShouldAbbreviateOversizedValueTest() {
+ Map<String, String> limited =
MessagePropertyDisplay.limitProperties(Map.of("big", "x".repeat(1500)));
+ assertThat(limited.get("big")).isEqualTo("x".repeat(1024) + "...");
+ }
+
+ @Test
+ void hasOversizedPropertyShouldDetectLongValuesTest() {
+ assertThat(MessagePropertyDisplay.hasOversizedProperty(Map.of("big",
"x".repeat(1500)))).isTrue();
+ assertThat(MessagePropertyDisplay.hasOversizedProperty(Map.of("k",
"short"))).isFalse();
+
assertThat(MessagePropertyDisplay.hasOversizedProperty(null)).isFalse();
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
index b44c4d691..e436f987b 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
@@ -333,6 +333,58 @@ class RocketMQDLQProviderTest {
verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
}
+ @Test
+ void listMessagesShouldCarryLimitedUserProperties() throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(dlqTopic, "broker-a", 0);
+
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(pullConsumer.searchOffset(eq(queue), anyLong())).thenReturn(0L);
+ MessageExt deadLetter = new MessageExt();
+ deadLetter.setMsgId("dlq-msg-1");
+ deadLetter.setTopic("orders");
+ deadLetter.setQueueId(0);
+ deadLetter.setQueueOffset(7L);
+ deadLetter.setStoreTimestamp(1_700_000_000_000L);
+ deadLetter.setKeys("key-1");
+ deadLetter.setBody("payload".getBytes(StandardCharsets.UTF_8));
+ deadLetter.putUserProperty("traceId", "abc-123");
+ deadLetter.putUserProperty("region", "cn-east-1");
+ PullResult pullResult = new PullResult(PullStatus.FOUND, 1L, 0L, 0L,
List.of(deadLetter));
+ when(pullConsumer.pull(eq(queue), eq("*"), anyLong(),
anyInt())).thenReturn(pullResult);
+
+ PageResult<DLQMessageVO> page = provider.listMessages(
+ "instance-a", "group-a", 1_699_999_000_000L,
1_700_100_000_000L, 1, 20);
+
+ assertThat(page.getItems()).hasSize(1);
+ assertThat(page.getItems().get(0).getProperties())
+ .containsEntry("traceId", "abc-123")
+ .containsEntry("region", "cn-east-1");
+ assertThat(page.getItems().get(0).isPropertiesTruncated()).isFalse();
+ }
+
+ @Test
+ void listMessagesShouldFlagTruncatedPropertyPayloads() throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(dlqTopic, "broker-a", 0);
+
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(pullConsumer.searchOffset(eq(queue), anyLong())).thenReturn(0L);
+ MessageExt deadLetter = new MessageExt();
+ deadLetter.setMsgId("dlq-msg-2");
+ deadLetter.setTopic("orders");
+ deadLetter.setStoreTimestamp(1_700_000_000_000L);
+ deadLetter.setBody("payload".getBytes(StandardCharsets.UTF_8));
+ deadLetter.putUserProperty("big", "x".repeat(1500));
+ PullResult pullResult = new PullResult(PullStatus.FOUND, 1L, 0L, 0L,
List.of(deadLetter));
+ when(pullConsumer.pull(eq(queue), eq("*"), anyLong(),
anyInt())).thenReturn(pullResult);
+
+ PageResult<DLQMessageVO> page = provider.listMessages(
+ "instance-a", "group-a", 1_699_999_000_000L,
1_700_100_000_000L, 1, 20);
+
+ DLQMessageVO message = page.getItems().get(0);
+
assertThat(message.getProperties().get("big")).isEqualTo("x".repeat(1024) +
"...");
+ assertThat(message.isPropertiesTruncated()).isTrue();
+ }
+
@Test
void resendSelectedMessagesResolvesInTopologyMsgIdNormally() throws
Exception {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
diff --git a/web/src/api/message.ts b/web/src/api/message.ts
index 101de0172..3ce1539a1 100644
--- a/web/src/api/message.ts
+++ b/web/src/api/message.ts
@@ -117,6 +117,8 @@ export interface DLQMessage {
keys: string | null;
body: string | null;
bodyBase64: string | null;
+ properties?: Record<string, string>;
+ propertiesTruncated?: boolean;
}
export interface DLQMessagePage {
diff --git a/web/src/pages/instance/__tests__/DLQPage.test.tsx
b/web/src/pages/instance/__tests__/DLQPage.test.tsx
index ed15295ed..d9324abdf 100644
--- a/web/src/pages/instance/__tests__/DLQPage.test.tsx
+++ b/web/src/pages/instance/__tests__/DLQPage.test.tsx
@@ -259,6 +259,42 @@ describe('DLQ page', () => {
);
});
+ it('shows user properties in the DLQ message drawer', async () => {
+ vi.mocked(messageService.listDLQMessages).mockResolvedValue({
+ items: [
+ {
+ msgId: 'dlq-1',
+ topic: 'orders',
+ queueId: 0,
+ offset: 7,
+ storeTime: 1_700_000_000_000,
+ keys: 'key-1',
+ body: 'payload',
+ bodyBase64: null,
+ properties: { traceId: 'abc-123', region: 'cn-east-1' },
+ propertiesTruncated: false,
+ },
+ ],
+ total: 1,
+ page: 1,
+ size: 20,
+ } satisfies DLQMessagePage);
+ const user = userEvent.setup();
+ renderWithProviders(<DLQPage />);
+
+ await screen.findByText('cg-order');
+ await user.click(screen.getByRole('button', { name: /消息明细/ }));
+
+ const keyCell = await screen.findByText('key-1');
+ const row = keyCell.closest('tr') as HTMLElement;
+ await user.click(within(row).getByRole('button', { name: /expand/i }));
+
+ expect(await screen.findByText('traceId')).toBeInTheDocument();
+ expect(screen.getByText('abc-123')).toBeInTheDocument();
+ expect(screen.getByText('region')).toBeInTheDocument();
+ expect(screen.getByText('cn-east-1')).toBeInTheDocument();
+ });
+
it('exports the dead-letter messages of a group as Excel', async () => {
vi.mocked(messageService.exportDLQExcel).mockResolvedValue({
blob: new Blob(['xlsx-bytes'], {
diff --git a/web/src/pages/instance/dlq.tsx b/web/src/pages/instance/dlq.tsx
index 3476cb666..03fa56033 100644
--- a/web/src/pages/instance/dlq.tsx
+++ b/web/src/pages/instance/dlq.tsx
@@ -858,6 +858,48 @@ const DLQPage = () => {
size="small"
loading={detailLoading}
dataSource={detailMessages}
+ expandable={{
+ expandedRowRender: (record) =>
+ record.properties && Object.keys(record.properties).length >
0 ? (
+ <div style={{ padding: '4px 0' }}>
+ {record.propertiesTruncated && (
+ <Text
+ type="warning"
+ style={{ fontSize: 14, display: 'block',
marginBottom: 4 }}
+ >
+ 属性过多或单值过长，服务端已截断展示
+ </Text>
+ )}
+ <Table
+ size="small"
+ pagination={false}
+ rowKey={(p) => p.key}
+
dataSource={Object.entries(record.properties).map(([key, value]) => ({
+ key,
+ value,
+ }))}
+ columns={[
+ {
+ title: '属性',
+ dataIndex: 'key',
+ key: 'key',
+ width: 200,
+ ellipsis: true,
+ },
+ { title: '值', dataIndex: 'value', key: 'value',
ellipsis: true },
+ ]}
+ locale={{ emptyText: '无属性' }}
+ />
+ </div>
+ ) : (
+ <Text
+ type="secondary"
+ style={{ padding: '4px 0', display: 'block', fontSize:
14 }}
+ >
+ 该消息无用户属性
+ </Text>
+ ),
+ }}
rowSelection={{
selectedRowKeys: detailSelectedMsgIds,
onChange: (keys) => setDetailSelectedMsgIds(keys.map(String)),
