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 fe81cec9a fix(ai): emit non-null cluster version and bound message
query timestamps (#2311)
fe81cec9a is described below
commit fe81cec9a5baec0c6974e7d5f316883aafa0e94d
Author: 0 <[email protected]>
AuthorDate: Wed Aug 19 14:27:13 2026 +0800
fix(ai): emit non-null cluster version and bound message query timestamps
(#2311)
* fix(ai): emit non-null cluster version in cluster list and capabilities
tools
The Apache runtime provider never populates ClusterVO.version, so the
rmq.cluster.list and rmq.capabilities tool projections emitted null into
the output schema's required version string, causing output schema
validation to throw and surface as HTTP 500. Project null id, type and
version members to blank strings so the output schema is satisfied.
* fix(ai): bound timestamp arguments in message query tool
The rmq.message.query tool parsed startTime/endTime with
Number.longValue(), which silently wraps values above Long.MAX_VALUE
(Jackson delivers them as BigInteger) into a nonsense, possibly negative,
epoch-millis timestamp, and Long.parseLong() propagated
NumberFormatException
as a 500. Convert with longValueExact, reject non-finite/out-of-range
floating-point values, and map unparseable/unrepresentable strings and
numbers to HTTP 400.
---
.../ops/ai/tool/CapabilitiesToolHandler.java | 15 +++-
.../studio/ops/ai/tool/ClusterListToolHandler.java | 15 +++-
.../ops/ai/tool/MessageQueryToolHandler.java | 37 +++++++++-
.../ops/ai/tool/CapabilitiesToolHandlerTest.java | 60 ++++++++++++++++
.../ops/ai/tool/ClusterListToolHandlerTest.java | 82 ++++++++++++++++++++++
.../ops/ai/tool/MessageQueryToolHandlerTest.java | 45 ++++++++++++
6 files changed, 246 insertions(+), 8 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandler.java
index e17505457..9b06eb25b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandler.java
@@ -46,10 +46,19 @@ public class CapabilitiesToolHandler implements ToolHandler
{
List<String> capabilities = capabilityResolver.resolve(cluster);
Map<String, Object> result = new LinkedHashMap<>();
- result.put("cluster", cluster.getId());
- result.put("type", cluster.getType().name());
- result.put("version", cluster.getVersion());
+ result.put("cluster", blankIfNull(cluster.getId()));
+ result.put("type", cluster.getType() == null ? "" :
cluster.getType().name());
+ result.put("version", blankIfNull(cluster.getVersion()));
result.put("capabilities", capabilities);
return result;
}
+
+ /**
+ * The output schema declares cluster/type/version as required strings;
providers that
+ * do not populate them (notably the Apache runtime provider, which
reports no cluster
+ * version) must not emit nulls or schema validation turns the tool call
into a 500.
+ */
+ private static String blankIfNull(String value) {
+ return value == null ? "" : value;
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/ClusterListToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/ClusterListToolHandler.java
index 9276729bb..67e6836c1 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/ClusterListToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/ClusterListToolHandler.java
@@ -46,14 +46,23 @@ public class ClusterListToolHandler implements ToolHandler {
private static Map<String, Object> safeProjection(ClusterVO cluster) {
Map<String, Object> result = new LinkedHashMap<>();
- result.put("id", cluster.getId());
- result.put("name", cluster.getName());
+ result.put("id", blankIfNull(cluster.getId()));
+ result.put("name", blankIfNull(cluster.getName()));
result.put("type", requiredEnumName(cluster.getType(), "type",
cluster.getId()));
result.put("status", requiredEnumName(cluster.getStatus(), "status",
cluster.getId()));
- result.put("version", cluster.getVersion());
+ result.put("version", blankIfNull(cluster.getVersion()));
return result;
}
+ /**
+ * The output schema declares id/name/version as required strings;
providers that do
+ * not populate them (notably the Apache runtime provider, which reports
no cluster
+ * version) must not emit nulls or schema validation turns the tool call
into a 500.
+ */
+ private static String blankIfNull(String value) {
+ return value == null ? "" : value;
+ }
+
private static String requiredEnumName(Enum<?> value, String field, String
clusterId) {
if (value == null) {
throw new IllegalStateException(
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/MessageQueryToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/MessageQueryToolHandler.java
index 6ea69ca04..569bbc00b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/MessageQueryToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/MessageQueryToolHandler.java
@@ -16,11 +16,13 @@
*/
package org.apache.rocketmq.studio.ops.ai.tool;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
import org.apache.rocketmq.studio.instance.message.MessageService;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
+import java.math.BigInteger;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -76,9 +78,40 @@ public class MessageQueryToolHandler implements ToolHandler {
return null;
}
if (value instanceof Number number) {
- return number.longValue();
+ return toEpochMillis(number);
}
- return Long.parseLong(value.toString());
+ try {
+ return Long.parseLong(value.toString().trim());
+ } catch (NumberFormatException ex) {
+ throw new BusinessException(
+ 400, "startTime and endTime must be integer
epoch-milliseconds");
+ }
+ }
+
+ /**
+ * Convert a caller-supplied numeric timestamp to epoch-milliseconds
without silent
+ * overflow. Jackson may deliver out-of-range values as {@link BigInteger}
(or as a
+ * floating-point number); {@code Number.longValue()} would silently wrap
these into a
+ * nonsense, possibly negative, timestamp. Reject values that do not fit
instead.
+ */
+ private static long toEpochMillis(Number number) {
+ if (number instanceof BigInteger bigInteger) {
+ try {
+ return bigInteger.longValueExact();
+ } catch (ArithmeticException ex) {
+ throw new BusinessException(
+ 400, "startTime and endTime must fit in a 64-bit
epoch-milliseconds value");
+ }
+ }
+ if (number instanceof Double || number instanceof Float) {
+ double doubleValue = number.doubleValue();
+ if (!Double.isFinite(doubleValue) || doubleValue > Long.MAX_VALUE
+ || doubleValue < Long.MIN_VALUE) {
+ throw new BusinessException(
+ 400, "startTime and endTime must fit in a 64-bit
epoch-milliseconds value");
+ }
+ }
+ return number.longValue();
}
private static String require(String value, String field) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandlerTest.java
new file mode 100644
index 000000000..d8362e576
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/CapabilitiesToolHandlerTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.ops.ai.tool;
+
+import org.apache.rocketmq.studio.cluster.broker.ClusterService;
+import org.apache.rocketmq.studio.cluster.broker.ClusterVO;
+import org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
+import org.apache.rocketmq.studio.common.domain.enums.ClusterType;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class CapabilitiesToolHandlerTest {
+
+ @Test
+ void nullClusterVersionIsEmittedAsBlankString() {
+ // The Apache runtime provider reports no cluster version; the
projection must not
+ // emit a null into the schema-required "version" string.
+ ClusterVO cluster = ClusterVO.builder()
+ .name("DefaultCluster")
+ .type(ClusterType.V4_DIRECT)
+ .status(ClusterStatus.healthy)
+ .build();
+ cluster.setId("DefaultCluster");
+
+ ClusterService clusterService = mock(ClusterService.class);
+ when(clusterService.getCluster("DefaultCluster")).thenReturn(cluster);
+ CapabilityResolver capabilityResolver = mock(CapabilityResolver.class);
+
when(capabilityResolver.resolve(cluster)).thenReturn(List.of("REMOTING"));
+
+ Object output = new CapabilitiesToolHandler(clusterService,
capabilityResolver)
+ .execute(Map.of("cluster", "DefaultCluster"));
+
+ @SuppressWarnings("unchecked")
+ Map<String, Object> result = (Map<String, Object>) output;
+ assertThat(result.get("cluster")).isEqualTo("DefaultCluster");
+ assertThat(result.get("type")).isEqualTo("V4_DIRECT");
+ assertThat(result.get("version")).isEqualTo("");
+ assertThat(result.get("capabilities")).isEqualTo(List.of("REMOTING"));
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ClusterListToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ClusterListToolHandlerTest.java
new file mode 100644
index 000000000..4186f19a7
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ClusterListToolHandlerTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.ops.ai.tool;
+
+import org.apache.rocketmq.studio.cluster.broker.ClusterService;
+import org.apache.rocketmq.studio.cluster.broker.ClusterVO;
+import org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
+import org.apache.rocketmq.studio.common.domain.enums.ClusterType;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class ClusterListToolHandlerTest {
+
+ @Test
+ void nullClusterVersionIsEmittedAsBlankString() {
+ // The Apache runtime provider reports no cluster version; the
projection must not
+ // emit a null into the schema-required "version" string.
+ ClusterVO cluster = ClusterVO.builder()
+ .name("DefaultCluster")
+ .type(ClusterType.V4_DIRECT)
+ .status(ClusterStatus.healthy)
+ .build();
+ cluster.setId("DefaultCluster");
+
+ ClusterService clusterService = mock(ClusterService.class);
+ when(clusterService.listClusters()).thenReturn(List.of(cluster));
+
+ Object output = new
ClusterListToolHandler(clusterService).execute(Map.of());
+
+ @SuppressWarnings("unchecked")
+ List<Map<String, Object>> rows = (List<Map<String, Object>>) output;
+ assertThat(rows).hasSize(1);
+ Map<String, Object> row = rows.get(0);
+ assertThat(row.get("id")).isEqualTo("DefaultCluster");
+ assertThat(row.get("name")).isEqualTo("DefaultCluster");
+ assertThat(row.get("type")).isEqualTo("V4_DIRECT");
+ assertThat(row.get("status")).isEqualTo("healthy");
+ assertThat(row.get("version")).isEqualTo("");
+ }
+
+ @Test
+ void populatedClusterVersionIsPassedThrough() {
+ ClusterVO cluster = ClusterVO.builder()
+ .name("VersionedCluster")
+ .type(ClusterType.V4_DIRECT)
+ .status(ClusterStatus.healthy)
+ .version("V5_3_1")
+ .build();
+ cluster.setId("VersionedCluster");
+
+ ClusterService clusterService = mock(ClusterService.class);
+ when(clusterService.listClusters()).thenReturn(List.of(cluster));
+
+ Object output = new
ClusterListToolHandler(clusterService).execute(Map.of());
+
+ @SuppressWarnings("unchecked")
+ List<Map<String, Object>> rows = (List<Map<String, Object>>) output;
+ @SuppressWarnings("unchecked")
+ Map<String, Object> row = rows.get(0);
+ assertThat(row.get("version")).isEqualTo("V5_3_1");
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/MessageQueryToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/MessageQueryToolHandlerTest.java
index 067ca8c89..f9b7cd0d8 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/MessageQueryToolHandlerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/MessageQueryToolHandlerTest.java
@@ -85,4 +85,49 @@ class MessageQueryToolHandlerTest {
verify(messageService)
.queryMessages(eq("instance-a"), any(), any(), any(), any(),
eq(1000L), eq(2000L));
}
+
+ @Test
+ void executeShouldRejectTimestampAboveLongRangeInsteadOfSilentlyWrapping()
{
+ java.math.BigInteger overflow =
+
java.math.BigInteger.valueOf(Long.MAX_VALUE).add(java.math.BigInteger.ONE);
+
+ org.assertj.core.api.Assertions.assertThatThrownBy(
+ () -> handler.execute(Map.of(
+ "cluster", "instance-a", "topic", "TopicA",
"startTime", overflow)))
+
.isInstanceOf(org.apache.rocketmq.studio.common.exception.BusinessException.class)
+ .hasMessageContaining("epoch-milliseconds");
+ }
+
+ @Test
+ void executeShouldRejectNonFiniteTimestampInsteadOfWrapping() {
+ org.assertj.core.api.Assertions.assertThatThrownBy(
+ () -> handler.execute(Map.of(
+ "cluster", "instance-a", "topic", "TopicA",
+ "startTime", Double.POSITIVE_INFINITY)))
+
.isInstanceOf(org.apache.rocketmq.studio.common.exception.BusinessException.class)
+ .hasMessageContaining("epoch-milliseconds");
+ }
+
+ @Test
+ void executeShouldRejectNonNumericTimestamp() {
+ org.assertj.core.api.Assertions.assertThatThrownBy(
+ () -> handler.execute(Map.of(
+ "cluster", "instance-a", "topic", "TopicA",
+ "startTime", "not-a-number")))
+
.isInstanceOf(org.apache.rocketmq.studio.common.exception.BusinessException.class)
+ .hasMessageContaining("epoch-milliseconds");
+ }
+
+ @Test
+ void executeShouldConvertInBigIntegerTimestampExactly() {
+ when(messageService.queryMessages(any(), any(), any(), any(), any(),
any(), any()))
+ .thenReturn(List.of());
+
+ handler.execute(Map.of("cluster", "instance-a", "topic", "TopicA",
+ "startTime", java.math.BigInteger.valueOf(123456789L)));
+
+ verify(messageService)
+ .queryMessages(eq("instance-a"), any(), any(), any(), any(),
+ eq(123456789L), any());
+ }
}
