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 57b93490d feat(tools): expose bounded DLQ resend failure details
(#7278)
57b93490d is described below
commit 57b93490da2558c906c8aa9b43c9f7a797e4b149
Author: pilichoumao <[email protected]>
AuthorDate: Mon Oct 26 16:37:14 2026 +0800
feat(tools): expose bounded DLQ resend failure details (#7278)
* feat(tools): expose bounded DLQ resend failure details
* docs(tools): place DLQ output notes after resend constraints
* docs(tools): preserve existing DLQ paragraph spacing
---
docs/api-spec.md | 6 ++
rmqctl/cmd/catalog_renderers_test.go | 65 ++++++++++++++++++++++
rmqctl/internal/catalog/catalog_gen.go | 2 +-
.../message/MessageRedeliveryDlqOutput.java | 20 ++++++-
.../main/resources/tool-catalog/tools/message.yaml | 22 ++++++++
.../MessageRedeliveryDlqToolHandlerTest.java | 58 +++++++++++++++++++
.../tool/service/ToolOutputSchemaContractTest.java | 36 +++++++++++-
7 files changed, 204 insertions(+), 5 deletions(-)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index 05e137883..0c5c7ec73 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -2075,6 +2075,12 @@ Topic，未指定 `targetTopic` 时即原 Topic）、`reason`（`string`，归
> 推断目的地（即其原 Topic），对推断出的每一个目的地跑同一次校验，任一跨集群即整批 `409` 拒绝，
> 不会出现「重投了一半」。§9.4 的按 `msgIds` 重投走同一套校验。
+AI / CLI 的 `rmq.message.redelivery_dlq` 工具在执行结果中提供相同的有界 `failures` 明细和
+`failuresTruncated` 标记：最多保留 100 条，`failed` 仍为失败总数；`scanIncomplete` 只描述扫描完整性。
+无法解析目的 Topic 时，明细省略 `targetTopic`。新服务总是返回这两个字段（无失败时为 `[]` / `false`），
+它们在工具 schema 中为可选项，以兼容仅返回聚合字段的旧服务。rmqctl 默认摘要以及 JSON / YAML 输出均保留明细，
+不会自动重试。
+
### 9.3 分页获取死信消息明细
```
diff --git a/rmqctl/cmd/catalog_renderers_test.go
b/rmqctl/cmd/catalog_renderers_test.go
index e4dcbda6b..83c064d5b 100644
--- a/rmqctl/cmd/catalog_renderers_test.go
+++ b/rmqctl/cmd/catalog_renderers_test.go
@@ -24,6 +24,7 @@ import (
"io"
"net/http"
"net/http/httptest"
+ "strings"
"sync/atomic"
"testing"
@@ -234,3 +235,67 @@ func executeCatalogResult(t *testing.T, tool string,
payload any, args ...string
}
return stdout, stderr, exitCode
}
+
+func TestDlqResendFailureDetailsOutputTest(t *testing.T) {
+ for _, legacy := range []bool{false, true} {
+ for _, format := range []string{"table", "json", "yaml"} {
+ t.Run(fmt.Sprintf("legacy=%t/%s", legacy, format),
func(t *testing.T) {
+ result := map[string]any{
+ "matched": 101, "resent": 0, "failed":
101, "outcome": "FAILED",
+ "scanIncomplete": false,
"failedQueueCount": 0,
+ }
+ if !legacy {
+ failures := make([]any, 100)
+ for i := range failures {
+ failure :=
map[string]any{"msgId": fmt.Sprintf("msg-%d", i), "reason": "broker
unavailable"}
+ if i > 0 {
+ failure["targetTopic"]
= "orders"
+ }
+ failures[i] = failure
+ }
+ result["failures"] = failures
+ result["failuresTruncated"] = true
+ }
+ payload := map[string]any{
+ "status": "EXECUTED", "instanceId":
"instance-dev",
+ "plan": map[string]any{"summary":
"Resend dead letters"}, "result": result,
+ }
+ args := []string{"message", "redelivery-dlq",
"--group-name", "group-1",
+ "--confirm-token", "confirmation",
"--yes", "--output", format}
+ stdout, stderr, exitCode :=
executeCatalogResult(t, "rmq.message.redelivery_dlq", payload, args...)
+ if exitCode != 0 || stderr != "" {
+ t.Fatalf("exit = %d, stderr = %q",
exitCode, stderr)
+ }
+ var got any
+ var err error
+ want := any(payload)
+ if format == "table" {
+ _, body, found := strings.Cut(stdout,
"\n\n")
+ if !found || !strings.Contains(stdout,
"EXECUTED") {
+ t.Fatalf("missing mutation
summary: %q", stdout)
+ }
+ err = json.Unmarshal([]byte(body), &got)
+ want = result
+ } else if format == "yaml" {
+ err = yaml.Unmarshal([]byte(stdout),
&got)
+ } else {
+ err = json.Unmarshal([]byte(stdout),
&got)
+ }
+ if err != nil {
+ t.Fatalf("decode %s output: %v",
format, err)
+ }
+ gotJSON, err := json.Marshal(got)
+ if err != nil {
+ t.Fatal(err)
+ }
+ wantJSON, err := json.Marshal(want)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(gotJSON, wantJSON) {
+ t.Fatalf("failure details changed: got
%s, want %s", gotJSON, wantJSON)
+ }
+ })
+ }
+ }
+}
diff --git a/rmqctl/internal/catalog/catalog_gen.go
b/rmqctl/internal/catalog/catalog_gen.go
index ea1dbc8d9..39871e7fb 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:
"e07c4c4a447545da57697f9e333a717c4bff6b660cefdaf6acfe4c1074791b39",
+ Digest:
"de70ac57695e5fb9db7b1a16d995c745eaa89c1e33f339954a9fb7831e7e9fd7",
Tools: []Tool{
{
Name: "rmq.acl.list",
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageRedeliveryDlqOutput.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageRedeliveryDlqOutput.java
index b85d490d7..7a52e33a8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageRedeliveryDlqOutput.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/message/MessageRedeliveryDlqOutput.java
@@ -16,15 +16,21 @@
*/
package org.apache.rocketmq.studio.ops.ai.tool.contract.message;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.rocketmq.studio.instance.dlq.DLQResendFailureVO;
import org.apache.rocketmq.studio.instance.dlq.DLQResendResultVO;
+import java.util.List;
+
public record MessageRedeliveryDlqOutput(
int matched,
int resent,
int failed,
String outcome,
boolean scanIncomplete,
- int failedQueueCount) {
+ int failedQueueCount,
+ List<Failure> failures,
+ boolean failuresTruncated) {
public static MessageRedeliveryDlqOutput from(DLQResendResultVO result) {
return new MessageRedeliveryDlqOutput(
@@ -33,6 +39,16 @@ public record MessageRedeliveryDlqOutput(
result.getFailed(),
result.getOutcome(),
result.isScanIncomplete(),
- result.getFailedQueueCount());
+ result.getFailedQueueCount(),
+ result.getFailures().stream().map(Failure::from).toList(),
+ result.isFailuresTruncated());
+ }
+
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public record Failure(String msgId, String targetTopic, String reason) {
+
+ public static Failure from(DLQResendFailureVO failure) {
+ return new Failure(failure.getMsgId(), failure.getTargetTopic(),
failure.getReason());
+ }
}
}
diff --git a/server/src/main/resources/tool-catalog/tools/message.yaml
b/server/src/main/resources/tool-catalog/tools/message.yaml
index 2da865a26..52fa0f448 100644
--- a/server/src/main/resources/tool-catalog/tools/message.yaml
+++ b/server/src/main/resources/tool-catalog/tools/message.yaml
@@ -732,5 +732,27 @@ tools:
type: boolean
failedQueueCount:
type: integer
+ failures:
+ type: array
+ description: At most 100 per-message failures, with the same
bounded reasons as the REST resend response.
+ maxItems: 100
+ items:
+ type: object
+ required:
+ - msgId
+ - reason
+ additionalProperties: false
+ properties:
+ msgId:
+ type: string
+ targetTopic:
+ type: string
+ description: Omitted when no destination could be resolved.
+ reason:
+ type: string
+ maxLength: 256
+ failuresTruncated:
+ type: boolean
+ description: True when more failures occurred than the retained
details; independent of scanIncomplete.
viewHint: object
deprecated: false
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryDlqToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryDlqToolHandlerTest.java
index 1590cf48e..d345b9f65 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryDlqToolHandlerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryDlqToolHandlerTest.java
@@ -18,6 +18,7 @@ package
org.apache.rocketmq.studio.ops.ai.tool.handler.message;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.dlq.DLQResendResultVO;
+import org.apache.rocketmq.studio.instance.dlq.DLQResendFailureVO;
import org.apache.rocketmq.studio.instance.dlq.DLQService;
import org.apache.rocketmq.studio.ops.ai.tool.contract.common.TimeRange;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageRedeliveryDlqInput;
@@ -35,6 +36,8 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Collections;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
@@ -92,11 +95,66 @@ class MessageRedeliveryDlqToolHandlerTest {
MessageRedeliveryDlqInput input = new
MessageRedeliveryDlqInput("instance-a", "group-1", null, null);
MessageRedeliveryDlqOutput output = handler.execute(input,
context("instance-a"));
assertThat(output.matched()).isZero();
+ assertThat(output.failures()).isEmpty();
+ assertThat(output.failuresTruncated()).isFalse();
verify(dlqService).resendMessages(eq("instance-a"), eq("group-1"),
eq(null), eq(null), eq(null));
verifyNoMoreInteractions(dlqService);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void executePreservesFailureDetailsAndTheirTruncationTest(boolean
truncated) {
+ DLQResendFailureVO failure = DLQResendFailureVO.builder()
+ .msgId("failed-1").targetTopic("orders").reason("Producer send
failed: broker unavailable").build();
+ List<DLQResendFailureVO> failures = truncated ?
Collections.nCopies(100, failure) : List.of(failure);
+ int failed = truncated ? 101 : 1;
+ when(dlqService.resendMessages("instance-a", "group-1", null, null,
null))
+
.thenReturn(DLQResendResultVO.builder().matched(failed).failed(failed).outcome("FAILED")
+
.failures(failures).failuresTruncated(truncated).build());
+
+ MessageRedeliveryDlqOutput output = handler.execute(
+ new MessageRedeliveryDlqInput("instance-a", "group-1", null,
null), context("instance-a"));
+
+ assertThat(output.matched()).isEqualTo(failed);
+ assertThat(output.resent()).isZero();
+ assertThat(output.failed()).isEqualTo(failed);
+ assertThat(output.outcome()).isEqualTo("FAILED");
+ assertThat(output.scanIncomplete()).isFalse();
+ assertThat(output.failedQueueCount()).isZero();
+ assertThat(output.failures()).hasSize(failures.size()).allSatisfy(item
-> {
+ assertThat(item.msgId()).isEqualTo("failed-1");
+ assertThat(item.targetTopic()).isEqualTo("orders");
+ assertThat(item.reason()).isEqualTo("Producer send failed: broker
unavailable");
+ });
+ assertThat(output.failuresTruncated()).isEqualTo(truncated);
+ verify(dlqService).resendMessages("instance-a", "group-1", null, null,
null);
+ verifyNoMoreInteractions(dlqService);
+ }
+
+ @Test
+ void executeKeepsUnresolvedTargetAbsentTest() {
+ DLQResendFailureVO failure = DLQResendFailureVO.builder()
+ .msgId("failed-1").reason("No target topic could be
resolved").build();
+ when(dlqService.resendMessages("instance-a", "group-1", null, null,
null))
+
.thenReturn(DLQResendResultVO.builder().matched(1).failed(1).outcome("FAILED")
+
.scanIncomplete(true).failedQueueCount(1).failures(List.of(failure)).build());
+
+ MessageRedeliveryDlqOutput output = handler.execute(
+ new MessageRedeliveryDlqInput("instance-a", "group-1", null,
null), context("instance-a"));
+
+ assertThat(output.failures()).singleElement().satisfies(item -> {
+ assertThat(item.msgId()).isEqualTo("failed-1");
+ assertThat(item.targetTopic()).isNull();
+ assertThat(item.reason()).isEqualTo("No target topic could be
resolved");
+ });
+ assertThat(output.scanIncomplete()).isTrue();
+ assertThat(output.failedQueueCount()).isOne();
+ assertThat(output.failuresTruncated()).isFalse();
+ verify(dlqService).resendMessages("instance-a", "group-1", null, null,
null);
+ verifyNoMoreInteractions(dlqService);
+ }
+
@Test
void executeMapsMissingDlqToNotFoundTest() {
when(dlqService.resendMessages(eq("instance-a"), eq("group-x"),
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 3377edf82..be0c9b259 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
@@ -71,12 +71,14 @@ import tools.jackson.databind.json.JsonMapper;
import java.time.LocalDateTime;
import java.util.LinkedHashMap;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Golden output-contract sweep: one representative sample per catalog tool,
validated through
@@ -112,6 +114,29 @@ class ToolOutputSchemaContractTest {
assertThat(covered).isEqualTo(declared);
}
+ @Test
+ void acceptsLegacyDlqResendOutputWithoutDetailsTest() {
+
validator.validateOutput(catalog.getDefinition("rmq.message.redelivery_dlq"),
executed(Map.of(
+ "matched", 5, "resent", 5, "failed", 0, "outcome",
"ALL_RESENT",
+ "scanIncomplete", false, "failedQueueCount", 0)));
+ }
+
+ @Test
+ void rejectsOversizedOrUndeclaredDlqFailureDetailsTest() {
+ ToolDefinition definition =
catalog.getDefinition("rmq.message.redelivery_dlq");
+ Map<String, Object> result = new LinkedHashMap<>(Map.of(
+ "matched", 101, "resent", 0, "failed", 101, "outcome",
"FAILED",
+ "scanIncomplete", false, "failedQueueCount", 0,
"failuresTruncated", true));
+ Map<String, Object> failure = new LinkedHashMap<>(Map.of("msgId",
"MSG-1", "reason", "broker unavailable"));
+ result.put("failures", Collections.nCopies(101, failure));
+ assertThatThrownBy(() -> validator.validateOutput(definition,
executed(result)))
+ .isInstanceOf(IllegalStateException.class);
+ failure.put("unexpected", "not part of the contract");
+ result.put("failures", List.of(failure));
+ assertThatThrownBy(() -> validator.validateOutput(definition,
executed(result)))
+ .isInstanceOf(IllegalStateException.class);
+ }
+
private static Map<String, List<Object>> samples() {
Map<String, List<Object>> samples = new LinkedHashMap<>();
@@ -349,8 +374,15 @@ class ToolOutputSchemaContractTest {
new MessageSendOutput("MSG-1", "OFF-1", TIMESTAMP))));
samples.put("rmq.message.redelivery", List.of(planned(), executed(
new MessageRedeliveryOutput("MSG-1", "MSG-2",
"%RETRY%cg-orders"))));
- samples.put("rmq.message.redelivery_dlq", List.of(planned(), executed(
- new MessageRedeliveryDlqOutput(5, 5, 0, "ALL_RESENT", false,
0))));
+ MessageRedeliveryDlqOutput.Failure failure = new
MessageRedeliveryDlqOutput.Failure(
+ "MSG-1", "orders", "Producer send failed: broker unavailable");
+ samples.put("rmq.message.redelivery_dlq", List.of(planned(),
+ executed(new MessageRedeliveryDlqOutput(5, 5, 0, "ALL_RESENT",
false, 0, List.of(), false)),
+ executed(new MessageRedeliveryDlqOutput(101, 0, 101, "FAILED",
false, 0,
+ Collections.nCopies(100, failure), true)),
+ executed(new MessageRedeliveryDlqOutput(1, 0, 1, "FAILED",
true, 1,
+ List.of(new MessageRedeliveryDlqOutput.Failure(
+ "MSG-1", null, "No target topic could be
resolved")), false))));
return samples;
}
