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 ec912d32 fix(server): correct Aliyun and Tencent OpenAPI parsing
(#1978)
ec912d32 is described below
commit ec912d32f7563bc9934358bf1bdda862e9b6bcec
Author: 0 <[email protected]>
AuthorDate: Thu Aug 13 16:12:07 2026 +0800
fix(server): correct Aliyun and Tencent OpenAPI parsing (#1978)
* fix(web): migrate deprecated Ant Design props across Studio pages
Consolidates 17 per-page migration PRs by 123123213weqw: message (#1835),
certificate (#1836), client (#1837), cluster (#1838), ACL (#1839),
consumer (#1840), DLQ (#1841), instance (#1842), topic (#1843),
alert rule (#1844), settings modal (#1845), alert asset modal (#1846),
group management (#1847), LiteTopic (#1848), producer (#1849),
proxy (#1850), SSL settings (#1851). Behavior unchanged.
* fix(server): harden validation, normalization and null-safety
Consolidates 40 backend robustness PRs by 123123213weqw (#1114,
#1733-#1735, #1737-#1739, #1742, #1744-#1745, #1854-#1882, #1884):
block new admin connections during shutdown, Locale.ROOT normalization
for LLM engines and agent providers, reject null or malformed persisted
JSON, clamp cloud catalog counts and retry values, skip null catalog,
trace and connection entries, report lost concurrent updates for alert
rules, ACL rules, credentials, instances and certificates, normalize
stored enum values, validate alert identifiers and stored auth modes.
* fix(web): guard stale responses and race conditions in Studio pages
Consolidates 16 frontend UX PRs by 123123213weqw (#1563, #1624, #1732,
#1743, #1747-#1750, #1852-#1853, #1883, #2020-#2024): DLQ resend request-id
guard, best-effort proxy address persistence, locale-independent navigation
search, preserve selected home prompt mode, normalize alert levels before
filtering, restore audit refresh loading state, bound AI SSE event buffers,
format blank dates as unavailable, hook mocks for instance changes and
topic-scoped traces, keep 401 errors for malformed request URLs, prevent
Mock-mode home reload loop, surface Message Explorer topic load failures,
refresh alerts after clearing acknowledged records, gate data source
mutations until the initial list is ready, ignore stale NameServer drift
bootstrap results.
* fix(server): correct Aliyun and Tencent OpenAPI parsing
Consolidates 3 cloud-vendor PRs by 123123213weqw (#1978-#1980): map
in-flight Tencent consume trace nodes to process instead of failed, parse
Aliyun OpenAPI timestamps as UTC+8 regardless of server zone, reuse the
TaskRequestId returned by the previous Tencent message query page.
---
.../studio/provider/alibaba/AliyunConverters.java | 39 ++++++++-
.../provider/tencent/TencentInstanceProvider.java | 37 +++++++--
.../alibaba/AliyunInstanceProviderTest.java | 11 +++
.../tencent/TencentInstanceProviderTest.java | 95 ++++++++++++++++++++++
4 files changed, 173 insertions(+), 9 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
index 8e501d0b..597b65d5 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
@@ -66,6 +66,9 @@ final class AliyunConverters {
static final int MESSAGE_MAX_PAGES = 5;
private static final DateTimeFormatter TIME_FORMATTER =
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+ // Aliyun RocketMQ OpenAPI timestamps are unzoned "yyyy-MM-dd HH:mm:ss"
strings interpreted as
+ // UTC+8 (Asia/Shanghai), regardless of the server's default zone.
+ private static final ZoneId ALIYUN_TIME_ZONE = ZoneId.of("Asia/Shanghai");
private AliyunConverters() {
}
@@ -96,6 +99,9 @@ final class AliyunConverters {
List<CloudInstanceDetailVO.CloudEndpoint> endpoints = new
ArrayList<>();
if (data.getNetworkInfo() != null &&
data.getNetworkInfo().getEndpoints() != null) {
for (GetInstanceResponseBody.Endpoints endpoint :
data.getNetworkInfo().getEndpoints()) {
+ if (endpoint == null) {
+ continue;
+ }
endpoints.add(new CloudInstanceDetailVO.CloudEndpoint(
endpoint.getEndpointType(),
endpoint.getEndpointUrl()));
}
@@ -211,7 +217,7 @@ final class AliyunConverters {
.msgId(data.getMessageId())
.topic(data.getTopicName())
.tag(data.getMessageTag())
- .key(data.getMessageKeys() == null ? null : String.join(" ",
data.getMessageKeys()))
+ .key(joinMessageKeys(data.getMessageKeys()))
.bornHost(data.getBornHost())
.storeHost(data.getStoreHost())
.storeTime(parseTimeMillis(data.getStoreTime()))
@@ -229,6 +235,9 @@ final class AliyunConverters {
List<TraceNodeVO> nodes = new ArrayList<>();
if (data.getProducerInfo() != null &&
data.getProducerInfo().getRecords() != null) {
for (GetTraceResponseBody.ProducerInfoRecords record :
data.getProducerInfo().getRecords()) {
+ if (record == null) {
+ continue;
+ }
nodes.add(TraceNodeVO.builder()
.title("Producer")
.timestamp(parseTimeMillis(record.getProduceTime()))
@@ -240,6 +249,9 @@ final class AliyunConverters {
}
if (data.getBrokerInfo() != null &&
data.getBrokerInfo().getOperations() != null) {
for (GetTraceResponseBody.Operations operation :
data.getBrokerInfo().getOperations()) {
+ if (operation == null) {
+ continue;
+ }
nodes.add(TraceNodeVO.builder()
.title("Broker " + operation.getOperateType())
.timestamp(parseTimeMillis(operation.getOperateTime()))
@@ -248,6 +260,9 @@ final class AliyunConverters {
}
if (data.getConsumerInfos() != null) {
for (GetTraceResponseBody.ConsumerInfos consumerInfo :
data.getConsumerInfos()) {
+ if (consumerInfo == null) {
+ continue;
+ }
if (consumerInfo.getRecords() == null ||
consumerInfo.getRecords().isEmpty()) {
nodes.add(TraceNodeVO.builder()
.title("Consumer " +
consumerInfo.getConsumerGroupId())
@@ -256,6 +271,9 @@ final class AliyunConverters {
continue;
}
for (GetTraceResponseBody.Records record :
consumerInfo.getRecords()) {
+ if (record == null) {
+ continue;
+ }
String operateTime = null;
if (record.getOperations() != null &&
!record.getOperations().isEmpty()) {
operateTime =
record.getOperations().get(0).getOperateTime();
@@ -289,7 +307,7 @@ final class AliyunConverters {
}
try {
LocalDateTime dateTime = LocalDateTime.parse(value,
TIME_FORMATTER);
- return
dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+ return
dateTime.atZone(ALIYUN_TIME_ZONE).toInstant().toEpochMilli();
} catch (RuntimeException ignored) {
return 0L;
}
@@ -297,7 +315,7 @@ final class AliyunConverters {
static String formatTimeMillis(long epochMillis) {
return TIME_FORMATTER.format(
- LocalDateTime.ofInstant(Instant.ofEpochMilli(epochMillis),
ZoneId.systemDefault()));
+ LocalDateTime.ofInstant(Instant.ofEpochMilli(epochMillis),
ALIYUN_TIME_ZONE));
}
static String tryBase64Decode(String raw) {
@@ -321,6 +339,10 @@ final class AliyunConverters {
}
}
+ private static String joinMessageKeys(List<String> keys) {
+ return keys == null ? null : joinParts(" ",
keys.toArray(String[]::new));
+ }
+
private static String joinParts(String separator, String... parts) {
StringBuilder sb = new StringBuilder();
for (String part : parts) {
@@ -336,6 +358,15 @@ final class AliyunConverters {
}
private static Integer toInteger(Long value) {
- return value == null ? null : value.intValue();
+ if (value == null) {
+ return null;
+ }
+ if (value > Integer.MAX_VALUE) {
+ return Integer.MAX_VALUE;
+ }
+ if (value < 0) {
+ return 0;
+ }
+ return value.intValue();
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
index f205d5bc..f249d92d 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
@@ -137,7 +137,13 @@ public class TencentInstanceProvider implements
InstanceProvider {
}
// Use the response total count if available; otherwise fall back to
full scan
Long total = response.getTotalCount();
- return total != null ? Math.toIntExact(total) : listTopics(instanceId,
null, null, false).size();
+ if (total == null) {
+ return listTopics(instanceId, null, null, false).size();
+ }
+ if (total <= 0L) {
+ return 0;
+ }
+ return total > Integer.MAX_VALUE ? Integer.MAX_VALUE :
total.intValue();
}
@Override
@@ -476,6 +482,8 @@ public class TencentInstanceProvider implements
InstanceProvider {
request.setTopic(topic);
request.setStartTime(begin);
request.setEndTime(end);
+ // Reuse the task id returned by the previous page so paging
continues the same
+ // logical query; fall back to the initial random id when the API
omits it.
request.setTaskRequestId(taskRequestId);
if (StringUtils.hasText(key)) {
request.setMsgKey(key);
@@ -489,6 +497,9 @@ public class TencentInstanceProvider implements
InstanceProvider {
client -> client.DescribeMessageList(request));
MessageItem[] data = response == null ? null : response.getData();
long total = response == null ? 0L : (response.getTotalCount() ==
null ? 0L : response.getTotalCount());
+ if (response != null &&
StringUtils.hasText(response.getTaskRequestId())) {
+ taskRequestId = response.getTaskRequestId();
+ }
if (data != null) {
for (MessageItem item : data) {
if (item != null) {
@@ -673,7 +684,12 @@ public class TencentInstanceProvider implements
InstanceProvider {
return Collections.emptyMap();
}
Map<String, String> properties = new LinkedHashMap<>();
- root.fields().forEachRemaining(entry ->
properties.put(entry.getKey(), entry.getValue().asText("")));
+ root.fields().forEachRemaining(entry -> {
+ JsonNode value = entry.getValue();
+ if (value != null && value.isValueNode() && !value.isNull()) {
+ properties.put(entry.getKey(), value.asText());
+ }
+ });
return properties;
} catch (Exception e) {
return Collections.emptyMap();
@@ -715,8 +731,16 @@ public class TencentInstanceProvider implements
InstanceProvider {
}
private static String toConsumeTraceStatus(int status) {
- // Tencent consume log Status uses the RocketMQ convention where 2
means consumed.
- return status == 2 ? "finish" : "failed";
+ // Tencent consume log Status uses the RocketMQ convention where 0/1
are in-flight and
+ // 2 means consumed; keep the trace status consistent with
toDeliveryStatus and with the
+ // frontend TraceNode.status values ('error' | 'wait' | 'process' |
'finish').
+ if (status == 2) {
+ return "finish";
+ }
+ if (status == 0 || status == 1) {
+ return "process";
+ }
+ return "error";
}
private static DeliveryStatus toDeliveryStatus(int status) {
@@ -859,7 +883,10 @@ public class TencentInstanceProvider implements
InstanceProvider {
}
private static int toInt(Long value) {
- return value == null ? 0 : Math.toIntExact(value);
+ if (value == null || value <= 0L) {
+ return 0;
+ }
+ return value > Integer.MAX_VALUE ? Integer.MAX_VALUE :
value.intValue();
}
private static ConsumeType toConsumeType(Boolean consumeMessageOrderly) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
index a3d461f4..d39d0a6f 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
@@ -555,4 +555,15 @@ class AliyunInstanceProviderTest {
org.junit.jupiter.api.Assertions.assertEquals("Concurrently",
AliyunInstanceProvider.normalizeDeliveryOrderType("Concurrently"));
}
+
+ @Test
+ void timeConversionUsesAliyunUtc8Zone() {
+ // "2024-01-01 00:00:00" is 2023-12-31T16:00:00Z in UTC+8 regardless
of server zone.
+ long expectedUtc8 = java.time.LocalDateTime.of(2024, 1, 1, 0, 0)
+
.atZone(java.time.ZoneId.of("Asia/Shanghai")).toInstant().toEpochMilli();
+ assertThat(AliyunConverters.parseTimeMillis("2024-01-01
00:00:00")).isEqualTo(expectedUtc8);
+
+ // Round-trip formatting must restore the same calendar time in UTC+8.
+
assertThat(AliyunConverters.formatTimeMillis(expectedUtc8)).isEqualTo("2024-01-01
00:00:00");
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
index 5fed4b7d..f61d5d0e 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
@@ -110,6 +110,16 @@ class TencentInstanceProviderTest {
});
}
+ @Test
+ void countTopicsShouldClampOversizedTotals() throws Exception {
+ DescribeTopicListResponse response = new DescribeTopicListResponse();
+ response.setData(new TopicItem[]{topicItem("orders", "NORMAL", 8L)});
+ response.setTotalCount(Long.MAX_VALUE);
+ when(client.DescribeTopicList(any())).thenReturn(response);
+
+
assertThat(provider.countTopics(STUDIO_INSTANCE_ID)).isEqualTo(Integer.MAX_VALUE);
+ }
+
@Test
void listTopicsShouldMapAndFilterAndEnrichTimesTest() throws Exception {
TopicItem normal = topicItem("orders", "NORMAL", 8L);
@@ -293,6 +303,20 @@ class TencentInstanceProviderTest {
.containsOnly(100L);
}
+ @Test
+ void listConsumerGroupsShouldClampOversizedRetryCounts() throws Exception {
+ ConsumeGroupItem item = new ConsumeGroupItem();
+ item.setConsumerGroup("GID_test");
+ item.setMaxRetryTimes(Long.MAX_VALUE);
+ DescribeConsumerGroupListResponse response = new
DescribeConsumerGroupListResponse();
+ response.setData(new ConsumeGroupItem[]{item});
+ when(client.DescribeConsumerGroupList(any())).thenReturn(response);
+
+ assertThat(provider.listConsumerGroups(STUDIO_INSTANCE_ID, null))
+ .singleElement()
+ .satisfies(group ->
assertThat(group.getRetryMaxTimes()).isEqualTo(Integer.MAX_VALUE));
+ }
+
@Test
void listConsumerGroupsShouldMapAndFilterTest() throws Exception {
ConsumeGroupItem one = new ConsumeGroupItem();
@@ -436,6 +460,25 @@ class TencentInstanceProviderTest {
assertThat(captor.getValue().getMsgId()).isEqualTo("MSG-1");
}
+ @Test
+ void queryMessagesShouldSkipNullAndStructuredPropertyValues() throws
Exception {
+ DescribeMessageResponse detail = new DescribeMessageResponse();
+ detail.setMessageId("MSG-1");
+ detail.setShowTopicName("orders");
+ detail.setProperties("{\"KEYS\":\"keyA\",\"TAGS\":null,"
+ + "\"nested\":{\"x\":1},\"retry\":3,\"enabled\":true}");
+ when(client.DescribeMessage(any())).thenReturn(detail);
+
+ MessageRecordVO record = provider.queryMessages(
+ STUDIO_INSTANCE_ID, "orders", "MSG-1", null, null, null,
null).get(0);
+
+ assertThat(record.getProperties())
+ .containsEntry("KEYS", "keyA")
+ .containsEntry("retry", "3")
+ .containsEntry("enabled", "true")
+ .doesNotContainKeys("TAGS", "nested");
+ }
+
@Test
void queryMessagesByTopicShouldUseMessageListTest() throws Exception {
MessageItem one = new MessageItem();
@@ -511,6 +554,38 @@ class TencentInstanceProviderTest {
.isNotBlank();
}
+ @Test
+ void messageQueryReusesTaskRequestIdReturnedByPreviousPage() throws
Exception {
+ MessageItem[] page1Items = new
MessageItem[TencentInstanceProvider.MESSAGE_LIMIT];
+ for (int i = 0; i < TencentInstanceProvider.MESSAGE_LIMIT; i++) {
+ MessageItem item = new MessageItem();
+ item.setMsgId("MSG-" + (i + 1));
+ item.setProduceTime("2024-09-12 14:06:55,591");
+ page1Items[i] = item;
+ }
+ MessageItem last = new MessageItem();
+ last.setMsgId("MSG-LAST");
+ last.setProduceTime("2024-09-12 14:06:56,591");
+ DescribeMessageListResponse page1 = new DescribeMessageListResponse();
+ page1.setData(page1Items);
+ page1.setTaskRequestId("task-abc");
+ DescribeMessageListResponse page2 = new DescribeMessageListResponse();
+ page2.setData(new MessageItem[]{last});
+ page2.setTaskRequestId("task-abc");
+ when(client.DescribeMessageList(any()))
+ .thenReturn(page1)
+ .thenReturn(page2);
+
+ provider.queryMessages(STUDIO_INSTANCE_ID, "orders", null, null, null,
+ 1600000000000L, 1600001000000L);
+
+ ArgumentCaptor<DescribeMessageListRequest> captor =
ArgumentCaptor.forClass(DescribeMessageListRequest.class);
+ verify(client,
org.mockito.Mockito.times(2)).DescribeMessageList(captor.capture());
+ java.util.List<DescribeMessageListRequest> requests =
captor.getAllValues();
+ // The second page must carry the task id returned by the first page,
not a fresh random id.
+ assertThat(requests.get(1).getTaskRequestId()).isEqualTo("task-abc");
+ }
+
@Test
void getMessageTraceShouldMapStagesTest() throws Exception {
MessageTraceItem produce = new MessageTraceItem();
@@ -546,4 +621,24 @@ class TencentInstanceProviderTest {
assertThat(captor.getValue().getTopic()).isEqualTo("orders");
assertThat(captor.getValue().getMsgId()).isEqualTo("MSG-1");
}
+
+ @Test
+ void getMessageTraceMarksInFlightConsumeAsProcessNotFailed() throws
Exception {
+ MessageTraceItem consume = new MessageTraceItem();
+ consume.setStage("consume");
+
consume.setData("{\"TotalCount\":1,\"RocketMqConsumeLogs\":[{\"MsgId\":\"MSG-2\",\"Status\":1,"
+ + "\"PushTime\":\"2024-09-12
14:06:55,600\",\"ConsumerGroup\":\"GID_test\",\"RetryTimes\":0}]}");
+ DescribeMessageTraceResponse response = new
DescribeMessageTraceResponse();
+ response.setData(new MessageTraceItem[]{consume});
+ when(client.DescribeMessageTrace(any())).thenReturn(response);
+
+ TraceRecordVO trace = provider.getMessageTrace(STUDIO_INSTANCE_ID,
"MSG-2", "orders");
+
+ assertThat(trace.getNodes()).hasSize(1);
+ // In-flight (Status 1) must read "process", consistent with
toDeliveryStatus and the
+ // frontend TraceNode status union, not "failed".
+ assertThat(trace.getNodes().get(0).getStatus()).isEqualTo("process");
+ assertThat(trace.getConsumerStatus().get(0).getDeliveryStatus())
+ .isEqualTo(DeliveryStatus.pending);
+ }
}
