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 b88582b12 [ISSUE #4143][ISSUE #4147] feat(ai): bound message queries
and support custom traces (#4388)
b88582b12 is described below
commit b88582b12a6e93ef7baa2e0ccb9797f3a974c0ac
Author: aias00 <[email protected]>
AuthorDate: Mon Sep 21 05:59:01 2026 -0700
[ISSUE #4143][ISSUE #4147] feat(ai): bound message queries and support
custom traces (#4388)
`rmq.message.query` and `rmq.message.query_by_topic` returned
`ListOutput<MessageItem>`, an unbounded list whose items always carried `body`
and `bodyEncoding`. A single topic scan could therefore put every message the
provider found, payloads included, into one MCP response, while
`MessageService.queryMessagesPage` — which already returns a bounded page with
a total and a truncation flag — went unused by both tools.
Both handlers now project into a new `MessageQueryOutput` record of
`items`, `resultMayBeTruncated` and `skippedCount`, backed by
`queryMessagesPage`. Their inputs gain `limit` (default 20, capped at 100 by
`resultLimit()`, with `minimum: 1` enforced by the runtime schema validator)
and `includeBody`, which defaults to false and leaves `body`, `bodyEncoding`
and `bodyTruncated` off the item entirely rather than emitting nulls. The
unique-key path keeps calling `queryMessageByUniqueKey [...]
`includeBody` defaulting to false is a breaking output change for existing
callers, so both catalog descriptions say so explicitly, and
`rmq.message.query_by_offset` documents that it still returns bodies
unconditionally.
Maintainer edits on top of the contribution:
`rmqctl/internal/catalog/catalog_gen.go` was regenerated again after the rebase
onto trunk, which had picked up another `tool-catalog` change (#4544's
`group.yaml` sentinel) in the meantime; the committed digest covers both.
Fixes #4143
Fixes #4147
---
rmqctl/internal/catalog/catalog_gen.go | 13 +-
.../contract/message/MessageQueryByTopicInput.java | 16 +-
.../tool/contract/message/MessageQueryInput.java | 16 +-
.../tool/contract/message/MessageQueryOutput.java | 81 ++++++++++
.../tool/contract/message/MessageTraceInput.java | 7 +-
.../message/MessageQueryByTopicToolHandler.java | 16 +-
.../handler/message/MessageQueryToolHandler.java | 28 ++--
.../handler/message/MessageTraceToolHandler.java | 8 +-
.../main/resources/tool-catalog/tools/message.yaml | 41 ++++-
.../ops/ai/tool/MessageTraceToolHandlerTest.java | 32 ++++
.../ops/ai/tool/catalog/ToolCatalogTest.java | 62 ++++++++
.../MessageQueryByTopicToolHandlerTest.java | 98 +++++++++---
.../message/MessageQueryToolHandlerTest.java | 171 +++++++++++++--------
.../tool/service/ToolOutputSchemaContractTest.java | 15 +-
14 files changed, 474 insertions(+), 130 deletions(-)
diff --git a/rmqctl/internal/catalog/catalog_gen.go
b/rmqctl/internal/catalog/catalog_gen.go
index 35db62557..622d4a59d 100644
--- a/rmqctl/internal/catalog/catalog_gen.go
+++ b/rmqctl/internal/catalog/catalog_gen.go
@@ -21,7 +21,7 @@ package catalog
var defaultDocument = Document{
Version: "2.0.0",
MinimumClientVersion: "2.0.0",
- Digest:
"71c2cc225321b7e83243ef77b7330f25800e9623c0a110f56b27a6834a21735e",
+ Digest:
"94c2164d0181f55aebc2f32ef1f3db2c0b92185e98ac7824e492b6f0e29014b9",
Tools: []Tool{
{
Name: "rmq.acl.list",
@@ -372,7 +372,7 @@ var defaultDocument = Document{
{
Name: "rmq.message.query",
CLI: CLI{Resource: "message", Verb:
"query"},
- Description: "Query RocketMQ messages by
identifier: the first non-empty value among msgId, uniqueKey and key (in this
order) selects the query path, and at least one of them must be provided. The
startTime/endTime window only applies to the uniqueKey and key paths.",
+ Description: "Query RocketMQ messages by
identifier: the first non-empty value among msgId, uniqueKey and key (in this
order) selects the query path, and at least one of them must be provided. The
startTime/endTime window only applies to the uniqueKey and key paths. Results
are limited to 20 by default; values above the server-side cap of 100 are
reduced to 100. Breaking output change: message bodies are omitted by default,
so callers that read items[].body must set includeBo [...]
RiskLevel: "L1",
Permission: "message:read",
RequiredCapabilities: []string{"MESSAGE_QUERY"},
@@ -385,6 +385,8 @@ var defaultDocument = Document{
{Name: "key", Flag: "key", Description:
"Business key (key index, 0..n messages).", Kind: StringField, MinLength: 1},
{Name: "startTime", Flag: "start-time",
Description: "Epoch milliseconds; only applies to the uniqueKey and key
paths.", Kind: IntegerField},
{Name: "endTime", Flag: "end-time",
Description: "Epoch milliseconds; only applies to the uniqueKey and key
paths.", Kind: IntegerField},
+ {Name: "limit", Flag: "limit",
Description: "Maximum rows to return; defaults to 20 and values above 100 are
capped at 100.", Kind: IntegerField, Minimum: 1, HasMinimum: true},
+ {Name: "includeBody", Flag:
"include-body", Description: "Include body, bodyEncoding and bodyTruncated;
defaults to false.", Kind: BooleanField},
},
},
ViewHint: "table",
@@ -393,7 +395,7 @@ var defaultDocument = Document{
{
Name: "rmq.message.query_by_topic",
CLI: CLI{Resource: "message", Verb:
"query-by-topic"},
- Description: "Query RocketMQ messages by topic
and optional time range in one Studio Instance.",
+ Description: "Query RocketMQ messages by topic
and optional time range in one Studio Instance. Results are limited to 20 by
default; values above the server-side cap of 100 are reduced to 100. Breaking
output change: message bodies are omitted by default, so callers that read
items[].body must set includeBody=true.",
RiskLevel: "L1",
Permission: "message:read",
RequiredCapabilities: []string{"MESSAGE_QUERY"},
@@ -404,6 +406,8 @@ var defaultDocument = Document{
{Name: "tag", Flag: "tag", Kind:
StringField},
{Name: "startTime", Flag: "start-time",
Kind: IntegerField},
{Name: "endTime", Flag: "end-time",
Kind: IntegerField},
+ {Name: "limit", Flag: "limit",
Description: "Maximum rows to return; defaults to 20 and values above 100 are
capped at 100.", Kind: IntegerField, Minimum: 1, HasMinimum: true},
+ {Name: "includeBody", Flag:
"include-body", Description: "Include body, bodyEncoding and bodyTruncated;
defaults to false.", Kind: BooleanField},
},
},
ViewHint: "table",
@@ -412,7 +416,7 @@ var defaultDocument = Document{
{
Name: "rmq.message.query_by_offset",
CLI: CLI{Resource: "message", Verb:
"query-by-offset"},
- Description: "Fetch the message stored at one
physical queue offset (admin queryMsgByOffset); returns empty items when the
offset is out of range or the CommitLog entry was already cleaned. Pair it with
the queueStats block of rmq.topic.detail.",
+ Description: "Fetch the message stored at one
physical queue offset (admin queryMsgByOffset); returns empty items when the
offset is out of range or the CommitLog entry was already cleaned. Unlike
rmq.message.query and rmq.message.query_by_topic, this single-message tool
always includes body, bodyEncoding and bodyTruncated and has no includeBody
flag. Pair it with the queueStats block of rmq.topic.detail.",
RiskLevel: "L1",
Permission: "message:read",
RequiredCapabilities: []string{"MESSAGE_QUERY"},
@@ -469,6 +473,7 @@ var defaultDocument = Document{
{Name: "instanceId", Flag:
"instance-id", Description: "Studio Instance identifier.", Kind: StringField,
Required: true, MinLength: 1},
{Name: "topicName", Flag: "topic-name",
Kind: StringField},
{Name: "msgId", Flag: "msg-id", Kind:
StringField, Required: true, MinLength: 1},
+ {Name: "traceTopicName", Flag:
"trace-topic-name", Description: "Optional custom trace Topic name; blank or
absent uses provider defaults.", Kind: StringField},
},
},
ViewHint: "object",
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryByTopicInput.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryByTopicInput.java
index 2b4a31b72..8ed6202a2 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryByTopicInput.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryByTopicInput.java
@@ -21,5 +21,19 @@ public record MessageQueryByTopicInput(
String topicName,
String tag,
Long startTime,
- Long endTime) {
+ Long endTime,
+ Integer limit,
+ boolean includeBody) {
+
+ private static final int DEFAULT_LIMIT = 20;
+ private static final int MAX_LIMIT = 100;
+
+ public MessageQueryByTopicInput(String instanceId, String topicName,
String tag,
+ Long startTime, Long endTime) {
+ this(instanceId, topicName, tag, startTime, endTime, null, false);
+ }
+
+ public int resultLimit() {
+ return limit == null ? DEFAULT_LIMIT : Math.min(limit, MAX_LIMIT);
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryInput.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryInput.java
index 5d55003a3..098892c5b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryInput.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryInput.java
@@ -26,5 +26,19 @@ public record MessageQueryInput(
String uniqueKey,
String key,
Long startTime,
- Long endTime) {
+ Long endTime,
+ Integer limit,
+ boolean includeBody) {
+
+ private static final int DEFAULT_LIMIT = 20;
+ private static final int MAX_LIMIT = 100;
+
+ public MessageQueryInput(String instanceId, String topicName, String msgId,
+ String uniqueKey, String key, Long startTime,
Long endTime) {
+ this(instanceId, topicName, msgId, uniqueKey, key, startTime, endTime,
null, false);
+ }
+
+ public int resultLimit() {
+ return limit == null ? DEFAULT_LIMIT : Math.min(limit, MAX_LIMIT);
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryOutput.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryOutput.java
new file mode 100644
index 000000000..cb3c33063
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageQueryOutput.java
@@ -0,0 +1,81 @@
+/*
+ * 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.contract.message;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.rocketmq.studio.instance.message.MessageQueryPageVO;
+import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
+
+import java.util.List;
+
+public record MessageQueryOutput(
+ List<Item> items,
+ boolean resultMayBeTruncated,
+ long skippedCount) {
+
+ public static MessageQueryOutput fromPage(MessageQueryPageVO result,
boolean includeBody) {
+ long skippedCount = Math.max(0, result.getTotal() -
result.getItems().size());
+ return new MessageQueryOutput(
+ result.getItems().stream()
+ .map(message -> Item.from(message, includeBody))
+ .toList(),
+ result.isResultMayBeTruncated() || skippedCount > 0,
+ skippedCount);
+ }
+
+ public static MessageQueryOutput fromUniqueKey(
+ List<MessageRecordVO> messages, int limit, boolean includeBody) {
+ int to = Math.min(limit, messages.size());
+ long skippedCount = messages.size() - to;
+ return new MessageQueryOutput(
+ messages.subList(0, to).stream()
+ .map(message -> Item.from(message, includeBody))
+ .toList(),
+ skippedCount > 0,
+ skippedCount);
+ }
+
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public record Item(
+ String msgId,
+ String topic,
+ String tag,
+ String key,
+ long storeTime,
+ String storeHost,
+ String bornHost,
+ String body,
+ String bodyEncoding,
+ Boolean bodyTruncated,
+ int size) {
+
+ static Item from(MessageRecordVO message, boolean includeBody) {
+ return new Item(
+ message.getMsgId(),
+ message.getTopic(),
+ message.getTag(),
+ message.getKey(),
+ message.getStoreTime(),
+ message.getStoreHost(),
+ message.getBornHost(),
+ includeBody ? message.getBody() : null,
+ includeBody ? message.getBodyEncoding() : null,
+ includeBody ? message.isBodyTruncated() : null,
+ message.getSize());
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageTraceInput.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageTraceInput.java
index 6207c2ac9..81b3a8ea4 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageTraceInput.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageTraceInput.java
@@ -19,5 +19,10 @@ package
org.apache.rocketmq.studio.ops.ai.tool.contract.message;
public record MessageTraceInput(
String instanceId,
String topicName,
- String msgId) {
+ String msgId,
+ String traceTopicName) {
+
+ public MessageTraceInput(String instanceId, String topicName, String
msgId) {
+ this(instanceId, topicName, msgId, null);
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryByTopicToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryByTopicToolHandler.java
index fda630c59..d3d369230 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryByTopicToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryByTopicToolHandler.java
@@ -17,11 +17,10 @@
package org.apache.rocketmq.studio.ops.ai.tool.handler.message;
import org.apache.rocketmq.studio.instance.message.MessageService;
-import org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageItem;
import org.apache.rocketmq.studio.ops.ai.tool.core.ToolExecutionContext;
import org.apache.rocketmq.studio.ops.ai.tool.core.ToolHandler;
-import org.apache.rocketmq.studio.ops.ai.tool.contract.common.ListOutput;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageQueryByTopicInput;
+import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageQueryOutput;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
@@ -32,7 +31,7 @@ import org.springframework.stereotype.Component;
@Component
@RequiredArgsConstructor
public class MessageQueryByTopicToolHandler
- implements ToolHandler<MessageQueryByTopicInput,
ListOutput<MessageItem>> {
+ implements ToolHandler<MessageQueryByTopicInput, MessageQueryOutput> {
private final MessageService messageService;
@@ -47,13 +46,10 @@ public class MessageQueryByTopicToolHandler
}
@Override
- public ListOutput<MessageItem> execute(
+ public MessageQueryOutput execute(
MessageQueryByTopicInput input, ToolExecutionContext context) {
- return new ListOutput<>(messageService.queryMessages(
- context.instanceId(), input.topicName(), null,
input.tag(), null,
- input.startTime(), input.endTime())
- .stream()
- .map(MessageItem::from)
- .toList());
+ return MessageQueryOutput.fromPage(messageService.queryMessagesPage(
+ context.instanceId(), input.topicName(), null, input.tag(),
null,
+ input.startTime(), input.endTime(), 1, input.resultLimit()),
input.includeBody());
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryToolHandler.java
index 2de366a54..d918c0998 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryToolHandler.java
@@ -17,19 +17,15 @@
package org.apache.rocketmq.studio.ops.ai.tool.handler.message;
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 org.apache.rocketmq.studio.ops.ai.tool.contract.common.ListOutput;
-import org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageItem;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageQueryInput;
+import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageQueryOutput;
import org.apache.rocketmq.studio.ops.ai.tool.core.ToolExecutionContext;
import org.apache.rocketmq.studio.ops.ai.tool.core.ToolHandler;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
-import java.util.List;
-
/**
* Unified message query: the first non-empty identifier among msgId,
uniqueKey and key (in this
* order) selects the query path; the time window only applies to the
uniqueKey and key paths.
@@ -37,7 +33,7 @@ import java.util.List;
@Component
@RequiredArgsConstructor
public class MessageQueryToolHandler
- implements ToolHandler<MessageQueryInput, ListOutput<MessageItem>> {
+ implements ToolHandler<MessageQueryInput, MessageQueryOutput> {
private final MessageService messageService;
@@ -52,22 +48,24 @@ public class MessageQueryToolHandler
}
@Override
- public ListOutput<MessageItem> execute(MessageQueryInput input,
ToolExecutionContext context) {
+ public MessageQueryOutput execute(MessageQueryInput input,
ToolExecutionContext context) {
String instanceId = context.instanceId();
- List<MessageRecordVO> messages;
+ int limit = input.resultLimit();
if (StringUtils.hasText(input.msgId())) {
- messages = messageService.queryMessages(
- instanceId, input.topicName(), input.msgId(), null, null,
null, null);
+ return
MessageQueryOutput.fromPage(messageService.queryMessagesPage(
+ instanceId, input.topicName(), input.msgId(), null, null,
null, null,
+ 1, limit), input.includeBody());
} else if (StringUtils.hasText(input.uniqueKey())) {
- messages = messageService.queryMessageByUniqueKey(
- instanceId, input.topicName(), input.uniqueKey(),
input.startTime(), input.endTime());
+ return
MessageQueryOutput.fromUniqueKey(messageService.queryMessageByUniqueKey(
+ instanceId, input.topicName(), input.uniqueKey(),
input.startTime(), input.endTime()),
+ limit, input.includeBody());
} else if (StringUtils.hasText(input.key())) {
- messages = messageService.queryMessages(
+ return
MessageQueryOutput.fromPage(messageService.queryMessagesPage(
instanceId, input.topicName(), null, null, input.key(),
- input.startTime(), input.endTime());
+ input.startTime(), input.endTime(), 1, limit),
+ input.includeBody());
} else {
throw new BusinessException(400, "message query requires one of:
msgId, uniqueKey, key");
}
- return new
ListOutput<>(messages.stream().map(MessageItem::from).toList());
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageTraceToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageTraceToolHandler.java
index b5ba91b40..9b7cdea33 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageTraceToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageTraceToolHandler.java
@@ -26,6 +26,7 @@ import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageTraceOutpu
import org.apache.rocketmq.studio.ops.ai.tool.core.ToolExecutionContext;
import org.apache.rocketmq.studio.ops.ai.tool.core.ToolHandler;
import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
@Component
@RequiredArgsConstructor
@@ -48,8 +49,11 @@ public class MessageTraceToolHandler
public MessageTraceOutput execute(
MessageTraceInput input,
ToolExecutionContext context) {
- TraceRecordVO trace = messageService.getMessageTrace(
- context.instanceId(), input.msgId(), input.topicName());
+ TraceRecordVO trace = StringUtils.hasText(input.traceTopicName())
+ ? messageService.getMessageTrace(
+ context.instanceId(), input.msgId(),
input.topicName(), input.traceTopicName())
+ : messageService.getMessageTrace(
+ context.instanceId(), input.msgId(),
input.topicName());
return project(input.msgId(), trace);
}
diff --git a/server/src/main/resources/tool-catalog/tools/message.yaml
b/server/src/main/resources/tool-catalog/tools/message.yaml
index 7550527a7..2da865a26 100644
--- a/server/src/main/resources/tool-catalog/tools/message.yaml
+++ b/server/src/main/resources/tool-catalog/tools/message.yaml
@@ -4,7 +4,7 @@ tools:
cli:
resource: message
verb: query
- description: "Query RocketMQ messages by identifier: the first non-empty
value among msgId, uniqueKey and key (in this order) selects the query path,
and at least one of them must be provided. The startTime/endTime window only
applies to the uniqueKey and key paths."
+ description: "Query RocketMQ messages by identifier: the first non-empty
value among msgId, uniqueKey and key (in this order) selects the query path,
and at least one of them must be provided. The startTime/endTime window only
applies to the uniqueKey and key paths. Results are limited to 20 by default;
values above the server-side cap of 100 are reduced to 100. Breaking output
change: message bodies are omitted by default, so callers that read
items[].body must set includeBody=true."
riskLevel: L1
permission: message:read
requiredCapabilities:
@@ -41,10 +41,19 @@ tools:
endTime:
type: integer
description: Epoch milliseconds; only applies to the uniqueKey and
key paths.
+ limit:
+ type: integer
+ description: Maximum rows to return; defaults to 20 and values above
100 are capped at 100.
+ minimum: 1
+ includeBody:
+ type: boolean
+ description: Include body, bodyEncoding and bodyTruncated; defaults
to false.
outputSchema:
type: object
required:
- items
+ - resultMayBeTruncated
+ - skippedCount
additionalProperties: false
properties:
items:
@@ -55,7 +64,6 @@ tools:
- msgId
- topic
- storeTime
- - bodyTruncated
- size
additionalProperties: false
properties:
@@ -81,13 +89,19 @@ tools:
type: boolean
size:
type: integer
+ resultMayBeTruncated:
+ type: boolean
+ description: True when rows were omitted by limit or the provider
may have stopped at its own scan bound.
+ skippedCount:
+ type: integer
+ description: Rows omitted by limit from the provider-bounded result;
more may exist when resultMayBeTruncated is true.
viewHint: table
deprecated: false
- name: rmq.message.query_by_topic
cli:
resource: message
verb: query-by-topic
- description: Query RocketMQ messages by topic and optional time range in
one Studio Instance.
+ description: "Query RocketMQ messages by topic and optional time range in
one Studio Instance. Results are limited to 20 by default; values above the
server-side cap of 100 are reduced to 100. Breaking output change: message
bodies are omitted by default, so callers that read items[].body must set
includeBody=true."
riskLevel: L1
permission: message:read
requiredCapabilities:
@@ -112,10 +126,19 @@ tools:
type: integer
endTime:
type: integer
+ limit:
+ type: integer
+ description: Maximum rows to return; defaults to 20 and values above
100 are capped at 100.
+ minimum: 1
+ includeBody:
+ type: boolean
+ description: Include body, bodyEncoding and bodyTruncated; defaults
to false.
outputSchema:
type: object
required:
- items
+ - resultMayBeTruncated
+ - skippedCount
additionalProperties: false
properties:
items:
@@ -126,7 +149,6 @@ tools:
- msgId
- topic
- storeTime
- - bodyTruncated
- size
additionalProperties: false
properties:
@@ -152,13 +174,19 @@ tools:
type: boolean
size:
type: integer
+ resultMayBeTruncated:
+ type: boolean
+ description: True when rows were omitted by limit or the provider
may have stopped at its own scan bound.
+ skippedCount:
+ type: integer
+ description: Rows omitted by limit from the provider-bounded result;
more may exist when resultMayBeTruncated is true.
viewHint: table
deprecated: false
- name: rmq.message.query_by_offset
cli:
resource: message
verb: query-by-offset
- description: "Fetch the message stored at one physical queue offset (admin
queryMsgByOffset); returns empty items when the offset is out of range or the
CommitLog entry was already cleaned. Pair it with the queueStats block of
rmq.topic.detail."
+ description: "Fetch the message stored at one physical queue offset (admin
queryMsgByOffset); returns empty items when the offset is out of range or the
CommitLog entry was already cleaned. Unlike rmq.message.query and
rmq.message.query_by_topic, this single-message tool always includes body,
bodyEncoding and bodyTruncated and has no includeBody flag. Pair it with the
queueStats block of rmq.topic.detail."
riskLevel: L1
permission: message:read
requiredCapabilities:
@@ -351,6 +379,9 @@ tools:
msgId:
type: string
minLength: 1
+ traceTopicName:
+ type: string
+ description: Optional custom trace Topic name; blank or absent uses
provider defaults.
outputSchema:
type: object
required:
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/MessageTraceToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/MessageTraceToolHandlerTest.java
index 6453633b6..a560a26f3 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/MessageTraceToolHandlerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/MessageTraceToolHandlerTest.java
@@ -84,4 +84,36 @@ class MessageTraceToolHandlerTest {
verify(messageService).getMessageTrace("instance-a", "msg-1",
"TopicA");
}
+
+ @Test
+ void customTraceTopicNameUsesTheExplicitServicePathTest() {
+ TraceRecordVO trace = TraceRecordVO.builder()
+ .nodes(List.of())
+ .consumerStatus(List.of())
+ .build();
+ when(messageService.getMessageTrace(
+ "instance-a", "msg-1", "TopicA", "CustomTraceTopic"))
+ .thenReturn(trace);
+
+ handler.execute(new MessageTraceInput(
+ "instance-a", "TopicA", "msg-1", "CustomTraceTopic"),
context("instance-a"));
+
+ verify(messageService).getMessageTrace(
+ "instance-a", "msg-1", "TopicA", "CustomTraceTopic");
+ }
+
+ @Test
+ void blankTraceTopicNameKeepsTheDefaultServicePathTest() {
+ TraceRecordVO trace = TraceRecordVO.builder()
+ .nodes(List.of())
+ .consumerStatus(List.of())
+ .build();
+ when(messageService.getMessageTrace("instance-a", "msg-1", "TopicA"))
+ .thenReturn(trace);
+
+ handler.execute(new MessageTraceInput("instance-a", "TopicA", "msg-1",
" "),
+ context("instance-a"));
+
+ verify(messageService).getMessageTrace("instance-a", "msg-1",
"TopicA");
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/catalog/ToolCatalogTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/catalog/ToolCatalogTest.java
index ac3d4815b..f6178e71b 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/catalog/ToolCatalogTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/catalog/ToolCatalogTest.java
@@ -16,14 +16,20 @@
*/
package org.apache.rocketmq.studio.ops.ai.tool.catalog;
+import org.apache.rocketmq.studio.common.config.LegacyJackson2Config;
import org.apache.rocketmq.studio.ops.ai.tool.core.ToolDefinition;
+import org.apache.rocketmq.studio.ops.ai.tool.core.ToolExecutionException;
import org.apache.rocketmq.studio.ops.ai.tool.core.ToolRiskLevel;
+import org.apache.rocketmq.studio.ops.ai.tool.service.ToolSchemaValidator;
import org.junit.jupiter.api.Test;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.DefaultResourceLoader;
import org.springframework.core.io.Resource;
import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import tools.jackson.databind.json.JsonMapper;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -45,6 +51,62 @@ class ToolCatalogTest {
assertThat(catalog.find("rmq.unknown")).isEmpty();
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void messageQueriesAdvertiseBoundedResultsAndOptionalBodiesTest() {
+ ToolCatalog catalog = new ToolCatalog(new DefaultResourceLoader());
+ for (String toolName : new String[]{"rmq.message.query",
"rmq.message.query_by_topic"}) {
+ ToolDefinition definition = catalog.getDefinition(toolName);
+ Map<String, Object> inputProperties =
+ (Map<String, Object>)
definition.inputSchema().get("properties");
+ Map<String, Object> outputProperties =
+ (Map<String, Object>)
definition.outputSchema().get("properties");
+
+ assertThat(inputProperties)
+ .containsKeys("limit", "includeBody")
+ .doesNotContainKeys("page", "pageSize");
+ assertThat((Map<String, Object>) inputProperties.get("limit"))
+ .containsEntry("minimum", 1);
+ assertThat(outputProperties)
+ .containsKeys("items", "resultMayBeTruncated",
"skippedCount")
+ .doesNotContainKeys("total", "page", "pageSize", "size");
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void messageTraceAdvertisesOptionalCustomTraceTopicNameTest() {
+ Map<String, Object> inputSchema = new ToolCatalog(new
DefaultResourceLoader())
+ .getDefinition("rmq.message.trace").inputSchema();
+ Map<String, Object> properties = (Map<String, Object>)
inputSchema.get("properties");
+
+
assertThat(properties).containsKey("traceTopicName").doesNotContainKey("traceTopic");
+ assertThat((Map<String, Object>) properties.get("traceTopicName"))
+ .containsEntry("type", "string");
+ assertThat((List<String>) inputSchema.get("required"))
+ .doesNotContain("traceTopicName");
+ }
+
+ @Test
+ void
rejectsInvalidMessageQueryLimitsAndLegacyPagingThroughTheRuntimeSchemaTest() {
+ ToolCatalog catalog = new ToolCatalog(new DefaultResourceLoader());
+ ToolSchemaValidator validator = new ToolSchemaValidator(catalog,
+ new LegacyJackson2Config().jackson2ObjectMapper(),
JsonMapper.builder().build());
+ for (String toolName : new String[]{"rmq.message.query",
"rmq.message.query_by_topic"}) {
+ ToolDefinition definition = catalog.getDefinition(toolName);
+ for (Map<String, Object> invalid : List.<Map<String, Object>>of(
+ Map.of("instanceId", "instance-a", "topicName", "TopicA",
+ "limit", 0),
+ Map.of("instanceId", "instance-a", "topicName", "TopicA",
+ "page", Map.of("page", 1, "pageSize", 10)),
+ Map.of("instanceId", "instance-a", "topicName", "TopicA",
"pageSize", 10))) {
+ assertThatThrownBy(() -> validator.validateInput(definition,
invalid))
+ .isInstanceOfSatisfying(ToolExecutionException.class,
+ error ->
assertThat(error.getCode()).isEqualTo(400));
+ }
+ }
+ }
+
@Test
void rejectsCatalogThatDoesNotMatchItsJsonSchema() {
Resource invalid = utf8Resource("""
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryByTopicToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryByTopicToolHandlerTest.java
index 7e59e36dd..b06835576 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryByTopicToolHandlerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryByTopicToolHandlerTest.java
@@ -16,10 +16,11 @@
*/
package org.apache.rocketmq.studio.ops.ai.tool.handler.message;
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.rocketmq.studio.common.config.LegacyJackson2Config;
+import org.apache.rocketmq.studio.instance.message.MessageQueryPageVO;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
import org.apache.rocketmq.studio.instance.message.MessageService;
-import org.apache.rocketmq.studio.ops.ai.tool.contract.common.ListOutput;
-import org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageItem;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageQueryByTopicInput;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -27,15 +28,16 @@ import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
+import static
org.apache.rocketmq.studio.ops.ai.tool.TestToolExecutionContexts.context;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import static
org.apache.rocketmq.studio.ops.ai.tool.TestToolExecutionContexts.context;
@ExtendWith(MockitoExtension.class)
class MessageQueryByTopicToolHandlerTest {
@@ -47,36 +49,82 @@ class MessageQueryByTopicToolHandlerTest {
private MessageQueryByTopicToolHandler handler;
@Test
- void executeShouldDelegateToMessageServiceAndProject() {
- MessageRecordVO message = MessageRecordVO.builder()
- .msgId("msg-1")
- .topic("TopicA")
- .storeTime(1000L)
- .size(5)
- .build();
- when(messageService.queryMessages(eq("instance-a"), eq("TopicA"),
isNull(), isNull(), isNull(), any(), any()))
- .thenReturn(List.of(message));
+ void executeUsesDefaultLimitAndReportsSkippedRowsTest() {
+ MessageRecordVO message = message("msg-1", null, false);
+ when(messageService.queryMessagesPage(eq("instance-a"), eq("TopicA"),
isNull(),
+ isNull(), isNull(), isNull(), isNull(), eq(1), eq(20)))
+ .thenReturn(page(Collections.nCopies(20, message), 25, 20,
false));
- ListOutput<MessageItem> result = handler.execute(
+ var result = handler.execute(
new MessageQueryByTopicInput("instance-a", "TopicA", null,
null, null),
context("instance-a"));
- assertThat(result.items()).hasSize(1);
- MessageItem row = result.items().getFirst();
- assertThat(row.msgId()).isEqualTo("msg-1");
- assertThat(row.topic()).isEqualTo("TopicA");
+ assertThat(result.items()).hasSize(20);
+ assertThat(result.skippedCount()).isEqualTo(5);
+ assertThat(result.resultMayBeTruncated()).isTrue();
+ assertThat(result.items().getFirst().body()).isNull();
+ Map<String, Object> serialized = new
LegacyJackson2Config().jackson2ObjectMapper()
+ .convertValue(result, new TypeReference<>() { });
+ assertThat(serialized)
+ .containsOnlyKeys("items", "resultMayBeTruncated",
"skippedCount");
}
@Test
- void executeShouldConvertNumericTimeArguments() {
- when(messageService.queryMessages(any(), any(), any(), any(), any(),
any(), any()))
- .thenReturn(List.of());
+ void executeUsesCustomLimitAndIncludesBodiesTest() {
+ MessageRecordVO message = message("msg-2", "hello", false);
+ when(messageService.queryMessagesPage("instance-a", "TopicA", null,
+ "TagA", null, 1000L, 2000L, 1, 10))
+ .thenReturn(page(List.of(message), 1, 10, false));
- handler.execute(
- new MessageQueryByTopicInput("instance-a", "TopicA", null,
1000L, 2000L),
+ var result = handler.execute(
+ new MessageQueryByTopicInput("instance-a", "TopicA", "TagA",
1000L, 2000L,
+ 10, true),
context("instance-a"));
- verify(messageService)
- .queryMessages(eq("instance-a"), eq("TopicA"), isNull(),
isNull(), isNull(), eq(1000L), eq(2000L));
+ verify(messageService).queryMessagesPage("instance-a", "TopicA", null,
+ "TagA", null, 1000L, 2000L, 1, 10);
+ assertThat(result.items()).singleElement().satisfies(row -> {
+ assertThat(row.body()).isEqualTo("hello");
+ assertThat(row.bodyEncoding()).isEqualTo("UTF-8");
+ assertThat(row.bodyTruncated()).isFalse();
+ });
+ assertThat(result.skippedCount()).isZero();
+ assertThat(result.resultMayBeTruncated()).isFalse();
+ }
+
+ @Test
+ void executeCapsRequestedLimitAtOneHundredTest() {
+ when(messageService.queryMessagesPage("instance-a", "TopicA", null,
+ null, null, null, null, 1, 100))
+ .thenReturn(page(List.of(), 0, 100, false));
+
+ handler.execute(new MessageQueryByTopicInput(
+ "instance-a", "TopicA", null, null, null, 500, false),
context("instance-a"));
+
+ verify(messageService).queryMessagesPage("instance-a", "TopicA", null,
+ null, null, null, null, 1, 100);
+ }
+
+ private MessageRecordVO message(String msgId, String body, boolean
bodyTruncated) {
+ return MessageRecordVO.builder()
+ .msgId(msgId)
+ .topic("TopicA")
+ .body(body)
+ .bodyEncoding(body == null ? null : "UTF-8")
+ .bodyTruncated(bodyTruncated)
+ .storeTime(1000L)
+ .size(body == null ? 0 : body.length())
+ .build();
+ }
+
+ private MessageQueryPageVO page(
+ List<MessageRecordVO> items, long total, int size, boolean
resultMayBeTruncated) {
+ return MessageQueryPageVO.builder()
+ .items(items)
+ .total(total)
+ .page(1)
+ .size(size)
+ .resultMayBeTruncated(resultMayBeTruncated)
+ .build();
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryToolHandlerTest.java
index 994bb49e3..ea4a5c6dd 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryToolHandlerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageQueryToolHandlerTest.java
@@ -16,12 +16,14 @@
*/
package org.apache.rocketmq.studio.ops.ai.tool.handler.message;
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.rocketmq.studio.common.config.LegacyJackson2Config;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.message.MessageQueryPageVO;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
import org.apache.rocketmq.studio.instance.message.MessageService;
-import org.apache.rocketmq.studio.ops.ai.tool.contract.common.ListOutput;
-import org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageItem;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageQueryInput;
+import org.apache.rocketmq.studio.ops.ai.tool.core.ToolExecutionContext;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
@@ -30,8 +32,11 @@ import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
+import static
org.apache.rocketmq.studio.ops.ai.tool.TestToolExecutionContexts.context;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.eq;
@@ -39,7 +44,6 @@ import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
-import static
org.apache.rocketmq.studio.ops.ai.tool.TestToolExecutionContexts.context;
@ExtendWith(MockitoExtension.class)
class MessageQueryToolHandlerTest {
@@ -57,84 +61,96 @@ class MessageQueryToolHandlerTest {
}
@Test
- void msgIdWinsOverUniqueKeyAndKeyTest() {
- MessageRecordVO message = MessageRecordVO.builder()
- .msgId("msg-1")
- .topic("TopicA")
- .tag("tag1")
- .key("key1")
- .body("hello")
- .bodyEncoding("UTF-8")
- .bodyTruncated(false)
- .storeTime(1000L)
- .bornHost("10.0.0.1")
- .storeHost("10.0.0.2")
- .size(5)
- .build();
- when(messageService.queryMessages(
+ void msgIdWinsOverOtherIdentifiersAndUsesDefaultLimitTest() {
+ MessageRecordVO message = message("msg-1", "hello", false);
+ when(messageService.queryMessagesPage(
eq("instance-a"), eq("TopicA"), eq("msg-1"),
- isNull(), isNull(), isNull(), isNull()))
- .thenReturn(List.of(message));
-
- ListOutput<MessageItem> result = handler.execute(
- new MessageQueryInput("instance-a", "TopicA", "msg-1",
"uniq-1", "order-123", 10L, 20L),
- context("instance-a"));
-
- assertThat(result.items()).hasSize(1);
- MessageItem row = result.items().getFirst();
- assertThat(row.msgId()).isEqualTo("msg-1");
- assertThat(row.topic()).isEqualTo("TopicA");
- assertThat(row.tag()).isEqualTo("tag1");
- assertThat(row.storeTime()).isEqualTo(1000L);
- assertThat(row.body()).isEqualTo("hello");
- assertThat(row.size()).isEqualTo(5);
- // The msgId path is a direct read: the time window never reaches the
service.
- verify(messageService).queryMessages(
- eq("instance-a"), eq("TopicA"), eq("msg-1"), isNull(),
isNull(), isNull(), isNull());
+ isNull(), isNull(), isNull(), isNull(), eq(1), eq(20)))
+ .thenReturn(page(List.of(message), 1, 20, false));
+
+ ToolExecutionContext request = ToolExecutionContext.of("instance-a",
null, Map.of(
+ "instanceId", "instance-a", "topicName", "TopicA", "msgId",
"msg-1",
+ "uniqueKey", "uniq-1", "key", "order-123", "startTime", 10L,
"endTime", 20L));
+ var result =
handler.execute(request.convertInput(handler.inputType()), request);
+
+ assertThat(result.items()).singleElement().satisfies(row -> {
+ assertThat(row.msgId()).isEqualTo("msg-1");
+ assertThat(row.topic()).isEqualTo("TopicA");
+ assertThat(row.body()).isNull();
+ assertThat(row.bodyEncoding()).isNull();
+ assertThat(row.bodyTruncated()).isNull();
+ });
+ assertThat(result.skippedCount()).isZero();
+ assertThat(result.resultMayBeTruncated()).isFalse();
+ Map<String, Object> serialized = new
LegacyJackson2Config().jackson2ObjectMapper()
+ .convertValue(result, new TypeReference<>() { });
+ assertThat(serialized)
+ .containsOnlyKeys("items", "resultMayBeTruncated",
"skippedCount");
+ Map<String, Object> item = new
LegacyJackson2Config().jackson2ObjectMapper()
+ .convertValue(((List<?>) serialized.get("items")).getFirst(),
new TypeReference<>() { });
+ assertThat(item).doesNotContainKeys("body", "bodyEncoding",
"bodyTruncated");
+ verify(messageService).queryMessagesPage(
+ eq("instance-a"), eq("TopicA"), eq("msg-1"), isNull(),
isNull(), isNull(), isNull(),
+ eq(1), eq(20));
}
@Test
void uniqueKeyPathPassesTimeWindowTest() {
- MessageRecordVO message = MessageRecordVO.builder()
- .msgId("msg-2")
- .topic("TopicA")
- .storeTime(2000L)
- .size(7)
- .build();
when(messageService.queryMessageByUniqueKey("instance-a", "TopicA",
"uniq-1", 1000L, 2000L))
- .thenReturn(List.of(message));
+ .thenReturn(List.of(message("msg-2", "private", false)));
- ListOutput<MessageItem> result = handler.execute(
- new MessageQueryInput("instance-a", "TopicA", null, "uniq-1",
"order-123", 1000L, 2000L),
+ var result = handler.execute(
+ new MessageQueryInput("instance-a", "TopicA", null, "uniq-1",
"order-123",
+ 1000L, 2000L),
context("instance-a"));
- assertThat(result.items()).hasSize(1);
- assertThat(result.items().getFirst().msgId()).isEqualTo("msg-2");
+ assertThat(result.items()).singleElement().satisfies(row -> {
+ assertThat(row.msgId()).isEqualTo("msg-2");
+ assertThat(row.body()).isNull();
+ });
+ assertThat(result.skippedCount()).isZero();
+ assertThat(result.resultMayBeTruncated()).isFalse();
verify(messageService).queryMessageByUniqueKey("instance-a", "TopicA",
"uniq-1", 1000L, 2000L);
}
@Test
- void keyPathPassesTimeWindowTest() {
- MessageRecordVO message = MessageRecordVO.builder()
- .msgId("msg-3")
- .topic("TopicA")
- .key("order-123")
- .storeTime(1000L)
- .size(5)
- .build();
- when(messageService.queryMessages(
- eq("instance-a"), eq("TopicA"), isNull(), isNull(),
- eq("order-123"), eq(1000L), eq(2000L)))
- .thenReturn(List.of(message));
+ void keyPathCapsLimitAndIncludesBodiesWhenRequestedTest() {
+ MessageRecordVO message = message("msg-3", "hello", true);
+ when(messageService.queryMessagesPage("instance-a", "TopicA", null,
null,
+ "order-123", 1000L, 2000L, 1, 100))
+ .thenReturn(page(Collections.nCopies(100, message), 150, 100,
false));
+ ToolExecutionContext request = ToolExecutionContext.of("instance-a",
null, Map.of(
+ "instanceId", "instance-a", "topicName", "TopicA", "key",
"order-123",
+ "startTime", 1000L, "endTime", 2000L, "limit", 500,
"includeBody", true));
+
+ var result =
handler.execute(request.convertInput(handler.inputType()), request);
+
+ assertThat(result.items()).hasSize(100);
+ assertThat(result.skippedCount()).isEqualTo(50);
+ assertThat(result.resultMayBeTruncated()).isTrue();
+ assertThat(result.items().getFirst()).satisfies(row -> {
+ assertThat(row.body()).isEqualTo("hello");
+ assertThat(row.bodyEncoding()).isEqualTo("UTF-8");
+ assertThat(row.bodyTruncated()).isTrue();
+ });
+ verify(messageService).queryMessagesPage("instance-a", "TopicA", null,
null,
+ "order-123", 1000L, 2000L, 1, 100);
+ }
- ListOutput<MessageItem> result = handler.execute(
- new MessageQueryInput("instance-a", "TopicA", null, null,
"order-123", 1000L, 2000L),
+ @Test
+ void providerTruncationIsPreservedWithoutLocallySkippedRowsTest() {
+ MessageRecordVO message = message("msg-4", null, false);
+ when(messageService.queryMessagesPage("instance-a", "TopicA", null,
null,
+ "order-123", null, null, 1, 10))
+ .thenReturn(page(List.of(message), 1, 10, true));
+
+ var result = handler.execute(new MessageQueryInput(
+ "instance-a", "TopicA", null, null, "order-123", null, null,
10, false),
context("instance-a"));
assertThat(result.items()).hasSize(1);
- MessageItem row = result.items().getFirst();
- assertThat(row.msgId()).isEqualTo("msg-3");
- assertThat(row.key()).isEqualTo("order-123");
+ assertThat(result.skippedCount()).isZero();
+ assertThat(result.resultMayBeTruncated()).isTrue();
}
@ParameterizedTest
@@ -149,4 +165,31 @@ class MessageQueryToolHandlerTest {
verifyNoInteractions(messageService);
}
+
+ private MessageRecordVO message(String msgId, String body, boolean
bodyTruncated) {
+ return MessageRecordVO.builder()
+ .msgId(msgId)
+ .topic("TopicA")
+ .tag("tag1")
+ .key("key1")
+ .body(body)
+ .bodyEncoding(body == null ? null : "UTF-8")
+ .bodyTruncated(bodyTruncated)
+ .storeTime(1000L)
+ .bornHost("10.0.0.1")
+ .storeHost("10.0.0.2")
+ .size(body == null ? 0 : body.length())
+ .build();
+ }
+
+ private MessageQueryPageVO page(
+ List<MessageRecordVO> items, long total, int size, boolean
resultMayBeTruncated) {
+ return MessageQueryPageVO.builder()
+ .items(items)
+ .total(total)
+ .page(1)
+ .size(size)
+ .resultMayBeTruncated(resultMayBeTruncated)
+ .build();
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolOutputSchemaContractTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolOutputSchemaContractTest.java
index 17b7a9610..d72960cc6 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolOutputSchemaContractTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolOutputSchemaContractTest.java
@@ -42,6 +42,7 @@ import
org.apache.rocketmq.studio.ops.ai.tool.contract.group.GroupListItem;
import org.apache.rocketmq.studio.ops.ai.tool.contract.group.ResetOffsetOutput;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.instance.InstanceCapabilitiesOutput;
import org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageItem;
+import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageQueryOutput;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageQueryDlqOutput;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageRedeliveryDlqOutput;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageRedeliveryOutput;
@@ -285,8 +286,18 @@ class ToolOutputSchemaContractTest {
MessageItem message = new MessageItem(
"MSG-1", "orders", "tagA", "keyA", TIMESTAMP,
"127.0.0.1:10911", "127.0.0.1:50000", "aGVsbG8=", "BASE64",
false, 5);
- samples.put("rmq.message.query", List.of(new
ListOutput<>(List.of(message))));
- samples.put("rmq.message.query_by_topic", List.of(new
ListOutput<>(List.of(message))));
+ MessageQueryOutput.Item withoutBody = new MessageQueryOutput.Item(
+ "MSG-1", "orders", "tagA", "keyA", TIMESTAMP,
+ "127.0.0.1:10911", "127.0.0.1:50000", null, null, null, 5);
+ MessageQueryOutput.Item withBody = new MessageQueryOutput.Item(
+ "MSG-1", "orders", "tagA", "keyA", TIMESTAMP,
+ "127.0.0.1:10911", "127.0.0.1:50000", "aGVsbG8=", "BASE64",
false, 5);
+ samples.put("rmq.message.query", List.of(
+ new MessageQueryOutput(List.of(withoutBody), false, 0),
+ new MessageQueryOutput(List.of(withBody), false, 0)));
+ samples.put("rmq.message.query_by_topic", List.of(
+ new MessageQueryOutput(List.of(withoutBody), true, 199),
+ new MessageQueryOutput(List.of(withBody), true, 199)));
samples.put("rmq.message.query_by_offset", List.of(new
ListOutput<>(List.of(message))));
samples.put("rmq.message.query_dlq", List.of(
MessageQueryDlqOutput.ofGroups(INSTANCE, 1, 20, 1L, List.of(
