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 cbe7d6533 feat(proxy,dlq): live proxy topology/health view and
dead-letter message export (#2063)
cbe7d6533 is described below
commit cbe7d653391ad76fdaf2aced27d9707753844a42
Author: zhaohai <[email protected]>
AuthorDate: Tue Aug 18 17:30:37 2026 +0800
feat(proxy,dlq): live proxy topology/health view and dead-letter message
export (#2063)
Proxy topology/health view (BASE-01 C):
- New ProxyTopologyVO + ProxyHealthProbe (TCP probe abstraction) +
SocketProxyHealthProbe default implementation.
- ProxyAddressService.buildTopology() probes every registered proxy on its
gRPC port (and the derived 5.0 remoting port for the well-known 8080/8081
pair) and reports UP / PARTIAL / DOWN with reachability flags and latency.
- GET /api/proxies/topology controller endpoint.
- Frontend: Proxy.tsx overlays the live status onto the node list
(UP->healthy, PARTIAL->warning, DOWN->unhealthy); api/proxy.ts adds
getProxyTopology.
DLQ export (BASE-01 K):
- DLQProvider.exportMessages() scans the %DLQ%<group> topic (reusing the
existing offset-legal-aware pull scan, now with a per-call cap) and maps
each dead letter to DLQMessageVO (msgId, topic, queue/offset, storeTime,
keys, UTF-8 body + lossless bodyBase64).
- GET /api/dlq/export returns the message list as a JSON attachment
(Content-Disposition: dlq-<group>.json); DLQService validates the request.
- Frontend: the DLQ row export button now downloads the dead-letter
messages as JSON (api/message.ts + services/messageService.ts +
dlq.tsx), while the batch export keeps the CSV summary.
Tests (all green): ProxyAddressServiceTest (service ctor updated + topology
UP/PARTIAL/DOWN and non-standard port cases), ProxyControllerTest (topology
endpoint), RocketMQDLQProviderTest (export mapping + cap fallback),
DLQControllerTest (export attachment), DLQProviderStubTest (export stub);
MultiBackendMetricsSourceTest address fallback fixed (upstream regression
from the SSRF guard: hosts without a site-local IPv4 fell back to loopback,
which UrlHostGuard rejects). Server suite 1061/1061; frontend DLQ 12/12 and
Proxy 7/7.
---
.../studio/cluster/proxy/ProxyAddressService.java | 167 ++++++++++++++++++++-
.../studio/cluster/proxy/ProxyController.java | 5 +
.../studio/cluster/proxy/ProxyHealthProbe.java | 44 ++++++
.../studio/cluster/proxy/ProxyTopologyVO.java | 56 +++++++
.../cluster/proxy/SocketProxyHealthProbe.java | 46 ++++++
.../studio/instance/dlq/DLQController.java | 27 ++++
.../dlq/{DLQProvider.java => DLQMessageVO.java} | 27 +++-
.../rocketmq/studio/instance/dlq/DLQProvider.java | 2 +
.../studio/instance/dlq/DLQProviderStub.java | 7 +
.../rocketmq/studio/instance/dlq/DLQService.java | 18 ++-
.../provider/apache/RocketMQDLQProvider.java | 49 +++++-
.../metrics/MultiBackendMetricsSourceTest.java | 15 +-
.../cluster/proxy/ProxyAddressServiceTest.java | 140 ++++++++++++++++-
.../studio/cluster/proxy/ProxyControllerTest.java | 35 +++++
.../studio/instance/dlq/DLQControllerTest.java | 50 ++++++
.../studio/instance/dlq/DLQProviderStubTest.java | 9 ++
.../studio/instance/dlq/DLQServiceTest.java | 35 +++++
.../provider/apache/RocketMQDLQProviderTest.java | 61 ++++++++
web/src/api/message.ts | 11 ++
web/src/api/proxy.ts | 16 ++
web/src/pages/instance/__tests__/DLQPage.test.tsx | 23 ++-
web/src/pages/instance/dlq.tsx | 46 +++++-
web/src/pages/studio/Proxy.tsx | 33 +++-
web/src/services/messageService.ts | 11 ++
24 files changed, 903 insertions(+), 30 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
index 71279b736..578d044b2 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
@@ -22,7 +22,9 @@ import
org.apache.rocketmq.studio.common.util.NoRedirectClientHttpRequestFactory
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatusCode;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
@@ -36,8 +38,17 @@ import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.stream.Stream;
@Slf4j
@Service
@@ -50,17 +61,58 @@ public class ProxyAddressService {
private static final String RELOAD_PATH = "/admin/reloadConfig";
+ /** Default connect timeout for a single topology probe, in milliseconds.
*/
+ private static final int HEALTH_PROBE_TIMEOUT_MILLIS = 2_000;
+
+ /**
+ * Overall time budget for one topology build. Serial probing would cost
up to
+ * {@code 2 x HEALTH_PROBE_TIMEOUT_MILLIS} per DOWN node, so with many
unreachable
+ * nodes the endpoint could exceed the frontend's request timeout. Probing
runs in
+ * parallel and, once the budget elapses, unfinished probes are reported as
+ * unreachable (DOWN/PARTIAL) instead of failing the whole endpoint.
+ */
+ private static final long TOPOLOGY_TOTAL_TIMEOUT_MILLIS = 10_000L;
+
+ /** Bounded pool for the I/O-bound TCP probes; probes queue beyond this
share the budget. */
+ private static final int PROBE_EXECUTOR_THREADS = 8;
+
private final Set<String> proxyAddrs = new
LinkedHashSet<>(List.of("127.0.0.1:8081"));
private String currentProxyAddr = "127.0.0.1:8081";
private final RestTemplate restTemplate;
+ private final ProxyHealthProbe healthProbe;
+ private final ExecutorService probeExecutor;
+ private final long topologyTotalTimeoutMillis;
- public ProxyAddressService() {
+ @Autowired
+ public ProxyAddressService(ProxyHealthProbe healthProbe) {
+ this(healthProbe, defaultProbeExecutor(),
TOPOLOGY_TOTAL_TIMEOUT_MILLIS);
+ }
+
+ ProxyAddressService(ProxyHealthProbe healthProbe, ExecutorService
probeExecutor,
+ long topologyTotalTimeoutMillis) {
+ this.healthProbe = healthProbe;
+ this.probeExecutor = probeExecutor;
+ this.topologyTotalTimeoutMillis = topologyTotalTimeoutMillis;
NoRedirectClientHttpRequestFactory factory = new
NoRedirectClientHttpRequestFactory();
factory.setConnectTimeout(Duration.ofSeconds(3));
factory.setReadTimeout(Duration.ofSeconds(3));
this.restTemplate = new RestTemplate(factory);
}
+ private static ExecutorService defaultProbeExecutor() {
+ AtomicInteger threadIndex = new AtomicInteger();
+ return Executors.newFixedThreadPool(PROBE_EXECUTOR_THREADS, runnable
-> {
+ Thread thread = new Thread(runnable, "proxy-health-probe-" +
threadIndex.incrementAndGet());
+ thread.setDaemon(true);
+ return thread;
+ });
+ }
+
+ @PreDestroy
+ public void shutdownProbeExecutor() {
+ probeExecutor.shutdownNow();
+ }
+
public synchronized ProxyHomeVO getHomePage() {
return ProxyHomeVO.builder()
.proxyAddrList(new ArrayList<>(proxyAddrs))
@@ -68,6 +120,119 @@ public class ProxyAddressService {
.build();
}
+ /**
+ * Builds the proxy topology/health view: every registered proxy address
is probed over
+ * TCP on its gRPC port (and the derived remoting port) so the console can
show live
+ * UP/PARTIAL/DOWN status instead of an address list with no runtime
signal.
+ *
+ * <p>All probes run in parallel on a bounded executor and the whole view
is capped by
+ * {@link #topologyTotalTimeoutMillis}: probes that do not finish in time
are reported as
+ * unreachable, so a few DOWN nodes can never stall the endpoint past its
budget.
+ */
+ public List<ProxyTopologyVO> buildTopology() {
+ List<String> addrs;
+ synchronized (this) {
+ addrs = new ArrayList<>(proxyAddrs);
+ }
+ List<ProbeTask> tasks = new ArrayList<>();
+ for (String addr : addrs) {
+ Matcher matcher = PROXY_ADDR_PATTERN.matcher(addr);
+ if (!matcher.matches()) {
+ log.warn("Skipping malformed proxy address in topology: {}",
addr);
+ continue;
+ }
+ String host = matcher.group(1);
+ int grpcPort = Integer.parseInt(matcher.group(2));
+ Integer remotingPort = deriveRemotingPort(grpcPort);
+ tasks.add(new ProbeTask(addr, grpcPort, remotingPort,
+ probeAsync(host, grpcPort),
+ remotingPort != null ? probeAsync(host, remotingPort) :
null));
+ }
+ awaitProbes(tasks);
+ return tasks.stream().map(this::toTopologyVO).toList();
+ }
+
+ private ProxyTopologyVO toTopologyVO(ProbeTask task) {
+ ProxyHealthProbe.ProbeResult grpc = awaitProbe(task.grpcProbe());
+ ProxyHealthProbe.ProbeResult remoting = task.remotingProbe() != null
+ ? awaitProbe(task.remotingProbe())
+ : ProxyHealthProbe.ProbeResult.unreachable();
+ boolean grpcReachable = grpc.reachable();
+ boolean remotingReachable = task.remotingPort() != null &&
remoting.reachable();
+ String status = grpcReachable ? "UP"
+ : (remotingReachable ? "PARTIAL" : "DOWN");
+ return ProxyTopologyVO.builder()
+ .proxyAddr(task.proxyAddr())
+ .status(status)
+ .grpcPort(task.grpcPort())
+ .remotingPort(task.remotingPort())
+ .grpcReachable(grpcReachable)
+ .remotingReachable(remotingReachable)
+ .latencyMs(grpcReachable ? grpc.latencyMs() : -1L)
+ .build();
+ }
+
+ private CompletableFuture<ProxyHealthProbe.ProbeResult> probeAsync(String
host, int port) {
+ try {
+ return CompletableFuture.supplyAsync(
+ () -> healthProbe.probe(host, port,
HEALTH_PROBE_TIMEOUT_MILLIS), probeExecutor);
+ } catch (RejectedExecutionException ex) {
+ return
CompletableFuture.completedFuture(ProxyHealthProbe.ProbeResult.unreachable());
+ }
+ }
+
+ private void awaitProbes(List<ProbeTask> tasks) {
+ if (tasks.isEmpty()) {
+ return;
+ }
+ CompletableFuture<?>[] futures = tasks.stream()
+ .flatMap(task -> task.remotingProbe() == null
+ ? Stream.of(task.grpcProbe())
+ : Stream.of(task.grpcProbe(), task.remotingProbe()))
+ .toArray(CompletableFuture[]::new);
+ try {
+ CompletableFuture.allOf(futures).get(topologyTotalTimeoutMillis,
TimeUnit.MILLISECONDS);
+ } catch (TimeoutException ex) {
+ log.warn("Proxy topology probing exceeded the {} ms budget;
unfinished probes are"
+ + " reported as unreachable", topologyTotalTimeoutMillis);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ log.warn("Proxy topology probing was interrupted");
+ } catch (ExecutionException ex) {
+ log.warn("Proxy topology probe failed: {}", ex.getMessage());
+ }
+ }
+
+ /**
+ * Reads a finished probe outcome; a probe that is still running (budget
exceeded), failed,
+ * or was rejected degrades to unreachable instead of propagating an error.
+ */
+ private ProxyHealthProbe.ProbeResult
awaitProbe(CompletableFuture<ProxyHealthProbe.ProbeResult> future) {
+ if (future.isDone() && !future.isCompletedExceptionally()) {
+ return future.join();
+ }
+ future.cancel(true);
+ return ProxyHealthProbe.ProbeResult.unreachable();
+ }
+
+ /** In-flight probe pair for one registered proxy address. */
+ private record ProbeTask(String proxyAddr, int grpcPort, Integer
remotingPort,
+ CompletableFuture<ProxyHealthProbe.ProbeResult>
grpcProbe,
+ CompletableFuture<ProxyHealthProbe.ProbeResult>
remotingProbe) {
+ }
+
+ /**
+ * Derives the counterpart port using the RocketMQ 5.0 default layout
(remoting
+ * {@code 8080} / gRPC {@code 8081}). Non-standard ports yield {@code
null} because the
+ * pairing cannot be assumed for custom port mappings.
+ */
+ private Integer deriveRemotingPort(int grpcPort) {
+ if (grpcPort == 8081) {
+ return 8080;
+ }
+ return grpcPort == 8080 ? 8081 : null;
+ }
+
public synchronized void addProxyAddr(String newProxyAddr) {
String normalized = normalizeProxyAddr(newProxyAddr, "newProxyAddr");
proxyAddrs.add(normalized);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyController.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyController.java
index dec10d8b6..34cba7970 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyController.java
@@ -48,6 +48,11 @@ public class ProxyController {
return Result.ok(proxies);
}
+ @GetMapping("/topology")
+ public Result<List<ProxyTopologyVO>> getProxyTopology() {
+ return Result.ok(proxyAddressService.buildTopology());
+ }
+
@PostMapping("/config/reload")
public Result<Map<String, Boolean>> reloadProxyConfig(@Valid @RequestBody
RestartProxyDTO command) {
proxyAddressService.reloadConfig(command.getAddr());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyHealthProbe.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyHealthProbe.java
new file mode 100644
index 000000000..38e2ef102
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyHealthProbe.java
@@ -0,0 +1,44 @@
+/*
+ * 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.cluster.proxy;
+
+/**
+ * TCP reachability probe used by {@link ProxyAddressService#buildTopology()}
to decide
+ * whether a proxy node is up. Abstracted so tests can simulate
reachable/unreachable
+ * ports without opening sockets.
+ */
+public interface ProxyHealthProbe {
+
+ /**
+ * Attempts a TCP connection to {@code host:port} within {@code
timeoutMillis}.
+ *
+ * @return the probe outcome with the measured latency (or -1 when
unreachable)
+ */
+ ProbeResult probe(String host, int port, int timeoutMillis);
+
+ /** Outcome of a single probe. */
+ record ProbeResult(boolean reachable, long latencyMs) {
+
+ public static ProbeResult unreachable() {
+ return new ProbeResult(false, -1L);
+ }
+
+ public static ProbeResult reachable(long latencyMs) {
+ return new ProbeResult(true, Math.max(0L, latencyMs));
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyTopologyVO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyTopologyVO.java
new file mode 100644
index 000000000..b95f6b672
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyTopologyVO.java
@@ -0,0 +1,56 @@
+/*
+ * 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.cluster.proxy;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Health/topology view of a single 5.0 Proxy node. {@code proxyAddr} is the
+ * registered {@code host:grpcPort} address; the remoting port is derived with
the
+ * 5.0 default layout ({@code grpcPort - 1}) when it looks like the well-known
+ * {@code 8080/8081} pair, otherwise the probe only reports the gRPC side.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ProxyTopologyVO {
+
+ /** Registered proxy address in {@code host:grpcPort} form. */
+ private String proxyAddr;
+
+ /** UP (gRPC reachable) / PARTIAL (gRPC down, remoting up) / DOWN. */
+ private String status;
+
+ /** gRPC port of the proxy. */
+ private int grpcPort;
+
+ /** Derived remoting port ({@code grpcPort - 1}), {@code null} when not
derivable. */
+ private Integer remotingPort;
+
+ /** Whether the gRPC port accepted a TCP connection. */
+ private boolean grpcReachable;
+
+ /** Whether the derived remoting port accepted a TCP connection
(null-safe). */
+ private boolean remotingReachable;
+
+ /** Round-trip latency of the successful probe in milliseconds, -1 when
unreachable. */
+ private long latencyMs;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/SocketProxyHealthProbe.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/SocketProxyHealthProbe.java
new file mode 100644
index 000000000..77c6678cf
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/SocketProxyHealthProbe.java
@@ -0,0 +1,46 @@
+/*
+ * 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.cluster.proxy;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+
+/**
+ * Default {@link ProxyHealthProbe} backed by a plain TCP connect. A connect
succeeds as
+ * soon as the remote end accepts the handshake, so this also works for
proxies that only
+ * listen on gRPC without exposing an HTTP health endpoint.
+ */
+@Slf4j
+@Component
+public class SocketProxyHealthProbe implements ProxyHealthProbe {
+
+ @Override
+ public ProbeResult probe(String host, int port, int timeoutMillis) {
+ long start = System.nanoTime();
+ try (Socket socket = new Socket()) {
+ socket.connect(new InetSocketAddress(host, port), timeoutMillis);
+ long latencyMs = (System.nanoTime() - start) / 1_000_000L;
+ return ProbeResult.reachable(latencyMs);
+ } catch (IOException exception) {
+ return ProbeResult.unreachable();
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQController.java
index 096b3b879..6487cba54 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQController.java
@@ -16,10 +16,15 @@
*/
package org.apache.rocketmq.studio.instance.dlq;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.studio.common.domain.Result;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
@@ -35,6 +40,7 @@ import java.util.List;
public class DLQController {
private final DLQService dlqService;
+ private final ObjectMapper objectMapper;
@GetMapping
public Result<List<DLQGroupVO>> listDLQGroups(@RequestParam String
instanceId) {
@@ -48,6 +54,27 @@ public class DLQController {
request.getStartTime(), request.getEndTime(),
request.getTargetTopic()));
}
+ @GetMapping("/export")
+ public ResponseEntity<byte[]> exportDLQMessages(@RequestParam String
instanceId,
+ @RequestParam String
groupName,
+ @RequestParam(required =
false) Long startTime,
+ @RequestParam(required =
false) Long endTime,
+ @RequestParam(required =
false) Integer maxCount) {
+ List<DLQMessageVO> messages = dlqService.exportMessages(
+ instanceId, groupName, startTime, endTime, maxCount);
+ byte[] body;
+ try {
+ body = objectMapper.writeValueAsBytes(messages);
+ } catch (JsonProcessingException exception) {
+ throw new BusinessException(500, "Failed to serialize DLQ export");
+ }
+ return ResponseEntity.ok()
+ .header(HttpHeaders.CONTENT_DISPOSITION,
+ "attachment; filename=\"dlq-" + groupName + ".json\"")
+ .contentType(MediaType.APPLICATION_JSON)
+ .body(body);
+ }
+
private void requireRequest(DLQResendRequestDTO request) {
if (request == null) {
throw new BusinessException(400, "DLQ resend request is required");
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageVO.java
similarity index 57%
copy from
server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
copy to
server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageVO.java
index e91cf484d..a27842460 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageVO.java
@@ -16,11 +16,28 @@
*/
package org.apache.rocketmq.studio.instance.dlq;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
-import java.util.List;
+/**
+ * A dead-letter message as exported by the DLQ export endpoint. {@code body}
carries the
+ * UTF-8-decoded payload (best effort) while {@code bodyBase64} preserves the
exact bytes
+ * so binary messages can be exported losslessly.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DLQMessageVO {
-public interface DLQProvider {
- List<DLQGroupVO> listDLQGroups(String instanceId);
- DLQResendResultVO resendMessages(String instanceId, String groupName, Long
startTime, Long endTime,
- String targetTopic);
+ private String msgId;
+ private String topic;
+ private int queueId;
+ private long offset;
+ private long storeTime;
+ private String keys;
+ private String body;
+ private String bodyBase64;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
index e91cf484d..4f1c7dbd8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
@@ -23,4 +23,6 @@ public interface DLQProvider {
List<DLQGroupVO> listDLQGroups(String instanceId);
DLQResendResultVO resendMessages(String instanceId, String groupName, Long
startTime, Long endTime,
String targetTopic);
+ List<DLQMessageVO> exportMessages(String instanceId, String groupName,
Long startTime, Long endTime,
+ Integer maxCount);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStub.java
index 2011a9660..bb9400c52 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStub.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStub.java
@@ -43,6 +43,13 @@ public class DLQProviderStub implements DLQProvider {
throw unsupported();
}
+ @Override
+ public List<DLQMessageVO> exportMessages(String instanceId, String
groupName, Long startTime, Long endTime,
+ Integer maxCount) {
+ log.warn("DLQProviderStub.exportMessages called but no real DLQ
provider is configured. group={}", groupName);
+ throw unsupported();
+ }
+
private BusinessException unsupported() {
return new BusinessException(501, "DLQ provider is not configured");
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
index 5923e4741..6447e0c5f 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
@@ -21,7 +21,6 @@ import
org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
-import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
@@ -49,6 +48,14 @@ public class DLQService {
return dlqProvider.resendMessages(instanceId, groupName, startTime,
endTime, targetTopic);
}
+ public List<DLQMessageVO> exportMessages(String instanceId, String
groupName, Long startTime, Long endTime,
+ Integer maxCount) {
+ requireApacheInstance(instanceId);
+ validateResendRequest(groupName, startTime, endTime);
+ log.info("Exporting DLQ messages: group={}, maxCount={}", groupName,
maxCount);
+ return dlqProvider.exportMessages(instanceId, groupName, startTime,
endTime, maxCount);
+ }
+
private void requireApacheInstance(String instanceId) {
providerRegistry.byInstanceId(instanceId).ifPresent(provider -> {
if (provider.vendor() != InstanceVendor.APACHE) {
@@ -62,19 +69,16 @@ public class DLQService {
throw new BusinessException(400, "groupName is required");
}
if ((startTime == null) != (endTime == null)) {
- throw new BusinessException(
- HttpStatus.BAD_REQUEST.value(), "startTime and endTime
must be provided together");
+ throw new BusinessException(400, "startTime and endTime must be
provided together");
}
if (startTime == null) {
return;
}
if (startTime <= 0 || endTime <= 0) {
- throw new BusinessException(
- HttpStatus.BAD_REQUEST.value(), "startTime and endTime
must be positive");
+ throw new BusinessException(400, "startTime and endTime must be
positive");
}
if (endTime < startTime) {
- throw new BusinessException(
- HttpStatus.BAD_REQUEST.value(), "endTime must not be
earlier than startTime");
+ throw new BusinessException(400, "endTime must not be earlier than
startTime");
}
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
index d8d8660e5..18a0c37e9 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
@@ -33,6 +33,7 @@ import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.dlq.DLQGroupVO;
+import org.apache.rocketmq.studio.instance.dlq.DLQMessageVO;
import org.apache.rocketmq.studio.instance.dlq.DLQProvider;
import org.apache.rocketmq.studio.instance.dlq.DLQResendResultVO;
import org.apache.rocketmq.studio.ops.audit.AuditService;
@@ -43,10 +44,12 @@ import org.springframework.util.StringUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
+import java.util.Base64;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -145,7 +148,7 @@ public class RocketMQDLQProvider implements DLQProvider {
DeadLetterScanResult scanResult;
try {
- scanResult = collectDeadLetters(endpoint, dlqTopic, begin, end);
+ scanResult = collectDeadLetters(endpoint, dlqTopic, begin, end,
RESEND_HARD_CAP);
} catch (BusinessException e) {
String detail = String.format("instanceId=%s, group=%s,
dlqTopic=%s, targetTopic=%s, "
+ "matched=0, resent=0, failed=0,
scanIncomplete=true, scanFailedQueues=all",
@@ -193,7 +196,45 @@ public class RocketMQDLQProvider implements DLQProvider {
.build();
}
- private DeadLetterScanResult collectDeadLetters(String endpoint, String
dlqTopic, long begin, long end) {
+ @Override
+ public List<DLQMessageVO> exportMessages(String instanceId, String
groupName, Long startTime, Long endTime,
+ Integer maxCount) {
+ String endpoint =
runtimeAdminClientResolver.resolveEndpoint(instanceId);
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
+ long end = endTime != null ? endTime : System.currentTimeMillis();
+ long begin = startTime != null ? startTime : end - ONE_HOUR_MILLIS;
+ int cap = maxCount == null || maxCount <= 0 ? RESEND_HARD_CAP :
Math.min(maxCount, RESEND_HARD_CAP);
+ DeadLetterScanResult scanResult = collectDeadLetters(endpoint,
dlqTopic, begin, end, cap);
+ return scanResult.messages().stream().map(this::toExportVO).toList();
+ }
+
+ private DLQMessageVO toExportVO(MessageExt message) {
+ return DLQMessageVO.builder()
+ .msgId(message.getMsgId())
+ .topic(message.getTopic())
+ .queueId(message.getQueueId())
+ .offset(message.getQueueOffset())
+ .storeTime(message.getStoreTimestamp())
+ .keys(message.getKeys())
+ .body(toUtf8Text(message.getBody()))
+ .bodyBase64(message.getBody() == null ? null
+ :
Base64.getEncoder().encodeToString(message.getBody()))
+ .build();
+ }
+
+ private String toUtf8Text(byte[] body) {
+ if (body == null) {
+ return null;
+ }
+ try {
+ return new String(body, StandardCharsets.UTF_8);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ private DeadLetterScanResult collectDeadLetters(String endpoint, String
dlqTopic, long begin, long end,
+ int cap) {
DefaultMQPullConsumer consumer = newPullConsumer(endpoint);
List<MessageExt> result = new ArrayList<>();
int failedQueueCount = 0;
@@ -212,7 +253,7 @@ public class RocketMQDLQProvider implements DLQProvider {
long maxOffset = consumer.searchOffset(queue, end);
int consecutiveIllegalOffsets = 0;
for (long offset = minOffset; offset <= maxOffset; ) {
- if (result.size() >= RESEND_HARD_CAP) {
+ if (result.size() >= cap) {
break outer;
}
PullResult pullResult = consumer.pull(queue, "*",
offset, 32);
@@ -254,7 +295,7 @@ public class RocketMQDLQProvider implements DLQProvider {
if (messageExt.getStoreTimestamp() >= begin
&& messageExt.getStoreTimestamp() <= end) {
result.add(messageExt);
- if (result.size() >= RESEND_HARD_CAP) {
+ if (result.size() >= cap) {
break outer;
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MultiBackendMetricsSourceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MultiBackendMetricsSourceTest.java
index 74c697424..e20662ec0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MultiBackendMetricsSourceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MultiBackendMetricsSourceTest.java
@@ -165,6 +165,7 @@ class MultiBackendMetricsSourceTest {
}
private static java.net.InetAddress findSiteLocalAddress() throws
java.net.SocketException {
+ InetAddress fallback = null;
Enumeration<NetworkInterface> interfaces =
NetworkInterface.getNetworkInterfaces();
while (interfaces.hasMoreElements()) {
NetworkInterface iface = interfaces.nextElement();
@@ -174,13 +175,23 @@ class MultiBackendMetricsSourceTest {
for (InterfaceAddress address : iface.getInterfaceAddresses()) {
InetAddress inet = address.getAddress();
if (inet instanceof java.net.Inet4Address
- && inet.isSiteLocalAddress()
&& !inet.isLoopbackAddress()
&& !inet.isLinkLocalAddress()) {
- return inet;
+ if (inet.isSiteLocalAddress()) {
+ return inet;
+ }
+ if (fallback == null) {
+ fallback = inet;
+ }
}
}
}
+ // No site-local interface (e.g. hosts that only expose public
ranges): fall back to any
+ // non-loopback IPv4 so the embedded server stays reachable under the
SSRF guard, which
+ // rejects loopback/link-local addresses (see UrlHostGuard).
+ if (fallback != null) {
+ return fallback;
+ }
return java.net.InetAddress.getLoopbackAddress();
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
index 94a14d11e..90776ba06 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
@@ -18,16 +18,30 @@
package org.apache.rocketmq.studio.cluster.proxy;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
class ProxyAddressServiceTest {
- private final ProxyAddressService proxyAddressService = new
ProxyAddressService();
+ private final ProxyHealthProbe healthProbe = mock(ProxyHealthProbe.class);
+ private final ProxyAddressService proxyAddressService = new
ProxyAddressService(healthProbe);
+
+ @BeforeEach
+ void setUp() {
+ // Default probe outcome: everything reachable with 1 ms latency.
+ when(healthProbe.probe(anyString(), anyInt(), anyInt()))
+ .thenReturn(ProxyHealthProbe.ProbeResult.reachable(1L));
+ }
@Test
void homePageShouldReturnDefaultProxyAddress() {
@@ -133,4 +147,128 @@ class ProxyAddressServiceTest {
.hasMessage("addr is not a registered proxy address")
.satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(400));
}
+
+ @Test
+ void buildTopologyShouldReportUpPartialAndDownStatus() {
+ proxyAddressService.addProxyAddr("10.0.0.2:8081");
+ proxyAddressService.addProxyAddr("10.0.0.3:8081");
+
+ // 10.0.0.2: gRPC down, remoting (8080) up → PARTIAL
+ when(healthProbe.probe(eq("10.0.0.2"), eq(8081), anyInt()))
+ .thenReturn(ProxyHealthProbe.ProbeResult.unreachable());
+ when(healthProbe.probe(eq("10.0.0.2"), eq(8080), anyInt()))
+ .thenReturn(ProxyHealthProbe.ProbeResult.reachable(2L));
+ // 10.0.0.3: both ports down → DOWN
+ when(healthProbe.probe(eq("10.0.0.3"), anyInt(), anyInt()))
+ .thenReturn(ProxyHealthProbe.ProbeResult.unreachable());
+
+ List<ProxyTopologyVO> topology = proxyAddressService.buildTopology();
+
+ assertThat(topology).hasSize(3);
+
+ ProxyTopologyVO up = topology.get(0);
+ assertThat(up.getProxyAddr()).isEqualTo("127.0.0.1:8081");
+ assertThat(up.getStatus()).isEqualTo("UP");
+ assertThat(up.getGrpcPort()).isEqualTo(8081);
+ assertThat(up.getRemotingPort()).isEqualTo(8080);
+ assertThat(up.isGrpcReachable()).isTrue();
+ assertThat(up.isRemotingReachable()).isTrue();
+ assertThat(up.getLatencyMs()).isEqualTo(1L);
+
+ ProxyTopologyVO partial = topology.get(1);
+ assertThat(partial.getProxyAddr()).isEqualTo("10.0.0.2:8081");
+ assertThat(partial.getStatus()).isEqualTo("PARTIAL");
+ assertThat(partial.isGrpcReachable()).isFalse();
+ assertThat(partial.isRemotingReachable()).isTrue();
+
+ ProxyTopologyVO down = topology.get(2);
+ assertThat(down.getProxyAddr()).isEqualTo("10.0.0.3:8081");
+ assertThat(down.getStatus()).isEqualTo("DOWN");
+ assertThat(down.isGrpcReachable()).isFalse();
+ assertThat(down.isRemotingReachable()).isFalse();
+ assertThat(down.getLatencyMs()).isEqualTo(-1L);
+ }
+
+ @Test
+ void buildTopologyShouldNotDeriveRemotingPortForNonStandardGrpcPort() {
+ proxyAddressService.addProxyAddr("10.0.0.4:8443");
+
+ List<ProxyTopologyVO> topology = proxyAddressService.buildTopology();
+
+ ProxyTopologyVO custom = topology.stream()
+ .filter(node -> node.getProxyAddr().equals("10.0.0.4:8443"))
+ .findFirst()
+ .orElseThrow();
+ assertThat(custom.getStatus()).isEqualTo("UP");
+ assertThat(custom.getRemotingPort()).isNull();
+ assertThat(custom.isRemotingReachable()).isFalse();
+ }
+
+ @Test
+ void buildTopologyShouldProbeAddressesInParallelTest() {
+ // Every probe blocks 300 ms; 4 addresses mean 8 probes, which would
take
+ // at least 2.4 s if probed serially but only one wave (~300 ms) in
parallel.
+ ProxyHealthProbe slowProbe = (host, port, timeoutMillis) -> {
+ try {
+ Thread.sleep(300L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return ProxyHealthProbe.ProbeResult.reachable(300L);
+ };
+ ProxyAddressService service = new ProxyAddressService(slowProbe);
+ service.addProxyAddr("10.0.0.11:8081");
+ service.addProxyAddr("10.0.0.12:8081");
+ service.addProxyAddr("10.0.0.13:8081");
+
+ long start = System.nanoTime();
+ List<ProxyTopologyVO> topology = service.buildTopology();
+ long elapsedMillis = (System.nanoTime() - start) / 1_000_000L;
+
+ assertThat(topology).hasSize(4);
+ assertThat(elapsedMillis).isLessThan(2_000L);
+ }
+
+ @Test
+ void buildTopologyShouldDegradeHungProbesWithinTotalBudgetTest() {
+ // 10.0.0.21: gRPC probe hangs, remoting probe returns → PARTIAL.
+ // 10.0.0.22: both probes hang → DOWN. Neither may exceed the 500 ms
budget.
+ ProxyHealthProbe selectiveProbe = (host, port, timeoutMillis) -> {
+ boolean hang = "10.0.0.21".equals(host) ? port == 8081 :
"10.0.0.22".equals(host);
+ if (hang) {
+ try {
+ Thread.sleep(5_000L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return ProxyHealthProbe.ProbeResult.reachable(1L);
+ };
+ java.util.concurrent.ExecutorService executor =
+ java.util.concurrent.Executors.newFixedThreadPool(4);
+ ProxyAddressService service = new ProxyAddressService(selectiveProbe,
executor, 500L);
+ service.addProxyAddr("10.0.0.21:8081");
+ service.addProxyAddr("10.0.0.22:8081");
+
+ long start = System.nanoTime();
+ List<ProxyTopologyVO> topology = service.buildTopology();
+ long elapsedMillis = (System.nanoTime() - start) / 1_000_000L;
+ executor.shutdownNow();
+
+ assertThat(elapsedMillis).isLessThan(3_000L);
+ assertThat(topology).hasSize(3);
+
+ ProxyTopologyVO partial = topology.get(1);
+ assertThat(partial.getProxyAddr()).isEqualTo("10.0.0.21:8081");
+ assertThat(partial.getStatus()).isEqualTo("PARTIAL");
+ assertThat(partial.isGrpcReachable()).isFalse();
+ assertThat(partial.isRemotingReachable()).isTrue();
+ assertThat(partial.getLatencyMs()).isEqualTo(-1L);
+
+ ProxyTopologyVO down = topology.get(2);
+ assertThat(down.getProxyAddr()).isEqualTo("10.0.0.22:8081");
+ assertThat(down.getStatus()).isEqualTo("DOWN");
+ assertThat(down.isGrpcReachable()).isFalse();
+ assertThat(down.isRemotingReachable()).isFalse();
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyControllerTest.java
index ee3d9404e..01c2c3ee0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyControllerTest.java
@@ -127,6 +127,41 @@ class ProxyControllerTest {
.andExpect(jsonPath("$.message").value("clusterId is
required"));
}
+ @Test
+ void proxyTopologyShouldReturnHealthView() throws Exception {
+ when(proxyAddressService.buildTopology())
+ .thenReturn(List.of(
+ ProxyTopologyVO.builder()
+ .proxyAddr("127.0.0.1:8081")
+ .status("UP")
+ .grpcPort(8081)
+ .remotingPort(8080)
+ .grpcReachable(true)
+ .remotingReachable(true)
+ .latencyMs(1L)
+ .build(),
+ ProxyTopologyVO.builder()
+ .proxyAddr("10.0.0.2:8081")
+ .status("DOWN")
+ .grpcPort(8081)
+ .remotingPort(8080)
+ .grpcReachable(false)
+ .remotingReachable(false)
+ .latencyMs(-1L)
+ .build()));
+
+ mockMvc.perform(get("/api/proxies/topology"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+
.andExpect(jsonPath("$.data[0].proxyAddr").value("127.0.0.1:8081"))
+ .andExpect(jsonPath("$.data[0].status").value("UP"))
+ .andExpect(jsonPath("$.data[0].grpcReachable").value(true))
+ .andExpect(jsonPath("$.data[1].status").value("DOWN"))
+ .andExpect(jsonPath("$.data[1].latencyMs").value(-1));
+
+ verify(proxyAddressService).buildTopology();
+ }
+
@Test
void reloadProxyConfigShouldReturnSuccess() throws Exception {
RestartProxyDTO request = RestartProxyDTO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQControllerTest.java
index 00aca6d3a..a6063bd1c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQControllerTest.java
@@ -23,6 +23,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import
org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.test.web.servlet.MockMvc;
@@ -37,6 +38,8 @@ import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.header;
import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@@ -176,4 +179,51 @@ class DLQControllerTest {
verifyNoInteractions(dlqService);
}
+
+ @Test
+ void exportDLQMessagesShouldReturnJsonAttachment() throws Exception {
+ when(dlqService.exportMessages(eq("instance-1"), eq("test-group"),
isNull(), isNull(), isNull()))
+ .thenReturn(List.of(
+ DLQMessageVO.builder()
+ .msgId("msg-1")
+ .topic("%DLQ%test-group")
+ .queueId(0)
+ .offset(5L)
+ .storeTime(150L)
+ .keys("key-a")
+ .body("hello dlq")
+ .bodyBase64("aGVsbG8gZGxx")
+ .build()));
+
+ mockMvc.perform(get("/api/dlq/export")
+ .param("instanceId", "instance-1")
+ .param("groupName", "test-group"))
+ .andExpect(status().isOk())
+ .andExpect(header().string(HttpHeaders.CONTENT_DISPOSITION,
+ "attachment; filename=\"dlq-test-group.json\""))
+
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON))
+ .andExpect(jsonPath("$[0].msgId").value("msg-1"))
+ .andExpect(jsonPath("$[0].body").value("hello dlq"));
+
+ verify(dlqService).exportMessages(eq("instance-1"), eq("test-group"),
isNull(), isNull(), isNull());
+ }
+
+ @Test
+ void exportDLQMessagesShouldPassTimeRangeTest() throws Exception {
+ when(dlqService.exportMessages(eq("instance-1"), eq("test-group"),
eq(1000L), eq(2000L), eq(100)))
+ .thenReturn(List.of());
+
+ mockMvc.perform(get("/api/dlq/export")
+ .param("instanceId", "instance-1")
+ .param("groupName", "test-group")
+ .param("startTime", "1000")
+ .param("endTime", "2000")
+ .param("maxCount", "100"))
+ .andExpect(status().isOk())
+ .andExpect(header().string(HttpHeaders.CONTENT_DISPOSITION,
+ "attachment; filename=\"dlq-test-group.json\""))
+
.andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON));
+
+ verify(dlqService).exportMessages(eq("instance-1"), eq("test-group"),
eq(1000L), eq(2000L), eq(100));
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStubTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStubTest.java
index 69700556f..7b3483532 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStubTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStubTest.java
@@ -43,4 +43,13 @@ class DLQProviderStubTest {
.extracting("code")
.isEqualTo(501);
}
+
+ @Test
+ void exportMessagesShouldFailExplicitlyWhenRealProviderIsMissing() {
+ assertThatThrownBy(() -> provider.exportMessages("instance-1",
"group-1", 1000L, 2000L, 100))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("DLQ provider is not configured")
+ .extracting("code")
+ .isEqualTo(501);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
index 8f33cefc4..7936cc444 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.instance.dlq;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -125,4 +126,38 @@ class DLQServiceTest {
verifyNoInteractions(dlqProvider);
}
+
+ @Test
+ void exportMessagesShouldDelegateToProviderTest() {
+ dlqService.exportMessages("instance-1", "group-1", 1000L, 2000L, 100);
+
+ verify(dlqProvider).exportMessages("instance-1", "group-1", 1000L,
2000L, 100);
+ }
+
+ @Test
+ void exportMessagesShouldAcceptNullTimeRangeTest() {
+ dlqService.exportMessages("instance-1", "group-1", null, null, null);
+
+ verify(dlqProvider).exportMessages("instance-1", "group-1", null,
null, null);
+ }
+
+ @Test
+ void exportMessagesShouldRejectPartialTimeRangeTest() {
+ assertThatThrownBy(() -> dlqService.exportMessages("instance-1",
"group-1", 1000L, null, null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("startTime and endTime must be provided together")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(400));
+
+ verifyNoInteractions(dlqProvider);
+ }
+
+ @Test
+ void exportMessagesShouldRejectReversedTimeRangeTest() {
+ assertThatThrownBy(() -> dlqService.exportMessages("instance-1",
"group-1", 2000L, 1000L, null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("endTime must not be earlier than startTime")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(400));
+
+ verifyNoInteractions(dlqProvider);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
index d43e5da28..b7e9f37e4 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
@@ -33,6 +33,7 @@ import
org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.dlq.DLQGroupVO;
+import org.apache.rocketmq.studio.instance.dlq.DLQMessageVO;
import org.apache.rocketmq.studio.ops.audit.AuditService;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.junit.jupiter.api.BeforeEach;
@@ -44,6 +45,8 @@ import org.mockito.Mock;
import org.mockito.MockedConstruction;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -375,4 +378,62 @@ class RocketMQDLQProviderTest {
assertThat(first).startsWith("studio-dlq-resend-");
assertThat(second).startsWith("studio-dlq-resend-").isNotEqualTo(first);
}
+
+ @Test
+ void exportMessagesReturnsMappedDeadLetterMessages() throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(dlqTopic, "broker-a", 0);
+ MessageExt deadLetter = new MessageExt();
+ deadLetter.setMsgId("msg-1");
+ deadLetter.setTopic(dlqTopic);
+ deadLetter.setQueueId(0);
+ deadLetter.setQueueOffset(5L);
+ deadLetter.setStoreTimestamp(150L);
+ deadLetter.setKeys("key-a,key-b");
+ deadLetter.setBody("hello dlq".getBytes(StandardCharsets.UTF_8));
+ PullResult pullResult = new PullResult(PullStatus.FOUND, 1L, 0L, 0L,
List.of(deadLetter));
+ try (MockedConstruction<DefaultMQPullConsumer> mockedConsumers =
+ mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ doNothing().when(consumer).start();
+
when(consumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(consumer.searchOffset(eq(queue),
anyLong())).thenReturn(0L);
+ when(consumer.pull(eq(queue), eq("*"), eq(0L),
eq(32))).thenReturn(pullResult);
+ doNothing().when(consumer).shutdown();
+ })) {
+ List<DLQMessageVO> exported =
+ provider.exportMessages("instance-a", "group-a", 100L,
200L, 1000);
+
+ assertThat(exported).hasSize(1);
+ DLQMessageVO vo = exported.get(0);
+ assertThat(vo.getMsgId()).isEqualTo("msg-1");
+ assertThat(vo.getTopic()).isEqualTo(dlqTopic);
+ assertThat(vo.getQueueId()).isEqualTo(0);
+ assertThat(vo.getOffset()).isEqualTo(5L);
+ assertThat(vo.getStoreTime()).isEqualTo(150L);
+ assertThat(vo.getKeys()).isEqualTo("key-a,key-b");
+ assertThat(vo.getBody()).isEqualTo("hello dlq");
+ assertThat(vo.getBodyBase64())
+ .isEqualTo(Base64.getEncoder().encodeToString("hello
dlq".getBytes(StandardCharsets.UTF_8)));
+ }
+ }
+
+ @Test
+ void exportMessagesHonorsMaxCountCap() throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(dlqTopic, "broker-a", 0);
+ try (MockedConstruction<DefaultMQPullConsumer> mockedConsumers =
+ mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ doNothing().when(consumer).start();
+
when(consumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(consumer.searchOffset(eq(queue),
anyLong())).thenReturn(0L);
+ when(consumer.pull(eq(queue), eq("*"), eq(0L),
eq(32)))
+ .thenReturn(new
PullResult(PullStatus.NO_NEW_MSG, 1L, 0L, 0L, List.of()));
+ doNothing().when(consumer).shutdown();
+ })) {
+ // maxCount=0 falls back to the hard cap instead of failing; scan
still completes.
+ List<DLQMessageVO> exported =
+ provider.exportMessages("instance-a", "group-a", 100L,
200L, 0);
+ assertThat(exported).isEmpty();
+ }
+ }
}
diff --git a/web/src/api/message.ts b/web/src/api/message.ts
index 30f750031..023283a1a 100644
--- a/web/src/api/message.ts
+++ b/web/src/api/message.ts
@@ -107,3 +107,14 @@ export async function resendDLQ(data: {
const res = await client.post<{ data: DLQResendResult }>('/dlq/resend',
data);
return res.data.data;
}
+
+export async function exportDLQMessages(params: {
+ instanceId: string;
+ groupName: string;
+ startTime?: number;
+ endTime?: number;
+ maxCount?: number;
+}): Promise<Blob> {
+ const res = await client.get<Blob>('/dlq/export', { params, responseType:
'blob' });
+ return res.data;
+}
diff --git a/web/src/api/proxy.ts b/web/src/api/proxy.ts
index aabab4629..b4d5003e9 100644
--- a/web/src/api/proxy.ts
+++ b/web/src/api/proxy.ts
@@ -37,6 +37,16 @@ export interface ProxyNode {
isSelected: boolean;
}
+export interface ProxyTopologyNode {
+ proxyAddr: string;
+ status: 'UP' | 'PARTIAL' | 'DOWN';
+ grpcPort: number;
+ remotingPort: number | null;
+ grpcReachable: boolean;
+ remotingReachable: boolean;
+ latencyMs: number;
+}
+
// ─── API Functions ───────────────────────────────────────────────
export async function queryProxyHomePage(): Promise<ProxyHomePageData> {
@@ -44,6 +54,12 @@ export async function queryProxyHomePage():
Promise<ProxyHomePageData> {
return res.data.data;
}
+/** Live TCP health/topology view of every registered proxy node. */
+export async function getProxyTopology(): Promise<ProxyTopologyNode[]> {
+ const res = await client.get<{ data: ProxyTopologyNode[]
}>('/proxies/topology');
+ return res.data.data;
+}
+
/**
* Trigger a configuration hot-reload for a proxy.
* Uses the same DTO as restartProxy ({ clusterId, addr }).
diff --git a/web/src/pages/instance/__tests__/DLQPage.test.tsx
b/web/src/pages/instance/__tests__/DLQPage.test.tsx
index 03ffa20e3..060944897 100644
--- a/web/src/pages/instance/__tests__/DLQPage.test.tsx
+++ b/web/src/pages/instance/__tests__/DLQPage.test.tsx
@@ -29,6 +29,7 @@ import DLQPage from '../dlq';
vi.mock('../../../services/messageService', () => ({
listDLQGroups: vi.fn(),
resendDLQ: vi.fn(),
+ exportDLQMessages: vi.fn(),
}));
vi.mock('../../../services/instanceService', () => ({
listInstances: vi.fn().mockResolvedValue([
@@ -189,16 +190,29 @@ describe('DLQ page', () => {
expect(screen.getByText('ACTIVE')).toBeInTheDocument();
});
- it('exports the selected group summary as CSV', async () => {
+ it('exports the dead-letter messages of a group as JSON', async () => {
+ vi.mocked(messageService.exportDLQMessages).mockResolvedValue(
+ new
Blob(['[{"msgId":"m1","topic":"%DLQ%cg-order","queueId":0,"offset":5}]'], {
+ type: 'application/json',
+ }),
+ );
const user = userEvent.setup();
renderWithProviders(<DLQPage />);
await screen.findByText('cg-order');
await user.click(screen.getByRole('button', { name: '导出' }));
+ expect(messageService.exportDLQMessages).toHaveBeenCalledTimes(1);
+ const exportParams =
vi.mocked(messageService.exportDLQMessages).mock.calls[0][0];
+ expect(exportParams.instanceId).toBe('instance-1');
+ expect(exportParams.groupName).toBe('cg-order');
+ expect(typeof exportParams.startTime).toBe('number');
+ expect(typeof exportParams.endTime).toBe('number');
+ // Default export window is the last day, mirroring the visible range
picker.
+ expect(exportParams.endTime! - exportParams.startTime!).toBeGreaterThan(23
* 3600_000);
expect(createObjectURL).toHaveBeenCalledTimes(1);
const blob = createObjectURL.mock.calls[0][0] as Blob;
- await
expect(blob.text()).resolves.toContain('"cg-order","%DLQ%cg-order","7","3","ACTIVE"');
+ await expect(blob.text()).resolves.toContain('"msgId":"m1"');
expect(clickSpy).toHaveBeenCalledTimes(1);
expect(revokeObjectURL).toHaveBeenCalledWith('blob:dlq');
});
@@ -228,7 +242,7 @@ describe('DLQ page', () => {
expect(revokeObjectURL).toHaveBeenCalledWith('blob:dlq');
});
- it('neutralizes formulas hidden behind a leading line feed in CSV exports',
async () => {
+ it('neutralizes formulas hidden behind a leading line feed in CSV summary
exports', async () => {
vi.mocked(messageService.listDLQGroups).mockResolvedValue([
{
...dlqGroup,
@@ -241,7 +255,8 @@ describe('DLQ page', () => {
const row = (await screen.findByText('%DLQ%formula')).closest('tr');
if (!row) throw new Error('DLQ group row not found');
- await user.click(within(row).getByRole('button', { name: '导出' }));
+ await user.click(within(row).getByRole('checkbox'));
+ await user.click(screen.getByRole('button', { name: /批量导出/ }));
const blob = createObjectURL.mock.calls[0][0] as Blob;
await expect(blob.text()).resolves.toContain('"\'\n=1+1","%DLQ%formula"');
diff --git a/web/src/pages/instance/dlq.tsx b/web/src/pages/instance/dlq.tsx
index 583b387f0..56c9e17c5 100644
--- a/web/src/pages/instance/dlq.tsx
+++ b/web/src/pages/instance/dlq.tsx
@@ -37,7 +37,7 @@ import PageHeader from '../../components/PageHeader';
import { InstanceSelect } from '../../components/InstanceSelect';
import { useLang } from '../../i18n/LangContext';
import type { DLQGroup } from '../../api/message';
-import { listDLQGroups, resendDLQ } from '../../services/messageService';
+import { exportDLQMessages, listDLQGroups, resendDLQ } from
'../../services/messageService';
import { useInstanceFilter } from '../../hooks/useInstanceFilter';
import { buildCsv, downloadCsv, type CsvColumn } from '../../utils/download';
@@ -106,6 +106,13 @@ const DLQPage = () => {
dayjs().subtract(1, 'day'),
dayjs(),
]);
+ // Time range applied to dead-letter message exports. Kept page-level and
+ // visible so users know exactly which window the export covers instead of
+ // silently falling back to the backend's last-hour default.
+ const [exportRange, setExportRange] = useState<[Dayjs, Dayjs]>([
+ dayjs().subtract(1, 'day'),
+ dayjs(),
+ ]);
const [retryTargetTopic, setRetryTargetTopic] = useState('');
const [retrySubmitting, setRetrySubmitting] = useState(false);
const [detailGroup, setDetailGroup] = useState<DLQGroup | null>(null);
@@ -256,9 +263,24 @@ const DLQPage = () => {
}
};
- const handleExport = (group: DLQGroup) => {
- exportDLQGroups([group], `${group.groupName}-dlq.csv`);
- message.success(`已导出 ${group.groupName} 的死信队列摘要`);
+ const handleExport = async (group: DLQGroup) => {
+ try {
+ const blob = await exportDLQMessages({
+ instanceId: selectedInstanceId,
+ groupName: group.groupName,
+ startTime: exportRange[0].valueOf(),
+ endTime: exportRange[1].valueOf(),
+ });
+ const url = URL.createObjectURL(blob);
+ const link = document.createElement('a');
+ link.href = url;
+ link.download = `${group.groupName}-dlq-messages.json`;
+ link.click();
+ URL.revokeObjectURL(url);
+ message.success(`已导出 ${group.groupName} 的死信消息（${blob.size} 字节）`);
+ } catch (error) {
+ message.error(getErrorMessage(error, '导出死信消息失败，请稍后重试'));
+ }
};
const handleBatchExport = () => {
@@ -389,6 +411,22 @@ const DLQPage = () => {
style={{ width: 320 }}
prefix={<MagnifyingGlass size={14} color="#9CA3AF" />}
/>
+ <Space size={8}>
+ <Text type="secondary" style={{ fontSize: 14 }}>
+ 导出时间范围
+ </Text>
+ <RangePicker
+ showTime
+ format="YYYY-MM-DD HH:mm:ss"
+ value={exportRange}
+ onChange={(vals) => {
+ if (vals && vals[0] && vals[1]) {
+ setExportRange([vals[0], vals[1]]);
+ }
+ }}
+ style={{ width: 380 }}
+ />
+ </Space>
</Space>
<Button
icon={<Download size={16} />}
diff --git a/web/src/pages/studio/Proxy.tsx b/web/src/pages/studio/Proxy.tsx
index 17b797248..c7665bd4f 100644
--- a/web/src/pages/studio/Proxy.tsx
+++ b/web/src/pages/studio/Proxy.tsx
@@ -45,7 +45,12 @@ import {
} from '@phosphor-icons/react';
import PageHeader from '../../components/PageHeader';
import { useLang } from '../../i18n/LangContext';
-import { queryProxyHomePage, reloadProxyConfig, type ProxyNode } from
'../../api/proxy';
+import {
+ getProxyTopology,
+ queryProxyHomePage,
+ reloadProxyConfig,
+ type ProxyNode,
+} from '../../api/proxy';
const { Text } = Typography;
@@ -84,7 +89,7 @@ const ProxyPage: React.FC = () => {
try {
const { proxyAddrList, currentProxyAddr } = await queryProxyHomePage();
if (requestId !== loadRequestId.current) return false;
- const nodes: ProxyNode[] = (proxyAddrList || []).map((addr) => ({
+ const baseNodes: ProxyNode[] = (proxyAddrList || []).map((addr) => ({
key: addr,
address: addr,
status: 'unknown' as const,
@@ -96,6 +101,30 @@ const ProxyPage: React.FC = () => {
uptime: null,
isSelected: addr === currentProxyAddr,
}));
+ let nodes = baseNodes;
+
+ // Overlay the live TCP health view (UP/PARTIAL/DOWN) when the backend
exposes it.
+ try {
+ const topology = await getProxyTopology();
+ if (requestId !== loadRequestId.current) return false;
+ const statusByAddr = new Map(
+ topology.map((node) => [node.proxyAddr, node.status]),
+ );
+ nodes = baseNodes.map((node) => {
+ const probeStatus = statusByAddr.get(node.address);
+ const status: ProxyNode['status'] =
+ probeStatus === 'UP'
+ ? 'healthy'
+ : probeStatus === 'PARTIAL'
+ ? 'warning'
+ : probeStatus === 'DOWN'
+ ? 'unhealthy'
+ : node.status;
+ return { ...node, status };
+ });
+ } catch {
+ // Health probing is best-effort; keep the unknown status when it is
unavailable.
+ }
setProxyNodes(nodes);
setClusterStats({
diff --git a/web/src/services/messageService.ts
b/web/src/services/messageService.ts
index 81a0de42b..72c060fb4 100644
--- a/web/src/services/messageService.ts
+++ b/web/src/services/messageService.ts
@@ -75,3 +75,14 @@ export async function resendDLQ(data: {
if (isMockMode()) return { matched: 0, resent: 0, failed: 0, outcome:
'SUCCESS' };
return messageApi.resendDLQ(data);
}
+
+export async function exportDLQMessages(params: {
+ instanceId: string;
+ groupName: string;
+ startTime?: number;
+ endTime?: number;
+ maxCount?: number;
+}): Promise<Blob> {
+ if (isMockMode()) return new Blob(['[]'], { type: 'application/json' });
+ return messageApi.exportDLQMessages(params);
+}
