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 bc8de77c fix: batch studio bug fixes across trace, ACL, auth, alert
and SSRF (#1056)
bc8de77c is described below
commit bc8de77c58e5447e77a29ed6073cde02a0d4449d
Author: yyqdbngt <[email protected]>
AuthorDate: Thu Aug 6 14:26:37 2026 +0800
fix: batch studio bug fixes across trace, ACL, auth, alert and SSRF (#1056)
Co-authored-by: yyqdbngt <[email protected]>
---
.../rocketmq/studio/auth/AuthInterceptor.java | 35 ++++++++-
.../apache/rocketmq/studio/auth/AuthWebConfig.java | 22 ++++--
.../studio/instance/acl/AclController.java | 2 +-
.../rocketmq/studio/instance/acl/AclService.java | 4 +-
.../studio/instance/acl/UpdateAclUserDTO.java | 8 +-
.../apache/rocketmq/studio/ops/ai/AiService.java | 12 +++
.../rocketmq/studio/ops/ai/CliAgentProvider.java | 71 +++++++++++-------
.../rocketmq/studio/ops/alert/AlertService.java | 35 ++++++---
.../studio/rocketmq/RocketMQAdminClientImpl.java | 15 +++-
.../studio/rocketmq/RocketMQMessageProvider.java | 47 ++++++------
.../studio/rocketmq/RocketMQMetadataProvider.java | 4 +-
.../rocketmq/studio/settings/SettingsService.java | 33 ++++++++-
.../rocketmq/studio/auth/AuthControllerTest.java | 4 +
.../studio/auth/AuthCorsIntegrationTest.java | 4 +
.../rocketmq/studio/auth/AuthInterceptorTest.java | 85 ++++++++++++++++++++--
.../studio/instance/acl/AclControllerTest.java | 11 ++-
.../studio/instance/acl/AclServiceTest.java | 64 ++++++++++------
.../rocketmq/studio/ops/ai/AiServiceTest.java | 20 +++++
.../studio/ops/ai/CliAgentProviderTest.java | 66 +++++++++++++++++
.../studio/ops/alert/AlertServiceTest.java | 31 ++++++++
.../rocketmq/RocketMQAdminClientImplTest.java | 50 +++++++++++++
.../rocketmq/RocketMQMessageProviderTest.java | 68 +++++++++++++++++
.../rocketmq/RocketMQMetadataProviderTest.java | 79 ++++++++++++++++++++
.../studio/settings/SettingsServiceTest.java | 26 +++++++
24 files changed, 683 insertions(+), 113 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
index be8aeff9..f642319b 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
@@ -20,6 +20,8 @@ package org.apache.rocketmq.studio.auth;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import lombok.RequiredArgsConstructor;
+import org.apache.rocketmq.studio.settings.GeneralSettingsVO;
+import org.apache.rocketmq.studio.settings.SettingsRepository;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
@@ -37,17 +39,17 @@ public class AuthInterceptor implements HandlerInterceptor {
"/api/ai/chat",
"/api/clusters/test-connection",
"/api/llm/config/test",
- "/api/metrics/query",
- "/api/settings/datasources/test");
+ "/api/metrics/query");
private final AuthProperties authProperties;
private final AuthService authService;
+ private final SettingsRepository settingsRepository;
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse
response,
Object handler) throws Exception {
AuthenticatedUserContext.clear();
- if (!authProperties.isLoginRequired() ||
CorsUtils.isPreFlightRequest(request)
+ if (!isLoginRequired() || CorsUtils.isPreFlightRequest(request)
|| isPublicPath(requestPath(request))) {
return true;
}
@@ -65,15 +67,40 @@ public class AuthInterceptor implements HandlerInterceptor {
return true;
}
+ /**
+ * Login enforcement comes from the static {@code
studio.auth.login-required} property OR the
+ * runtime "requireLogin" toggle persisted in the settings database, so
toggling it in the
+ * settings UI actually changes the enforced policy.
+ */
+ private boolean isLoginRequired() {
+ if (authProperties != null && authProperties.isLoginRequired()) {
+ return true;
+ }
+ if (settingsRepository == null) {
+ return false;
+ }
+ try {
+ GeneralSettingsVO settings =
settingsRepository.loadGeneralSettings();
+ return settings != null && settings.isRequireLogin();
+ } catch (Exception exception) {
+ return false;
+ }
+ }
+
private boolean requiresAdmin(HttpServletRequest request, String path) {
String method = request.getMethod();
if (HttpMethod.GET.matches(method) || HttpMethod.HEAD.matches(method)
|| HttpMethod.OPTIONS.matches(method)) {
- return false;
+ // Read endpoints stay open to readers, except credential views
that expose secrets.
+ return isAdminOnlyGetPath(path);
}
return !HttpMethod.POST.matches(method) ||
!READER_POST_PATHS.contains(normalizePath(path));
}
+ private boolean isAdminOnlyGetPath(String path) {
+ return path.startsWith("/api/acl/users/") &&
path.endsWith("/credentials");
+ }
+
private void writeError(HttpServletResponse response, HttpStatus status,
String message)
throws Exception {
response.setStatus(status.value());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthWebConfig.java
b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthWebConfig.java
index c89d0485..f4b1aec4 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthWebConfig.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthWebConfig.java
@@ -18,21 +18,33 @@
package org.apache.rocketmq.studio.auth;
import lombok.RequiredArgsConstructor;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.apache.rocketmq.studio.settings.SettingsRepository;
+import org.springframework.beans.factory.ObjectProvider;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
+/**
+ * The interceptor is always registered: enforcement itself decides whether
login is required,
+ * merging the static {@code studio.auth.login-required} property with the
runtime "requireLogin"
+ * toggle from the settings database. A conditional registration would make
the UI toggle a no-op
+ * whenever the static property is false.
+ */
@Configuration
@RequiredArgsConstructor
-@ConditionalOnProperty(prefix = "studio.auth", name = "login-required",
havingValue = "true")
public class AuthWebConfig implements WebMvcConfigurer {
- private final AuthProperties authProperties;
- private final AuthService authService;
+ private final ObjectProvider<AuthProperties> authPropertiesProvider;
+ private final ObjectProvider<AuthService> authServiceProvider;
+ private final ObjectProvider<SettingsRepository>
settingsRepositoryProvider;
@Override
public void addInterceptors(InterceptorRegistry registry) {
- registry.addInterceptor(new AuthInterceptor(authProperties,
authService)).addPathPatterns("/api/**");
+ // Slice tests and minimal contexts may not provide any of these
beans; when they are
+ // missing the interceptor falls back to the static login-required
property (effectively
+ // no enforcement), matching the old conditional-registration
behaviour.
+ registry.addInterceptor(new
AuthInterceptor(authPropertiesProvider.getIfAvailable(),
+ authServiceProvider.getIfAvailable(),
settingsRepositoryProvider.getIfAvailable()))
+ .addPathPatterns("/api/**");
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclController.java
index 39d3894f..4d25bf71 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclController.java
@@ -77,7 +77,7 @@ public class AclController {
@PostMapping("/users/update")
public Result<AclUserVO> updateUser(@Valid @RequestBody(required = false)
UpdateAclUserDTO user) {
- return Result.ok(aclService.updateUser(requireRequest(user, "ACL user
request is required").toAclUserVO()));
+ return Result.ok(aclService.updateUser(requireRequest(user, "ACL user
request is required")));
}
@PostMapping("/users/delete")
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
index a4772e1a..65c7995e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
@@ -93,7 +93,7 @@ public class AclService {
return aclRepository.saveUser(user);
}
- public AclUserVO updateUser(AclUserVO user) {
+ public AclUserVO updateUser(UpdateAclUserDTO user) {
if (isBlank(user.getId())) {
throw new BusinessException(400, "ACL user id is required");
}
@@ -105,7 +105,7 @@ public class AclService {
.username(user.getUsername() == null ? existing.getUsername()
: user.getUsername())
.accessKey(existing.getAccessKey())
.secretKey(existing.getSecretKey())
- .admin(user.isAdmin())
+ .admin(user.getAdmin() == null ? existing.isAdmin() :
user.getAdmin())
.clusters(user.getClusters() == null ? existing.getClusters()
: user.getClusters())
.createdAt(existing.getCreatedAt())
.build();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/UpdateAclUserDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/UpdateAclUserDTO.java
index 19d19e00..52dd97fb 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/UpdateAclUserDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/UpdateAclUserDTO.java
@@ -26,14 +26,18 @@ public class UpdateAclUserDTO {
@NotBlank(message = "id is required")
private String id;
private String username;
- private boolean admin;
+ /**
+ * Null when the admin flag was not part of the partial update, in which
case the existing
+ * value must be preserved instead of being reset to {@code false}.
+ */
+ private Boolean admin;
private List<String> clusters;
public AclUserVO toAclUserVO() {
return AclUserVO.builder()
.id(id)
.username(username)
- .admin(admin)
+ .admin(admin != null && admin)
.clusters(clusters)
.build();
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/AiService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/AiService.java
index a4394535..00033143 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/AiService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/AiService.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.ops.ai;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@@ -34,12 +35,23 @@ public class AiService {
public SseEmitter chat(ChatDTO request) {
+ if (request == null) {
+ log.warn("Chat request body is missing");
+ throw new BusinessException(400, "Chat request is required");
+ }
log.info("Chat request received: mode={}, conversationId={}",
request.getMode(), request.getConversationId());
return llmGateway.chat(request);
}
public AiExecuteResultVO execute(AiCommandDTO command) {
+ if (command == null) {
+ log.warn("AI command body is missing");
+ return AiExecuteResultVO.builder()
+ .success(false)
+ .result("Command request is required")
+ .build();
+ }
log.info("Executing AI command: {}", command.getCommand());
try {
String result = llmGateway.execute(command);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/CliAgentProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/CliAgentProvider.java
index ccec3ec4..47b37d0c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/CliAgentProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/CliAgentProvider.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
/**
@@ -68,41 +69,61 @@ public abstract class CliAgentProvider implements
AgentProvider {
ProcessBuilder builder = new ProcessBuilder(command);
Map<String, String> env = builder.environment();
env.putAll(childEnv(config));
- builder.redirectErrorStream(false);
+ // Merge stderr into stdout and drain the stream on a background
thread. Reading stdout
+ // then stderr sequentially on the caller thread deadlocks once the
child fills a pipe
+ // buffer (64 KiB), and a timeout that runs only after both reads can
never fire while the
+ // child stays alive. With the merged stream drained in the
background, waitFor can enforce
+ // the timeout and a hung child is destroyed instead of leaking the
caller thread.
+ builder.redirectErrorStream(true);
+ Process process;
try {
- Process process = builder.start();
- String stdout = new
String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8);
- String stderr = new
String(process.getErrorStream().readAllBytes(), StandardCharsets.UTF_8);
- boolean finished = process.waitFor(TIMEOUT_SECONDS,
TimeUnit.SECONDS);
- if (!finished) {
- process.destroyForcibly();
- throw new LlmGatewayException(504, "llm.provider.timeout",
- binaryName() + " CLI timed out after " +
TIMEOUT_SECONDS + "s",
- "Retry with a shorter prompt or check the gateway
latency.");
- }
- if (process.exitValue() != 0) {
- log.warn("{} CLI failed rc={}, stderr={}", binaryName(),
process.exitValue(), abbreviate(stderr));
- throw new LlmGatewayException(502, "llm.provider.cli_error",
- binaryName() + " CLI failed: " + abbreviate(
- StringUtils.hasText(stderr) ? stderr : stdout),
- "Check the provider credentials, base URL and model
name.");
- }
- String result = stdout.trim();
- if (!StringUtils.hasText(result)) {
- throw new LlmGatewayException(502,
"llm.provider.empty_completion",
- binaryName() + " CLI returned an empty completion",
- "Check the selected model and provider response.");
- }
- return result;
+ process = builder.start();
} catch (IOException exception) {
throw new LlmGatewayException(502, "llm.provider.io_error",
"Failed to execute " + binaryName() + " CLI",
"Check that the CLI binary is installed and executable.",
exception);
+ }
+ CompletableFuture<String> outputFuture =
CompletableFuture.supplyAsync(() -> {
+ try (java.io.InputStream in = process.getInputStream()) {
+ return new String(in.readAllBytes(), StandardCharsets.UTF_8);
+ } catch (IOException ignored) {
+ return "";
+ }
+ });
+ boolean finished;
+ try {
+ finished = process.waitFor(TIMEOUT_SECONDS, TimeUnit.SECONDS);
} catch (InterruptedException exception) {
+ process.destroyForcibly();
Thread.currentThread().interrupt();
throw new LlmGatewayException(502, "llm.provider.interrupted",
binaryName() + " CLI execution was interrupted", "Retry
the request.", exception);
}
+ String output;
+ try {
+ output = outputFuture.get(10, TimeUnit.SECONDS);
+ } catch (Exception exception) {
+ output = "";
+ }
+ if (!finished) {
+ process.destroyForcibly();
+ throw new LlmGatewayException(504, "llm.provider.timeout",
+ binaryName() + " CLI timed out after " + TIMEOUT_SECONDS +
"s",
+ "Retry with a shorter prompt or check the gateway
latency.");
+ }
+ if (process.exitValue() != 0) {
+ log.warn("{} CLI failed rc={}, output={}", binaryName(),
process.exitValue(), abbreviate(output));
+ throw new LlmGatewayException(502, "llm.provider.cli_error",
+ binaryName() + " CLI failed: " + abbreviate(output),
+ "Check the provider credentials, base URL and model
name.");
+ }
+ String result = output.trim();
+ if (!StringUtils.hasText(result)) {
+ throw new LlmGatewayException(502, "llm.provider.empty_completion",
+ binaryName() + " CLI returned an empty completion",
+ "Check the selected model and provider response.");
+ }
+ return result;
}
private String abbreviate(String value) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
index c9c7ea71..053bd567 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
@@ -21,7 +21,10 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
@Slf4j
@@ -47,20 +50,28 @@ public class AlertService {
StringBuilder yaml = new StringBuilder();
yaml.append("groups:\n");
int index = 1;
+ // Prometheus requires each group name to be unique, so rules sharing
a group must be
+ // emitted under a single " - name:" block instead of one block per
rule.
+ Map<String, List<PrometheusAlertRule>> rulesByGroup = new
LinkedHashMap<>();
for (PrometheusAlertRule rule : prometheusRules) {
- yaml.append(" - name: ").append(rule.group()).append('\n');
+ rulesByGroup.computeIfAbsent(rule.group(), key -> new
ArrayList<>()).add(rule);
+ }
+ for (Map.Entry<String, List<PrometheusAlertRule>> group :
rulesByGroup.entrySet()) {
+ yaml.append(" - name: ").append(group.getKey()).append('\n');
yaml.append(" rules:\n");
- yaml.append(" # Rule ").append(index++).append(":
").append(rule.alert()).append('\n');
- yaml.append(" - alert: ").append(rule.alert()).append('\n');
- yaml.append(" expr: ").append(rule.expr()).append('\n');
- yaml.append(" for: ").append(rule.duration()).append('\n');
- yaml.append(" labels:\n");
- yaml.append(" severity:
").append(rule.severity()).append('\n');
- yaml.append(" team: ").append(rule.team()).append('\n');
- yaml.append(" annotations:\n");
- yaml.append(" summary:
\"").append(escapeDoubleQuotedValue(rule.summary())).append("\"\n");
- yaml.append(" description:
\"").append(escapeDoubleQuotedValue(rule.description()))
- .append("\"\n");
+ for (PrometheusAlertRule rule : group.getValue()) {
+ yaml.append(" # Rule ").append(index++).append(":
").append(rule.alert()).append('\n');
+ yaml.append(" - alert:
").append(rule.alert()).append('\n');
+ yaml.append(" expr: ").append(rule.expr()).append('\n');
+ yaml.append(" for:
").append(rule.duration()).append('\n');
+ yaml.append(" labels:\n");
+ yaml.append(" severity:
").append(rule.severity()).append('\n');
+ yaml.append(" team:
").append(rule.team()).append('\n');
+ yaml.append(" annotations:\n");
+ yaml.append(" summary:
\"").append(escapeDoubleQuotedValue(rule.summary())).append("\"\n");
+ yaml.append(" description:
\"").append(escapeDoubleQuotedValue(rule.description()))
+ .append("\"\n");
+ }
}
return yaml.toString();
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
index 1706d2ed..f6d4e72b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
@@ -149,8 +149,13 @@ public class RocketMQAdminClientImpl implements
AdminClient {
int readQueues = topic.getReadQueues() > 0 ? topic.getReadQueues() : 8;
try {
+ String clusterName = getClusterName();
+ // Match on the (cluster_id, name) key: the same topic name can
exist in several
+ // clusters, and a name-only lookup would blow up with
TooManyResultsException.
RmqTopic existing = topicMapper.selectOne(
- new LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName,
topicName));
+ new LambdaQueryWrapper<RmqTopic>()
+ .eq(RmqTopic::getClusterId, clusterName)
+ .eq(RmqTopic::getName, topicName));
TopicPerm effectivePerm = topic.getPerm() != null
? topic.getPerm()
: existing == null ? TopicPerm.RW :
fromRocketMQPerm(existing.getPerm());
@@ -172,7 +177,6 @@ public class RocketMQAdminClientImpl implements AdminClient
{
// Persist to DB. Re-creating a topic that already has a record
(for example when
// rebuilding a broker route from the console) must update it
instead of failing on
// the unique (cluster_id, name) key.
- String clusterName = getClusterName();
RmqTopic entity = topicMapper.selectOne(new
LambdaQueryWrapper<RmqTopic>()
.eq(RmqTopic::getClusterId, clusterName)
.eq(RmqTopic::getName, topicName));
@@ -228,8 +232,13 @@ public class RocketMQAdminClientImpl implements
AdminClient {
int readQueues = topic.getReadQueues() > 0 ? topic.getReadQueues() : 8;
try {
+ // Match on the (cluster_id, name) key to avoid ambiguity when
several clusters share
+ // the same topic name (a name-only lookup would throw
TooManyResultsException).
+ String clusterName = getClusterName();
RmqTopic existing = topicMapper.selectOne(
- new LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName,
topicName));
+ new LambdaQueryWrapper<RmqTopic>()
+ .eq(RmqTopic::getClusterId, clusterName)
+ .eq(RmqTopic::getName, topicName));
TopicPerm effectivePerm = topic.getPerm() != null
? topic.getPerm()
: existing == null ? TopicPerm.RW :
fromRocketMQPerm(existing.getPerm());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
index c8626813..1f2bac1e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
@@ -286,11 +286,15 @@ public class RocketMQMessageProvider implements
MessageProvider {
if (fields.length == 0) {
continue;
}
- if (!targetMsgId.equals(field(fields, 5))) {
+ String traceType = fields[0].trim();
+ // The message id column differs by trace type: Pub/EndTransaction
encode the trace bean
+ // (msgId at index 5), while SubAfter in RocketMQ 5.3.3 places
msgId at index 2.
+ int msgIdIndex = "SubAfter".equals(traceType) ? 2 : 5;
+ if (!targetMsgId.equals(field(fields, msgIdIndex))) {
continue;
}
try {
- switch (fields[0].trim()) {
+ switch (traceType) {
case "Pub":
nodes.add(buildProduceNode(fields));
break;
@@ -311,49 +315,50 @@ public class RocketMQMessageProvider implements
MessageProvider {
}
}
- // Pub layout: type, time, region, group, topic, msgId, tags, keys,
storeHost, clientHost,
- // retryTimes, msgType, status, costTime
+ // Pub layout (RocketMQ 5.3.3 TraceDataEncoder): type, time, region,
group, topic, msgId,
+ // tags, keys, storeHost, bodyLength, costTime, msgType,
offsetMsgId, isSuccess
private TraceNodeVO buildProduceNode(String[] f) {
return TraceNodeVO.builder()
.title("produce")
.timestamp(parseLong(field(f, 1)))
- .status(parseBoolean(field(f, 12)) ? "finish" : "failed")
- .costTime(parseLong(field(f, 13)))
- .description("producer=" + field(f, 3) + ", clientHost=" +
field(f, 9)
- + ", storeHost=" + field(f, 8))
+ .status(parseBoolean(field(f, 13)) ? "finish" : "failed")
+ .costTime(parseLong(field(f, 10)))
+ .description("producer=" + field(f, 3) + ", storeHost=" +
field(f, 8))
.build();
}
- // SubAfter layout: type, time, region, group, traceId, msgId, retryTimes,
keys, costTime,
- // status, timestamp
+ // SubAfter layout (RocketMQ 5.3.3 TraceDataEncoder): type, requestId,
msgId, costTime,
+ // isSuccess, keys, contextCode, timeStamp, groupName. The
trailing
+ // timeStamp/groupName columns may be absent when the trace
has no region info,
+ // so lookups tolerate short lines.
private TraceNodeVO buildConsumeNode(String[] f) {
return TraceNodeVO.builder()
.title("consume")
- .timestamp(parseLong(field(f, 1)))
- .status(parseBoolean(field(f, 9)) ? "finish" : "failed")
- .costTime(parseLong(field(f, 8)))
- .description("group=" + field(f, 3) + ", retryTimes=" +
field(f, 6))
+ .timestamp(parseLong(field(f, 7)))
+ .status(parseBoolean(field(f, 4)) ? "finish" : "failed")
+ .costTime(parseLong(field(f, 3)))
+ .description("group=" + field(f, 8) + ", contextCode=" +
field(f, 6))
.build();
}
private ConsumerStatusVO buildConsumerStatus(String[] f) {
return ConsumerStatusVO.builder()
- .group(field(f, 3))
- .deliveryStatus(parseBoolean(field(f, 9)) ?
DeliveryStatus.success : DeliveryStatus.failed)
- .consumeTime(parseLong(field(f, 1)))
- .retryCount((int) parseLong(field(f, 6)))
+ .group(field(f, 8))
+ .deliveryStatus(parseBoolean(field(f, 4)) ?
DeliveryStatus.success : DeliveryStatus.failed)
+ .consumeTime(parseLong(field(f, 7)))
+ .retryCount(0)
.build();
}
- // EndTransaction layout: type, time, region, group, topic, msgId, tags,
keys, storeHost,
- // clientHost, retryTimes, msgType, status,
transactionId, txState
+ // EndTransaction layout (RocketMQ 5.3.3): type, time, region, group,
topic, msgId, tags,
+ // keys, storeHost, bodyLength, costTime, msgType,
transactionId, txState
private TraceNodeVO buildTransactionNode(String[] f) {
return TraceNodeVO.builder()
.title("endTransaction")
.timestamp(parseLong(field(f, 1)))
.status("finish")
.costTime(0L)
- .description("group=" + field(f, 3) + ", transactionState=" +
field(f, 14))
+ .description("group=" + field(f, 3) + ", transactionState=" +
field(f, 13))
.build();
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProvider.java
index fe8b2c5d..75f83344 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProvider.java
@@ -166,7 +166,9 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
vo.setName(entity.getName());
vo.setClusterId(entity.getClusterId());
vo.setInstanceId(entity.getInstanceId());
- vo.setConsumeType(parseConsumeType(entity.getMessageModel()));
+ // consumeType stores the real ConsumeType
("CLUSTERING"/"BROADCASTING"); messageModel
+ // holds the subscription mode ("Push"/"Pop") and is not a
ConsumeType.
+ vo.setConsumeType(parseConsumeType(entity.getConsumeType()));
vo.setRetryMaxTimes(entity.getMaxRetry() == null ? 0 :
entity.getMaxRetry());
vo.setCreatedAt(entity.getCreatedAt());
vo.setUpdatedAt(entity.getUpdatedAt());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
b/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
index ceda5b7c..2b5ec081 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/settings/SettingsService.java
@@ -33,11 +33,14 @@ import
org.springframework.web.client.RestClientResponseException;
import org.springframework.web.util.UriComponentsBuilder;
import java.io.IOException;
+import java.net.InetAddress;
+import java.net.SocketTimeoutException;
import java.net.URI;
import java.net.URISyntaxException;
-import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
import java.time.Duration;
import java.util.List;
+import java.util.Locale;
import java.util.Set;
import java.util.UUID;
@@ -227,12 +230,40 @@ public class SettingsService {
if (!"http".equalsIgnoreCase(baseUri.getScheme()) &&
!"https".equalsIgnoreCase(baseUri.getScheme())) {
throw new IllegalArgumentException("Data source URL must start
with http:// or https://";);
}
+ if (!isAllowedDataSourceHost(baseUri.getHost())) {
+ throw new IllegalArgumentException(
+ "Data source URL must not point to a local or private
address");
+ }
return UriComponentsBuilder.fromUriString(normalized + "/api/v1/query")
.queryParam("query", PROMETHEUS_TEST_QUERY)
.build()
.toUri();
}
+ /**
+ * SSRF guard: the test endpoint performs a server-side HTTP request to an
attacker-supplied
+ * URL. The hostname {@code localhost} and link-local addresses
(169.254.x.x, fe80:: — the
+ * cloud metadata range) are never legitimate Prometheus endpoints and are
rejected. Loopback
+ * IPs and private site-local ranges stay allowed because on-premise
Prometheus servers live
+ * on the internal network and the endpoint itself requires admin rights.
+ */
+ private boolean isAllowedDataSourceHost(String host) {
+ if (!StringUtils.hasText(host)) {
+ return false;
+ }
+ String normalized = host.toLowerCase(Locale.ROOT);
+ if ("localhost".equals(normalized)) {
+ return false;
+ }
+ try {
+ InetAddress address = InetAddress.getByName(normalized);
+ return !address.isAnyLocalAddress() &&
!address.isLinkLocalAddress();
+ } catch (UnknownHostException exception) {
+ // Unresolvable host: let the connection attempt surface the real
connectivity error.
+ return true;
+ }
+ }
+
private DataSourceTestResultVO prometheusSuccess(JsonNode response) {
if (response != null &&
"success".equals(response.path("status").asText())) {
return DataSourceTestResultVO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthControllerTest.java
index bed876d6..44d0fa93 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthControllerTest.java
@@ -18,6 +18,7 @@
package org.apache.rocketmq.studio.auth;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.settings.SettingsRepository;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import
org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
@@ -52,6 +53,9 @@ class AuthControllerTest {
@MockBean
private AuthProperties authProperties;
+ @MockBean
+ private SettingsRepository settingsRepository;
+
@Autowired
private ObjectMapper objectMapper;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthCorsIntegrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthCorsIntegrationTest.java
index 0d5701ae..5d02aba0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthCorsIntegrationTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthCorsIntegrationTest.java
@@ -20,6 +20,7 @@ package org.apache.rocketmq.studio.auth;
import org.apache.rocketmq.studio.common.config.CorsConfig;
import org.apache.rocketmq.studio.instance.InstanceController;
import org.apache.rocketmq.studio.instance.InstanceService;
+import org.apache.rocketmq.studio.settings.SettingsRepository;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -60,6 +61,9 @@ class AuthCorsIntegrationTest {
@MockBean
private AuthService authService;
+ @MockBean
+ private SettingsRepository settingsRepository;
+
@BeforeEach
void enableLoginProtection() {
when(authProperties.isLoginRequired()).thenReturn(true);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthInterceptorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthInterceptorTest.java
index d1f77071..9f2467e5 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthInterceptorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthInterceptorTest.java
@@ -41,7 +41,7 @@ class AuthInterceptorTest {
@Test
void shouldAllowRequestsWhenLoginIsDisabled() throws Exception {
AuthProperties properties = new AuthProperties();
- AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties));
+ AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties), settingsRepository());
MockHttpServletRequest request = new MockHttpServletRequest("GET",
"/api/clusters");
boolean allowed = interceptor.preHandle(request, new
MockHttpServletResponse(), new Object());
@@ -55,7 +55,7 @@ class AuthInterceptorTest {
void shouldRejectProtectedApiWithoutTokenWhenLoginIsEnabled() throws
Exception {
AuthProperties properties = new AuthProperties();
properties.setLoginRequired(true);
- AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties));
+ AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties), settingsRepository());
MockHttpServletRequest request = new MockHttpServletRequest("GET",
"/api/clusters");
MockHttpServletResponse response = new MockHttpServletResponse();
@@ -76,7 +76,7 @@ class AuthInterceptorTest {
user.setAdmin(true);
properties.setUsers(List.of(user));
AuthService authService = authService(properties);
- AuthInterceptor interceptor = new AuthInterceptor(properties,
authService);
+ AuthInterceptor interceptor = new AuthInterceptor(properties,
authService, settingsRepository());
LoginDTO login = new LoginDTO();
login.setUsername("admin");
login.setPassword("secret");
@@ -97,11 +97,25 @@ class AuthInterceptorTest {
.isEqualTo(AuthenticatedUserContext.SYSTEM_ACTOR);
}
+ @Test
+ void shouldEnforceLoginWhenDatabaseRequiresItEvenIfPropertyIsDisabled()
throws Exception {
+ AuthProperties properties = new AuthProperties();
+ AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties),
+ settingsRepositoryRequiringLogin());
+ MockHttpServletRequest request = new MockHttpServletRequest("GET",
"/api/clusters");
+ MockHttpServletResponse response = new MockHttpServletResponse();
+
+ boolean allowed = interceptor.preHandle(request, response, new
Object());
+
+ assertThat(allowed).isFalse();
+ assertThat(response.getStatus()).isEqualTo(401);
+ }
+
@Test
void shouldAllowLoginEndpointWhenLoginIsEnabled() throws Exception {
AuthProperties properties = new AuthProperties();
properties.setLoginRequired(true);
- AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties));
+ AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties), settingsRepository());
MockHttpServletRequest request = new MockHttpServletRequest("POST",
"/api/auth/login");
boolean allowed = interceptor.preHandle(request, new
MockHttpServletResponse(), new Object());
@@ -113,7 +127,7 @@ class AuthInterceptorTest {
void shouldAllowLoginEndpointWithTrailingSlashWhenLoginIsEnabled() throws
Exception {
AuthProperties properties = new AuthProperties();
properties.setLoginRequired(true);
- AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties));
+ AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties), settingsRepository());
MockHttpServletRequest request = new MockHttpServletRequest("POST",
"/api/auth/login/");
boolean allowed = interceptor.preHandle(request, new
MockHttpServletResponse(), new Object());
@@ -125,7 +139,7 @@ class AuthInterceptorTest {
void shouldAllowAuthStatusEndpointWhenLoginIsEnabled() throws Exception {
AuthProperties properties = new AuthProperties();
properties.setLoginRequired(true);
- AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties));
+ AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties), settingsRepository());
MockHttpServletRequest request = new MockHttpServletRequest("GET",
"/api/auth/status");
boolean allowed = interceptor.preHandle(request, new
MockHttpServletResponse(), new Object());
@@ -137,7 +151,7 @@ class AuthInterceptorTest {
void shouldAllowAuthStatusEndpointWithTrailingSlashWhenLoginIsEnabled()
throws Exception {
AuthProperties properties = new AuthProperties();
properties.setLoginRequired(true);
- AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties));
+ AuthInterceptor interceptor = new AuthInterceptor(properties,
authService(properties), settingsRepository());
MockHttpServletRequest request = new MockHttpServletRequest("GET",
"/api/auth/status/");
boolean allowed = interceptor.preHandle(request, new
MockHttpServletResponse(), new Object());
@@ -157,6 +171,22 @@ class AuthInterceptorTest {
return settingsRepository;
}
+ private SettingsRepository settingsRepository() {
+ SettingsRepository settingsRepository = mock(SettingsRepository.class);
+
when(settingsRepository.loadGeneralSettings()).thenReturn(GeneralSettingsVO.builder()
+ .requireLogin(false)
+ .build());
+ return settingsRepository;
+ }
+
+ private SettingsRepository settingsRepositoryRequiringLogin() {
+ SettingsRepository settingsRepository = mock(SettingsRepository.class);
+
when(settingsRepository.loadGeneralSettings()).thenReturn(GeneralSettingsVO.builder()
+ .requireLogin(true)
+ .build());
+ return settingsRepository;
+ }
+
@Test
void shouldAllowReadOnlyGetForNonAdminUser() throws Exception {
TestSession session = login(false);
@@ -207,6 +237,19 @@ class AuthInterceptorTest {
assertThat(allowed).isTrue();
}
+ @Test
+ void shouldRejectDataSourceTestForNonAdminUser() throws Exception {
+ TestSession session = login(false);
+ MockHttpServletRequest request = authenticatedRequest(
+ "POST", "/api/settings/datasources/test", session.token());
+ MockHttpServletResponse response = new MockHttpServletResponse();
+
+ boolean allowed = session.interceptor().preHandle(request, response,
new Object());
+
+ assertThat(allowed).isFalse();
+ assertThat(response.getStatus()).isEqualTo(403);
+ }
+
@Test
void shouldAllowLogoutForNonAdminUser() throws Exception {
TestSession session = login(false);
@@ -219,6 +262,32 @@ class AuthInterceptorTest {
assertThat(allowed).isTrue();
}
+ @Test
+ void shouldRejectCredentialsViewForNonAdminUser() throws Exception {
+ TestSession session = login(false);
+ MockHttpServletRequest request = authenticatedRequest(
+ "GET", "/api/acl/users/user-1/credentials", session.token());
+ MockHttpServletResponse response = new MockHttpServletResponse();
+
+ boolean allowed = session.interceptor().preHandle(request, response,
new Object());
+
+ assertThat(allowed).isFalse();
+ assertThat(response.getStatus()).isEqualTo(403);
+ assertThat(response.getContentAsString()).contains("Admin permission
required");
+ }
+
+ @Test
+ void shouldAllowCredentialsViewForAdminUser() throws Exception {
+ TestSession session = login(true);
+ MockHttpServletRequest request = authenticatedRequest(
+ "GET", "/api/acl/users/user-1/credentials", session.token());
+
+ boolean allowed = session.interceptor().preHandle(
+ request, new MockHttpServletResponse(), new Object());
+
+ assertThat(allowed).isTrue();
+ }
+
private TestSession login(boolean admin) {
AuthProperties properties = new AuthProperties();
properties.setLoginRequired(true);
@@ -232,7 +301,7 @@ class AuthInterceptorTest {
login.setUsername("test-user");
login.setPassword("secret");
String token = authService.login(login).getToken();
- return new TestSession(new AuthInterceptor(properties, authService),
token);
+ return new TestSession(new AuthInterceptor(properties, authService,
settingsRepository()), token);
}
private MockHttpServletRequest authenticatedRequest(String method, String
path, String token) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclControllerTest.java
index 8e66a1a2..f987ff9e 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclControllerTest.java
@@ -284,11 +284,10 @@ class AclControllerTest {
@Test
void updateUserShouldReturnMaskedUpdatedUser() throws Exception {
- AclUserVO input = AclUserVO.builder()
- .id("user-1")
- .username("admin")
- .admin(false)
- .build();
+ UpdateAclUserDTO input = new UpdateAclUserDTO();
+ input.setId("user-1");
+ input.setUsername("admin");
+ input.setAdmin(false);
AclUserVO updated = AclUserVO.builder()
.id("user-1")
.username("admin")
@@ -297,7 +296,7 @@ class AclControllerTest {
.admin(false)
.build();
- when(aclService.updateUser(any(AclUserVO.class))).thenReturn(updated);
+
when(aclService.updateUser(any(UpdateAclUserDTO.class))).thenReturn(updated);
mockMvc.perform(post("/api/acl/users/update")
.contentType(MediaType.APPLICATION_JSON)
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
index 739aa811..0a78fbcc 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
@@ -337,9 +337,8 @@ class AclServiceTest {
@Test
void updateUserShouldRequireId() {
- AclUserVO input = AclUserVO.builder()
- .username("newuser")
- .build();
+ UpdateAclUserDTO input = new UpdateAclUserDTO();
+ input.setUsername("newuser");
assertThatThrownBy(() -> aclService.updateUser(input))
.hasMessage("ACL user id is required");
@@ -347,13 +346,10 @@ class AclServiceTest {
@Test
void updateUserShouldSaveExistingUser() {
- AclUserVO input = AclUserVO.builder()
- .id("user-1")
- .username("newuser")
- .accessKey("client-access-key")
- .secretKey("client-secret-key")
- .admin(true)
- .build();
+ UpdateAclUserDTO input = new UpdateAclUserDTO();
+ input.setId("user-1");
+ input.setUsername("newuser");
+ input.setAdmin(true);
ArgumentCaptor<AclUserVO> captor =
ArgumentCaptor.forClass(AclUserVO.class);
when(aclRepository.findUserById("user-1")).thenReturn(Optional.of(existingUser));
@@ -372,11 +368,37 @@ class AclServiceTest {
}
@Test
- void updateUserShouldThrowWhenUserDoesNotExist() {
- AclUserVO input = AclUserVO.builder()
- .id("missing")
- .username("ghost")
+ void updateUserShouldPreserveAdminWhenNotProvided() {
+ AclUserVO adminUser = AclUserVO.builder()
+ .id("user-1")
+ .username("orders")
+ .accessKey("access-key-123456")
+ .secretKey("secret-key-987654")
+ .admin(true)
+ .clusters(List.of("cluster-a"))
.build();
+ UpdateAclUserDTO input = new UpdateAclUserDTO();
+ input.setId("user-1");
+ input.setUsername("renamed");
+ // admin intentionally left null: the existing admin flag must survive
the partial update.
+
+ ArgumentCaptor<AclUserVO> captor =
ArgumentCaptor.forClass(AclUserVO.class);
+
when(aclRepository.findUserById("user-1")).thenReturn(Optional.of(adminUser));
+ when(aclRepository.saveUser(any(AclUserVO.class))).thenAnswer(inv ->
inv.getArgument(0));
+
+ AclUserVO result = aclService.updateUser(input);
+
+ assertThat(result.getUsername()).isEqualTo("renamed");
+ assertThat(result.isAdmin()).isTrue();
+ verify(aclRepository).saveUser(captor.capture());
+ assertThat(captor.getValue().isAdmin()).isTrue();
+ }
+
+ @Test
+ void updateUserShouldThrowWhenUserDoesNotExist() {
+ UpdateAclUserDTO input = new UpdateAclUserDTO();
+ input.setId("missing");
+ input.setUsername("ghost");
when(aclRepository.findUserById("missing")).thenReturn(Optional.empty());
@@ -448,14 +470,12 @@ class AclServiceTest {
String secretKey = created.getSecretKey();
AclUserVO listed = aclService.listUsers().get(0);
- AclUserVO updated = aclService.updateUser(AclUserVO.builder()
- .id(listed.getId())
- .username("orders-admin")
- .accessKey(listed.getAccessKey())
- .secretKey(listed.getSecretKey())
- .admin(true)
- .clusters(listed.getClusters())
- .build());
+ UpdateAclUserDTO update = new UpdateAclUserDTO();
+ update.setId(listed.getId());
+ update.setUsername("orders-admin");
+ update.setAdmin(true);
+ update.setClusters(listed.getClusters());
+ AclUserVO updated = aclService.updateUser(update);
assertThat(listed.getAccessKey()).isNotEqualTo(accessKey);
assertThat(listed.getSecretKey()).isNotEqualTo(secretKey);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/AiServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/AiServiceTest.java
index a10c2a8c..d98a751f 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/AiServiceTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/AiServiceTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.studio.ops.ai;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -30,8 +31,10 @@ import java.util.List;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -203,4 +206,21 @@ class AiServiceTest {
assertThat(result).isSameAs(output);
verify(mcpServerRegistry).execute("rmq.capabilities", input);
}
+
+ @Test
+ void chatRejectsNullRequest() {
+ assertThatThrownBy(() -> aiService.chat(null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Chat request is required");
+ verifyNoInteractions(llmGateway);
+ }
+
+ @Test
+ void executeHandlesNullCommand() {
+ AiExecuteResultVO result = aiService.execute(null);
+
+ assertThat(result.isSuccess()).isFalse();
+ assertThat(result.getResult()).isEqualTo("Command request is
required");
+ verifyNoInteractions(llmGateway);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/CliAgentProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/CliAgentProviderTest.java
new file mode 100644
index 00000000..9a1fcdcd
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/CliAgentProviderTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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
+ */
+package org.apache.rocketmq.studio.ops.ai;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class CliAgentProviderTest {
+
+ private static final class FakeCli extends CliAgentProvider {
+ private final String script;
+
+ FakeCli(String script) {
+ this.script = script;
+ }
+
+ @Override
+ public String engine() {
+ return "fake";
+ }
+
+ @Override
+ protected List<String> buildCommand(LlmConfigVO config, String prompt,
String modelOverride) {
+ return List.of("sh", "-c", script);
+ }
+
+ @Override
+ protected Map<String, String> childEnv(LlmConfigVO config) {
+ return Map.of();
+ }
+
+ @Override
+ protected String binaryName() {
+ return "sh";
+ }
+ }
+
+ @Test
+ void completeSurvivesLargeStderrOutput() {
+ // Write well over the 64 KiB pipe buffer to stderr, then print the
completion on stdout.
+ // Before the fix, sequential stdout/stderr reads deadlocked the
caller forever.
+ String script = "for i in $(seq 1 10000); do "
+ + "echo 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
>&2; done; "
+ + "echo DONE";
+ FakeCli cli = new FakeCli(script);
+
+ String result = cli.complete(null, "prompt", null);
+
+ assertThat(result).contains("DONE");
+ // All stderr bytes were drained as well (merged into the single
output stream),
+ // well past the 64 KiB pipe buffer that used to deadlock the
sequential reads.
+ assertThat(result.length()).isGreaterThan(500_000);
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
index 6ca3eba6..35a46555 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
@@ -113,6 +113,37 @@ class AlertServiceTest {
.contains("description: \"Lag too high\"");
}
+ @Test
+ void exportPrometheusRulesYamlShouldEmitSingleGroupForSameTeamRules() {
+ AlertRuleVO first = AlertRuleVO.builder()
+ .name("Lag Alert A")
+ .metric("rocketmq_consumer_lag_messages")
+ .operator(">")
+ .threshold(1000)
+ .duration("3m")
+ .description("First lag alert")
+ .build();
+ AlertRuleVO second = AlertRuleVO.builder()
+ .name("Lag Alert B")
+ .metric("rocketmq_consumer_lag_messages")
+ .operator(">")
+ .threshold(2000)
+ .duration("5m")
+ .description("Second lag alert")
+ .build();
+ when(alertRepository.findAllRules()).thenReturn(List.of(first,
second));
+
+ String result = alertService.exportPrometheusRulesYaml();
+
+ long groupOccurrences = result.split("- name:
rocketmq-consumer.rules", -1).length - 1;
+ assertThat(groupOccurrences).isEqualTo(1);
+ assertThat(result)
+ .contains("# Rule 1: LagAlertA")
+ .contains("# Rule 2: LagAlertB")
+ .contains("expr: rocketmq_consumer_lag_messages > 1000")
+ .contains("expr: rocketmq_consumer_lag_messages > 2000");
+ }
+
@Test
void
exportPrometheusRulesYamlShouldRenderReplicationLagRuleWithScopeAndSeverity() {
AlertRuleVO rule = AlertRuleVO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImplTest.java
index 3d5ca7cb..46a0f3d0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImplTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImplTest.java
@@ -10,24 +10,45 @@
*/
package org.apache.rocketmq.studio.rocketmq;
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.topic.TopicVO;
import org.apache.rocketmq.studio.ops.audit.AuditService;
+import org.apache.rocketmq.studio.persistence.entity.RmqTopic;
import org.apache.rocketmq.studio.persistence.mapper.RmqGroupMapper;
import org.apache.rocketmq.studio.persistence.mapper.RmqTopicMapper;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -82,4 +103,33 @@ class RocketMQAdminClientImplTest {
.isInstanceOf(BusinessException.class)
.hasMessageContaining("ACL denied");
}
+
+ @Test
+ void createTopicScopesLookupToCluster() throws Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+ ClusterInfo clusterInfo = new ClusterInfo();
+ Map<String, Set<String>> clusterAddrTable = new HashMap<>();
+ clusterAddrTable.put("cluster-1", new HashSet<>(List.of("broker-1")));
+ clusterInfo.setClusterAddrTable(clusterAddrTable);
+ Map<String, BrokerData> brokerAddrTable = new HashMap<>();
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("broker-1");
+ brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
+ brokerAddrTable.put("broker-1", brokerData);
+ clusterInfo.setBrokerAddrTable(brokerAddrTable);
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo);
+ when(topicMapper.selectOne(any())).thenReturn(null);
+ doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(),
any(TopicConfig.class));
+
+ TopicVO topic = new TopicVO();
+ topic.setName("topicA");
+ adminClient.createTopic(topic);
+
+ ArgumentCaptor<LambdaQueryWrapper<RmqTopic>> captor =
+ ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+ verify(topicMapper, times(2)).selectOne(captor.capture());
+ for (LambdaQueryWrapper<RmqTopic> wrapper : captor.getAllValues()) {
+ assertThat(wrapper.getSqlSegment()).contains("cluster_id");
+ }
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProviderTest.java
index 40520901..e8b783b1 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProviderTest.java
@@ -16,10 +16,14 @@
*/
package org.apache.rocketmq.studio.rocketmq;
+import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
+import org.apache.rocketmq.studio.instance.message.TraceNodeVO;
+import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
import org.apache.rocketmq.studio.queryhistory.QueryHistoryService;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.BeforeEach;
@@ -120,4 +124,68 @@ class RocketMQMessageProviderTest {
assertThat(record.getBody()).isEqualTo("wyg=");
assertThat(record.isBodyTruncated()).isFalse();
}
+
+ @Test
+ void getMessageTraceParsesPubAndSubAfterPerRocketMq533Layout() throws
Exception {
+ // Field order follows RocketMQ 5.3.3 TraceDataEncoder: Pub = type,
time, region, group,
+ // topic, msgId, tags, keys, storeHost, bodyLength, costTime, msgType,
offsetMsgId, isSuccess.
+ // SubAfter = type, requestId, msgId, costTime, isSuccess, keys,
contextCode, timeStamp,
+ // groupName.
+ String pub = "Pub" + '' + "1000" + '' + "cn" + '' + "prod-group"
+ + '' + "TopicA" + '' + "msg-123" + '' + "tag1" + '' +
"key1"
+ + '' + "broker:10911" + '' + "15" + '' + "50" + '' + "0"
+ + '' + "offset-1" + '' + "true";
+ String subAfter = "SubAfter" + '' + "req-1" + '' + "msg-123" + '' +
"20"
+ + '' + "true" + '' + "key1" + '' + "3" + '' + "3000"
+ + '' + "cons-group";
+ String otherMessage = "SubAfter" + '' + "req-2" + '' + "other-msg" +
'' + "5"
+ + '' + "false" + '' + "key-other" + '' + "0" + '' + "0"
+ + '' + "other-group";
+ MessageExt traceMessage = new MessageExt();
+ traceMessage.setBody(String.join("\n", pub, subAfter,
otherMessage).getBytes(StandardCharsets.UTF_8));
+ QueryResult queryResult = new QueryResult(0L, List.of(traceMessage));
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenReturn(queryResult);
+
+ TraceRecordVO record = provider.getMessageTrace("msg-123");
+
+ assertThat(record.getNodes()).hasSize(2);
+ TraceNodeVO produce = record.getNodes().get(0);
+ assertThat(produce.getTitle()).isEqualTo("produce");
+ assertThat(produce.getStatus()).isEqualTo("finish");
+ assertThat(produce.getCostTime()).isEqualTo(50);
+ assertThat(produce.getTimestamp()).isEqualTo(1000);
+
assertThat(produce.getDescription()).contains("prod-group").contains("broker:10911");
+ TraceNodeVO consume = record.getNodes().get(1);
+ assertThat(consume.getTitle()).isEqualTo("consume");
+ assertThat(consume.getStatus()).isEqualTo("finish");
+ assertThat(consume.getCostTime()).isEqualTo(20);
+ assertThat(consume.getTimestamp()).isEqualTo(3000);
+ assertThat(consume.getDescription()).contains("cons-group");
+ assertThat(record.getConsumerStatus()).hasSize(1);
+
assertThat(record.getConsumerStatus().get(0).getGroup()).isEqualTo("cons-group");
+ assertThat(record.getConsumerStatus().get(0).getDeliveryStatus())
+ .isEqualTo(DeliveryStatus.success);
+ }
+
+ @Test
+ void getMessageTraceParsesEndTransactionState() throws Exception {
+ String body = "EndTransaction" + '' + "2000" + '' + "cn" + '' +
"tx-group"
+ + '' + "TopicA" + '' + "msg-tx" + '' + "tag2" + '' + "key2"
+ + '' + "broker:10911" + '' + "10" + '' + "40" + '' + "0"
+ + '' + "tx-1" + '' + "COMMIT_MESSAGE";
+ MessageExt traceMessage = new MessageExt();
+ traceMessage.setBody(body.getBytes(StandardCharsets.UTF_8));
+ QueryResult queryResult = new QueryResult(0L, List.of(traceMessage));
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenReturn(queryResult);
+
+ TraceRecordVO record = provider.getMessageTrace("msg-tx");
+
+ assertThat(record.getNodes()).hasSize(1);
+ TraceNodeVO transaction = record.getNodes().get(0);
+ assertThat(transaction.getTitle()).isEqualTo("endTransaction");
+
assertThat(transaction.getDescription()).contains("tx-group").contains("COMMIT_MESSAGE");
+ assertThat(record.getConsumerStatus()).isEmpty();
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProviderTest.java
new file mode 100644
index 00000000..7f06e7cb
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProviderTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.rocketmq;
+
+import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.persistence.entity.RmqGroup;
+import org.apache.rocketmq.studio.persistence.mapper.RmqGroupMapper;
+import org.apache.rocketmq.studio.persistence.mapper.RmqTopicMapper;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class RocketMQMetadataProviderTest {
+
+ @Mock
+ private RmqTopicMapper topicMapper;
+
+ @Mock
+ private RmqGroupMapper groupMapper;
+
+ @Test
+ void listConsumerGroupsReadsConsumeTypeColumn() {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("group-broadcast");
+ entity.setClusterId("cluster-1");
+ entity.setConsumeType("BROADCASTING");
+ entity.setMessageModel("Push");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+
+ RocketMQMetadataProvider provider =
+ new RocketMQMetadataProvider(null, topicMapper, groupMapper);
+
+ List<ConsumerGroupVO> groups =
provider.listConsumerGroups("cluster-1", null);
+
+ assertThat(groups).hasSize(1);
+
assertThat(groups.get(0).getConsumeType()).isEqualTo(ConsumeType.BROADCASTING);
+ }
+
+ @Test
+ void listConsumerGroupsFallsBackToClusteringWhenConsumeTypeIsBlank() {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("group-legacy");
+ entity.setClusterId("cluster-1");
+ entity.setConsumeType(null);
+ entity.setMessageModel("Push");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+
+ RocketMQMetadataProvider provider =
+ new RocketMQMetadataProvider(null, topicMapper, groupMapper);
+
+ List<ConsumerGroupVO> groups =
provider.listConsumerGroups("cluster-1", null);
+
+ assertThat(groups).hasSize(1);
+
assertThat(groups.get(0).getConsumeType()).isEqualTo(ConsumeType.CLUSTERING);
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
index 5a4de053..25d4ac22 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/settings/SettingsServiceTest.java
@@ -373,6 +373,32 @@ class SettingsServiceTest {
assertThat(authorization.get()).isEqualTo("Bearer token-1");
}
+ @Test
+ void testConnectionShouldRejectLocalhostHostname() {
+ DataSourceTestDTO request = DataSourceTestDTO.builder()
+ .url("http://localhost:9090";)
+ .type("Prometheus")
+ .build();
+
+ DataSourceTestResultVO result =
settingsService.testDataSource(request);
+
+ assertThat(result.isSuccess()).isFalse();
+ assertThat(result.getMessage()).contains("local or private address");
+ }
+
+ @Test
+ void testConnectionShouldRejectLinkLocalMetadataAddress() {
+ DataSourceTestDTO request = DataSourceTestDTO.builder()
+ .url("http://169.254.169.254/latest/meta-data/";)
+ .type("Prometheus")
+ .build();
+
+ DataSourceTestResultVO result =
settingsService.testDataSource(request);
+
+ assertThat(result.isSuccess()).isFalse();
+ assertThat(result.getMessage()).contains("local or private address");
+ }
+
@Test
void testConnectionShouldRejectIncompleteBasicAuthentication() {
DataSourceTestResultVO result =
settingsService.testDataSource(DataSourceTestDTO.builder()
