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 f38f156a fix(provider): scope deletes to the selected cluster and add
regression coverage (#2034)
f38f156a is described below
commit f38f156a02b79e001e49edb9ecf7d99add965ed0
Author: xdz997 <[email protected]>
AuthorDate: Fri Aug 14 10:35:52 2026 +0800
fix(provider): scope deletes to the selected cluster and add regression
coverage (#2034)
Consolidates #2030, #2032, #2034, #2037, #2039. Implementation changes:
topic/group deletes and DLQ resend metadata handling are scoped to the
selected cluster, DLQ resend skips system properties, trace time windows
derive from message store time, and the no-redirect HTTP request factory
is shared from common/util instead of duplicated per package. Regression
tests added for system group filtering, OFFSET_ILLEGAL retries, alert and
message query validation, reset-offset guards and no-redirect factories.
---
.../AbstractPrometheusCompatibleMetricsSource.java | 14 +--
.../studio/cluster/proxy/ProxyAddressService.java | 14 +--
.../util/NoRedirectClientHttpRequestFactory.java | 31 ++++++
.../provider/apache/RocketMQAdminClientImpl.java | 12 ++-
.../provider/apache/RocketMQDLQProvider.java | 5 +-
.../provider/apache/RocketMQMessageProvider.java | 19 +++-
.../NoRedirectClientHttpRequestFactoryTest.java | 51 +++++++++
.../studio/common/util/SystemGroupFilterTest.java | 42 ++++++++
.../studio/common/util/SystemTopicFilterTest.java | 48 +++++++++
.../studio/ops/alert/AlertServiceTest.java | 45 ++++++++
.../apache/RocketMQAdminClientImplTest.java | 78 ++++++++++++++
.../provider/apache/RocketMQDLQProviderTest.java | 98 +++++++++++++++++
.../apache/RocketMQMessageProviderTest.java | 117 +++++++++++++++++++++
.../apache/RocketMQMetadataProviderTest.java | 26 +++++
.../studio/settings/SettingsServiceTest.java | 13 +++
15 files changed, 585 insertions(+), 28 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
index 4bf11e7d..b7d044b9 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
@@ -14,8 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.rocketmq.studio.cluster.metrics;
+import
org.apache.rocketmq.studio.common.util.NoRedirectClientHttpRequestFactory;
+
+
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
@@ -23,7 +27,6 @@ import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.HttpStatusCode;
import org.springframework.http.MediaType;
-import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.util.StringUtils;
@@ -68,14 +71,7 @@ public abstract class
AbstractPrometheusCompatibleMetricsSource implements Metri
protected AbstractPrometheusCompatibleMetricsSource(RestClient.Builder
restClientBuilder,
ObjectMapper
objectMapper,
MetricsSourceSettings
settings) {
- // Disable redirect following to prevent SSRF bypass via HTTP redirect
- SimpleClientHttpRequestFactory requestFactory = new
SimpleClientHttpRequestFactory() {
- @Override
- protected void prepareConnection(java.net.HttpURLConnection
connection, String httpMethod) throws IOException {
- super.prepareConnection(connection, httpMethod);
- connection.setInstanceFollowRedirects(false);
- }
- };
+ NoRedirectClientHttpRequestFactory requestFactory = new
NoRedirectClientHttpRequestFactory();
requestFactory.setConnectTimeout(settings.getConnectTimeout());
requestFactory.setReadTimeout(settings.getReadTimeout());
this.restClient =
restClientBuilder.requestFactory(requestFactory).build();
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 4804d267..71279b73 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
@@ -17,17 +17,19 @@
package org.apache.rocketmq.studio.cluster.proxy;
+import
org.apache.rocketmq.studio.common.util.NoRedirectClientHttpRequestFactory;
+
+
+
import org.apache.rocketmq.studio.common.exception.BusinessException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatusCode;
import org.springframework.http.ResponseEntity;
-import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.stereotype.Service;
import org.springframework.web.client.HttpStatusCodeException;
import org.springframework.web.client.ResourceAccessException;
import org.springframework.web.client.RestTemplate;
-import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
@@ -53,13 +55,7 @@ public class ProxyAddressService {
private final RestTemplate restTemplate;
public ProxyAddressService() {
- SimpleClientHttpRequestFactory factory = new
SimpleClientHttpRequestFactory() {
- @Override
- protected void prepareConnection(java.net.HttpURLConnection
connection, String httpMethod) throws IOException {
- super.prepareConnection(connection, httpMethod);
- connection.setInstanceFollowRedirects(false);
- }
- };
+ NoRedirectClientHttpRequestFactory factory = new
NoRedirectClientHttpRequestFactory();
factory.setConnectTimeout(Duration.ofSeconds(3));
factory.setReadTimeout(Duration.ofSeconds(3));
this.restTemplate = new RestTemplate(factory);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/common/util/NoRedirectClientHttpRequestFactory.java
b/server/src/main/java/org/apache/rocketmq/studio/common/util/NoRedirectClientHttpRequestFactory.java
new file mode 100644
index 00000000..bf89fc9e
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/common/util/NoRedirectClientHttpRequestFactory.java
@@ -0,0 +1,31 @@
+/*
+ * 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.common.util;
+
+import org.springframework.http.client.SimpleClientHttpRequestFactory;
+
+import java.io.IOException;
+import java.net.HttpURLConnection;
+
+public class NoRedirectClientHttpRequestFactory extends
SimpleClientHttpRequestFactory {
+
+ @Override
+ protected void prepareConnection(HttpURLConnection connection, String
httpMethod) throws IOException {
+ super.prepareConnection(connection, httpMethod);
+ connection.setInstanceFollowRedirects(false);
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
index d22accd8..31e58954 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
@@ -283,7 +283,8 @@ public class RocketMQAdminClientImpl implements AdminClient
{
String namesrvAddr = namesrvAddr(instanceId);
executeForInstance(instanceId, admin -> {
try {
- Set<String> brokerAddrs = getAllMasterBrokerAddrs(admin);
+ String clusterName = getClusterName(admin);
+ Set<String> brokerAddrs =
getMasterBrokerAddrsForCluster(admin, clusterName);
// Delete from brokers
if (!brokerAddrs.isEmpty()) {
@@ -298,11 +299,11 @@ public class RocketMQAdminClientImpl implements
AdminClient {
nsAddrs.add(trimmed);
}
}
- admin.deleteTopicInNameServer(nsAddrs, getClusterName(admin),
name);
+ admin.deleteTopicInNameServer(nsAddrs, clusterName, name);
// Topic names may be shared by several clusters managed by
this Studio instance.
topicMapper.delete(new LambdaQueryWrapper<RmqTopic>()
- .eq(RmqTopic::getClusterId, getClusterName(admin))
+ .eq(RmqTopic::getClusterId, clusterName)
.eq(RmqTopic::getName, name));
recordAudit("DELETE_TOPIC", name, "", "SUCCESS");
@@ -467,7 +468,8 @@ public class RocketMQAdminClientImpl implements AdminClient
{
private void doDeleteConsumerGroup(MQAdminExt admin, String name) {
try {
- Set<String> brokerAddrs = getAllMasterBrokerAddrs(admin);
+ String clusterName = getClusterName(admin);
+ Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin,
clusterName);
for (String addr : brokerAddrs) {
admin.deleteSubscriptionGroup(addr, name, true);
@@ -475,7 +477,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
// Consumer group names may be shared by several clusters managed
by this Studio instance.
groupMapper.delete(new LambdaQueryWrapper<RmqGroup>()
- .eq(RmqGroup::getClusterId, getClusterName(admin))
+ .eq(RmqGroup::getClusterId, clusterName)
.eq(RmqGroup::getName, name));
recordAudit("DELETE_GROUP", name, "", "SUCCESS");
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 87ad7901..d8d8660e 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
@@ -300,7 +300,10 @@ public class RocketMQDLQProvider implements DLQProvider {
if (userProperties != null) {
for (Map.Entry<String, String> entry :
userProperties.entrySet()) {
String key = entry.getKey();
- if (!MessageConst.STRING_HASH_SET.contains(key)) {
+ if (!MessageConst.STRING_HASH_SET.contains(key)
+ && !MessageConst.PROPERTY_TAGS.equals(key)
+ && !MessageConst.PROPERTY_KEYS.equals(key)
+ &&
!MessageConst.PROPERTY_WAIT_STORE_MSG_OK.equals(key)) {
message.putUserProperty(key, entry.getValue());
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
index 86702559..8d315b20 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
@@ -267,15 +267,15 @@ public class RocketMQMessageProvider implements
MessageProvider {
@Override
public TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic) {
return runtimeAdminClientResolver.execute(instanceId,
- adminExt -> getMessageTrace(instanceId, (DefaultMQAdminExt)
adminExt, msgId));
+ adminExt -> getMessageTrace(instanceId, (DefaultMQAdminExt)
adminExt, msgId, topic));
}
- private TraceRecordVO getMessageTrace(String instanceId, DefaultMQAdminExt
adminExt, String msgId) {
+ private TraceRecordVO getMessageTrace(String instanceId, DefaultMQAdminExt
adminExt, String msgId, String topic) {
long now = System.currentTimeMillis();
long begin;
long end;
- long messageStoreTimestamp = resolveMessageStoreTimestamp(adminExt,
msgId);
+ long messageStoreTimestamp = resolveMessageStoreTimestamp(adminExt,
msgId, topic);
if (messageStoreTimestamp > 0) {
// Derive the trace query window from the message's own store
timestamp
// instead of a hardcoded 1-hour lookback. This ensures traces for
messages
@@ -320,7 +320,18 @@ public class RocketMQMessageProvider implements
MessageProvider {
* query window can be derived from the message's own timeline rather than
the
* current time. Returns 0 if the message cannot be located.
*/
- private long resolveMessageStoreTimestamp(DefaultMQAdminExt adminExt,
String msgId) {
+ private long resolveMessageStoreTimestamp(DefaultMQAdminExt adminExt,
String msgId, String topic) {
+ if (StringUtils.hasText(topic)) {
+ try {
+ MessageExt messageExt = adminExt.viewMessage(topic, msgId);
+ if (messageExt != null) {
+ return messageExt.getStoreTimestamp();
+ }
+ } catch (Exception e) {
+ log.debug("Could not view message {} in topic {} for trace
timestamp: {}",
+ msgId, topic, e.getMessage());
+ }
+ }
try {
// No topic hint is available in the trace flow, so locate the
message purely
// by its offset msgId.
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/common/util/NoRedirectClientHttpRequestFactoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/common/util/NoRedirectClientHttpRequestFactoryTest.java
new file mode 100644
index 00000000..722ed53f
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/common/util/NoRedirectClientHttpRequestFactoryTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.common.util;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.http.HttpMethod;
+
+import java.net.HttpURLConnection;
+import java.net.URI;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class NoRedirectClientHttpRequestFactoryTest {
+
+ @Test
+ void prepareConnectionShouldDisableRedirectsTest() throws Exception {
+ HttpURLConnection connection = (HttpURLConnection)
URI.create("http://example.com";)
+ .toURL().openConnection();
+ TestableNoRedirectClientHttpRequestFactory requestFactory =
+ new TestableNoRedirectClientHttpRequestFactory();
+
+ assertThat(connection.getInstanceFollowRedirects()).isTrue();
+
+ requestFactory.prepare(connection, HttpMethod.GET.name());
+
+ assertThat(connection.getInstanceFollowRedirects()).isFalse();
+ connection.disconnect();
+ }
+
+ private static class TestableNoRedirectClientHttpRequestFactory
+ extends NoRedirectClientHttpRequestFactory {
+
+ void prepare(HttpURLConnection connection, String httpMethod) throws
Exception {
+ prepareConnection(connection, httpMethod);
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemGroupFilterTest.java
b/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemGroupFilterTest.java
new file mode 100644
index 00000000..7068d46e
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemGroupFilterTest.java
@@ -0,0 +1,42 @@
+/*
+ * 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.common.util;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class SystemGroupFilterTest {
+
+ @Test
+ void shouldRecognizeKnownSystemConsumerGroupPrefixesTest() {
+ assertThat(SystemGroupFilter.isSystem(null)).isTrue();
+ assertThat(SystemGroupFilter.isSystem("")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("CID_RMQ_SYS_TRANS")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("CID_ONSAPI_OWNER")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("CID_SYS_RMQ_TRANS")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("CID_HOUSEKEEPING")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("rmq_sys_TRACE_DATA")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("%RETRY%consumer-a")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("%DLQ%consumer-a")).isTrue();
+
assertThat(SystemGroupFilter.isSystem("TOOLS_CONSUMER_monitor")).isTrue();
+
assertThat(SystemGroupFilter.isSystem("FILTERSRV_CONSUMER_filter")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("SELF_TEST_GROUP")).isTrue();
+
+
assertThat(SystemGroupFilter.isSystem("order-service-consumer")).isFalse();
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemTopicFilterTest.java
b/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemTopicFilterTest.java
new file mode 100644
index 00000000..a3e0f5fa
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemTopicFilterTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.common.util;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class SystemTopicFilterTest {
+
+ @Test
+ void shouldRecognizeSharedSystemTopicPrefixesAndBrokerNamesTest() {
+ Set<String> brokerNames = Set.of("broker-prod-a", "broker-prod-b");
+
+ assertThat(SystemTopicFilter.isSystem(null, brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("", brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("RMQ_SYS_TRANS_HALF_TOPIC",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("rmq_sys_TRACE_DATA",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("SCHEDULE_TOPIC_XXXX",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("%RETRY%consumer-a",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("%DLQ%consumer-a",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("CID_RMQ_SYS_TRANS",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("broker_config",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("BenchmarkTestTopic",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("TBW102", brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("SELF_TEST_TOPIC",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("OFFSET_MOVED_EVENT",
brokerNames)).isTrue();
+ assertThat(SystemTopicFilter.isSystem("broker-prod-a",
brokerNames)).isTrue();
+
+ assertThat(SystemTopicFilter.isSystem("orders",
brokerNames)).isFalse();
+ }
+}
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 3c17f0c3..f4bc37b6 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
@@ -350,6 +350,51 @@ class AlertServiceTest {
assertThat(exportedRule.path("alert").asText()).isEqualTo("RocketMQAlert");
}
+ @Test
+ void exportPrometheusRulesYamlShouldPrefixAlertNamesStartingWithDigit()
throws Exception {
+ AlertRuleVO rule = AlertRuleVO.builder()
+ .name("5xx spike")
+ .metric("rocketmq_consumer_lag_messages")
+ .operator(">")
+ .threshold(1)
+ .enabled(true)
+ .build();
+ when(alertRepository.findAllRules()).thenReturn(List.of(rule));
+
+ String result = alertService.exportPrometheusRulesYaml();
+
+ JsonNode exportedRule = new ObjectMapper(new
YAMLFactory()).readTree(result)
+ .path("groups").get(0).path("rules").get(0);
+
assertThat(exportedRule.path("alert").asText()).isEqualTo("A_5xxspike");
+ assertThat(result).contains("# Rule 1: A_5xxspike");
+ }
+
+ @Test
+ void exportPrometheusRulesYamlShouldReplaceNonFiniteThresholds() throws
Exception {
+ AlertRuleVO nanRule = AlertRuleVO.builder()
+ .name("Bad Threshold A")
+ .metric("rocketmq_consumer_lag_messages")
+ .operator(">")
+ .threshold(Double.NaN)
+ .enabled(true)
+ .build();
+ AlertRuleVO infinityRule = AlertRuleVO.builder()
+ .name("Bad Threshold B")
+ .metric("rocketmq_consumer_lag_messages")
+ .operator("<")
+ .threshold(Double.POSITIVE_INFINITY)
+ .enabled(true)
+ .build();
+ when(alertRepository.findAllRules()).thenReturn(List.of(nanRule,
infinityRule));
+
+ String result = alertService.exportPrometheusRulesYaml();
+
+ JsonNode rules = new ObjectMapper(new
YAMLFactory()).readTree(result).path("groups").get(0).path("rules");
+
assertThat(rules.get(0).path("expr").asText()).isEqualTo("rocketmq_consumer_lag_messages
> 0");
+
assertThat(rules.get(1).path("expr").asText()).isEqualTo("rocketmq_consumer_lag_messages
< 0");
+ assertThat(result).doesNotContain("NaN", "Infinity");
+ }
+
@Test
void exportPrometheusRulesYamlShouldExcludeDisabledRules() {
AlertRuleVO enabled = AlertRuleVO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index b935b742..16834041 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
@@ -134,6 +134,18 @@ class RocketMQAdminClientImplTest {
"instanceId=instance-a, topic=orders,
timestamp=1784246400000", "SUCCESS");
}
+ @Test
+ void resetOffsetShouldRejectBlankTopicBeforeResolvingAdmin() {
+ assertThatThrownBy(() -> adminClient.resetOffset("instance-a",
"cg-orders", 1784246400000L, " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("topic is required for offset reset")
+ .satisfies(exception -> assertThat(((BusinessException)
exception).getCode()).isEqualTo(400));
+
+ verifyNoInteractions(runtimeAdminClientResolver);
+ verify(adminFactory, never()).execute(anyString(), any(), any());
+ verifyNoInteractions(auditService);
+ }
+
@Test
void getConsumerGroupSurfacesAdminTimeout() throws Exception {
when(adminExt.examineConsumerConnectionInfo("orders"))
@@ -183,6 +195,25 @@ class RocketMQAdminClientImplTest {
}
}
+ @Test
+ void
createTopicShouldOnlyWriteTargetClusterBrokersInMultiClusterTopology() throws
Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+ ClusterInfo clusterInfo = clusterInfoWithTwoClusters();
+ 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("orders");
+
+ adminClient.createTopic(topic);
+
+ verify(adminExt).createAndUpdateTopicConfig(
+ org.mockito.ArgumentMatchers.eq("10.0.0.1:10911"),
any(TopicConfig.class));
+ verify(adminExt, never()).createAndUpdateTopicConfig(
+ org.mockito.ArgumentMatchers.eq("10.0.1.1:10911"),
any(TopicConfig.class));
+ }
+
@Test
void topicWritesUseSelectedInstanceAdmin() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
@@ -249,6 +280,20 @@ class RocketMQAdminClientImplTest {
verify(selectedAdmin).deleteTopicInNameServer(Set.of("10.0.0.2:9876"),
"cluster-1", "orders");
}
+ @Test
+ void deleteTopicScopesBrokerDeletionToSelectedClusterOnly() throws
Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithTwoClusters());
+ doNothing().when(adminExt).deleteTopicInBroker(any(), anyString());
+ doNothing().when(adminExt).deleteTopicInNameServer(any(), anyString(),
anyString());
+
+ adminClient.deleteTopic(null, "orders");
+
+ verify(adminExt).deleteTopicInBroker(Set.of("10.0.0.1:10911"),
"orders");
+ verify(adminExt).deleteTopicInNameServer(Set.of("10.0.0.1:9876"),
"cluster-1", "orders");
+ verify(topicMapper).delete(any());
+ }
+
@Test
void createConsumerGroupUsesSelectedInstanceAdmin() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqGroup.class);
@@ -307,6 +352,20 @@ class RocketMQAdminClientImplTest {
assertThat(captor.getValue().getSqlSegment()).contains("cluster_id",
"name");
}
+ @Test
+ void deleteConsumerGroupScopesBrokerDeletionToSelectedClusterOnly() throws
Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqGroup.class);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithTwoClusters());
+ doNothing().when(adminExt).deleteSubscriptionGroup(anyString(),
anyString(),
+ org.mockito.ArgumentMatchers.anyBoolean());
+
+ adminClient.deleteConsumerGroup(null, "cg-orders");
+
+ verify(adminExt).deleteSubscriptionGroup("10.0.0.1:10911",
"cg-orders", true);
+ verify(adminExt, never()).deleteSubscriptionGroup("10.0.1.1:10911",
"cg-orders", true);
+ verify(groupMapper).delete(any());
+ }
+
@Test
void createTopicSucceedsWhenAuditRecordingFails() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
@@ -353,6 +412,25 @@ class RocketMQAdminClientImplTest {
return clusterInfo;
}
+ private ClusterInfo clusterInfoWithTwoClusters() {
+ ClusterInfo clusterInfo = new ClusterInfo();
+ Map<String, Set<String>> clusterAddrTable = new HashMap<>();
+ clusterAddrTable.put("cluster-1", new HashSet<>(List.of("broker-1")));
+ clusterAddrTable.put("cluster-2", new HashSet<>(List.of("broker-2")));
+ clusterInfo.setClusterAddrTable(clusterAddrTable);
+ Map<String, BrokerData> brokerAddrTable = new HashMap<>();
+ BrokerData firstBroker = new BrokerData();
+ firstBroker.setBrokerName("broker-1");
+ firstBroker.setBrokerAddrs(new HashMap<>(Map.of(0L,
"10.0.0.1:10911")));
+ BrokerData secondBroker = new BrokerData();
+ secondBroker.setBrokerName("broker-2");
+ secondBroker.setBrokerAddrs(new HashMap<>(Map.of(0L,
"10.0.1.1:10911")));
+ brokerAddrTable.put("broker-1", firstBroker);
+ brokerAddrTable.put("broker-2", secondBroker);
+ clusterInfo.setBrokerAddrTable(brokerAddrTable);
+ return clusterInfo;
+ }
+
@Test
void sendMessageShouldNotFailWhenAuditRecordingFails() throws Exception {
when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
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 a566b048..d43e5da2 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
@@ -24,6 +24,7 @@ import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
@@ -38,6 +39,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockedConstruction;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -226,6 +228,55 @@ class RocketMQDLQProviderTest {
}
}
+ @Test
+ void resendMessagesRetriesFromCorrectedOffsetAfterOffsetIllegal() 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("dlq-after-correction");
+ deadLetter.setTopic(dlqTopic);
+ deadLetter.setBody(new byte[] {1, 2, 3});
+ deadLetter.setStoreTimestamp(150L);
+ PullResult illegalOffset = new PullResult(PullStatus.OFFSET_ILLEGAL,
20L, 0L, 30L, null);
+ PullResult foundAfterCorrection = new PullResult(PullStatus.FOUND,
40L, 20L, 30L, List.of(deadLetter));
+ PullResult endOfQueue = new PullResult(PullStatus.NO_NEW_MSG, 50L,
40L, 40L, List.of());
+ SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.SEND_OK);
+
+ try (MockedConstruction<DefaultMQPullConsumer> mockedConsumers =
+ mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ doNothing().when(consumer).start();
+
when(consumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(consumer.searchOffset(queue,
100L)).thenReturn(10L);
+ when(consumer.searchOffset(queue,
200L)).thenReturn(50L);
+ when(consumer.pull(queue, "*", 10L,
32)).thenReturn(illegalOffset);
+ when(consumer.pull(queue, "*", 20L,
32)).thenReturn(foundAfterCorrection);
+ when(consumer.pull(queue, "*", 40L,
32)).thenReturn(endOfQueue);
+ doNothing().when(consumer).shutdown();
+ });
+ MockedConstruction<DefaultMQProducer> mockedProducers =
+ mockConstruction(DefaultMQProducer.class, (producer,
context) -> {
+ doNothing().when(producer).start();
+
when(producer.send(any(Message.class))).thenReturn(sendResult);
+ doNothing().when(producer).shutdown();
+ })) {
+ assertThat(provider.resendMessages("instance-a", "group-a", 100L,
200L, "orders"))
+ .extracting("matched", "resent", "failed", "outcome")
+ .containsExactly(1, 1, 0, "SUCCESS");
+
+ DefaultMQPullConsumer consumer =
mockedConsumers.constructed().get(0);
+ verify(consumer).pull(queue, "*", 10L, 32);
+ verify(consumer).pull(queue, "*", 20L, 32);
+ verify(consumer).pull(queue, "*", 40L, 32);
+
verify(mockedProducers.constructed().get(0)).send(any(Message.class));
+ }
+ verify(auditService).record(
+ eq("RESEND_DLQ"),
+ eq("group-a"),
+ contains("matched=1, resent=1, failed=0"),
+ eq("SUCCESS"));
+ }
+
@Test
void resendMessagesCountsNonSendOkResultsAsFailures() throws Exception {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
@@ -269,6 +320,53 @@ class RocketMQDLQProviderTest {
eq("FAILED"));
}
+ @Test
+ void resendMessagesShouldCopyUserPropertiesAndSkipSystemProperties()
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-with-properties");
+ deadLetter.setTopic(dlqTopic);
+ deadLetter.setBody(new byte[] {1});
+ deadLetter.setStoreTimestamp(150L);
+ deadLetter.putUserProperty("traceId", "trace-123");
+ deadLetter.putUserProperty("tenantId", "tenant-a");
+ deadLetter.getProperties().put(MessageConst.PROPERTY_REAL_TOPIC,
"system-topic-should-not-copy");
+ PullResult pullResult = new PullResult(PullStatus.FOUND, 1L, 0L, 0L,
List.of(deadLetter));
+ SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.SEND_OK);
+
+ try (MockedConstruction<DefaultMQPullConsumer> ignoredConsumers =
+ mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ doNothing().when(consumer).start();
+
when(consumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(consumer.searchOffset(queue,
100L)).thenReturn(0L);
+ when(consumer.searchOffset(queue,
200L)).thenReturn(0L);
+ when(consumer.pull(queue, "*", 0L,
32)).thenReturn(pullResult);
+ doNothing().when(consumer).shutdown();
+ });
+ MockedConstruction<DefaultMQProducer> mockedProducers =
+ mockConstruction(DefaultMQProducer.class, (producer,
context) -> {
+ doNothing().when(producer).start();
+
when(producer.send(any(Message.class))).thenReturn(sendResult);
+ doNothing().when(producer).shutdown();
+ })) {
+ assertThat(provider.resendMessages("instance-a", "group-a", 100L,
200L, "target-topic"))
+ .extracting("matched", "resent", "failed", "outcome")
+ .containsExactly(1, 1, 0, "SUCCESS");
+
+ ArgumentCaptor<Message> messageCaptor =
ArgumentCaptor.forClass(Message.class);
+
verify(mockedProducers.constructed().get(0)).send(messageCaptor.capture());
+ Message resent = messageCaptor.getValue();
+ assertThat(resent.getProperties()).containsEntry("traceId",
"trace-123")
+ .containsEntry("tenantId", "tenant-a")
+ .containsEntry("studio_dlq_origin_message_id",
"msg-with-properties")
+ .containsEntry("studio_dlq_origin_topic", dlqTopic);
+ assertThat(resent.getProperties())
+ .doesNotContainEntry(MessageConst.PROPERTY_REAL_TOPIC,
"system-topic-should-not-copy");
+ }
+ }
+
@Test
void createsUniqueProducerGroupsForConcurrentDlqResends() {
String first = RocketMQDLQProvider.nextResendProducerGroup();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
index 6ff3e36e..87bbed05 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
@@ -21,7 +21,9 @@ import
org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.client.trace.TraceConstants;
+import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
@@ -36,10 +38,12 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockedConstruction;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Set;
@@ -110,6 +114,36 @@ class RocketMQMessageProviderTest {
100L, 200L, 0);
}
+ @Test
+ void queryMessagesShouldRejectInvertedTimeRangeBeforeAdminLookup() throws
Exception {
+ assertThatThrownBy(() -> provider.queryMessages(
+ "instance-a", "TopicA", null, null, null, 200L, 100L))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Message query start time must be before end time")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+
+ verify(runtimeAdminClientResolver).resolveEndpoint("instance-a");
+ verify(runtimeAdminClientResolver).execute(eq("instance-a"), any());
+ verify(adminExt, never()).queryMessage(anyString(), anyString(),
anyInt(), anyLong(), anyLong());
+ verify(queryHistoryService, never()).recordMessageQuery(anyString(),
anyString(), anyString(),
+ anyString(), anyString(), anyString(), any(), any(), anyInt());
+ }
+
+ @Test
+ void queryMessagesShouldRejectEqualTimeRangeBeforeAdminLookup() throws
Exception {
+ assertThatThrownBy(() -> provider.queryMessages(
+ "instance-a", "TopicA", null, null, null, 100L, 100L))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Message query start time must be before end time")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+
+ verify(runtimeAdminClientResolver).resolveEndpoint("instance-a");
+ verify(runtimeAdminClientResolver).execute(eq("instance-a"), any());
+ verify(adminExt, never()).queryMessage(anyString(), anyString(),
anyInt(), anyLong(), anyLong());
+ verify(queryHistoryService, never()).recordMessageQuery(anyString(),
anyString(), anyString(),
+ anyString(), anyString(), anyString(), any(), any(), anyInt());
+ }
+
@Test
void queryByKeySurfacesAdminFailure() throws Exception {
when(adminExt.queryMessage("TopicA", "order-1", 64, 100L, 200L))
@@ -187,6 +221,41 @@ class RocketMQMessageProviderTest {
}
}
+ @Test
+ void queryByTopicRetriesFromCorrectedOffsetAfterOffsetIllegal() throws
Exception {
+ MessageQueue queue = new MessageQueue("TopicA", "broker-a", 0);
+ MessageExt message = new MessageExt();
+ message.setMsgId("msg-after-correction");
+ message.setTopic("TopicA");
+ message.setBody("payload".getBytes(StandardCharsets.UTF_8));
+ message.setStoreTimestamp(150L);
+ PullResult illegalOffset = new PullResult(PullStatus.OFFSET_ILLEGAL,
20L, 0L, 30L, null);
+ PullResult foundAfterCorrection = new PullResult(PullStatus.FOUND,
40L, 20L, 30L, List.of(message));
+ PullResult endOfQueue = new PullResult(PullStatus.NO_NEW_MSG, 50L,
40L, 40L, List.of());
+ try (MockedConstruction<DefaultMQPullConsumer> mockedConsumers =
+ mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ doNothing().when(consumer).start();
+
when(consumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of(queue));
+ when(consumer.searchOffset(queue,
100L)).thenReturn(10L);
+ when(consumer.searchOffset(queue,
200L)).thenReturn(50L);
+ when(consumer.pull(eq(queue), eq("*"), eq(10L),
eq(32))).thenReturn(illegalOffset);
+ when(consumer.pull(eq(queue), eq("*"), eq(20L),
eq(32))).thenReturn(foundAfterCorrection);
+ when(consumer.pull(eq(queue), eq("*"), eq(40L),
eq(32))).thenReturn(endOfQueue);
+ doNothing().when(consumer).shutdown();
+ })) {
+ List<MessageRecordVO> messages = provider.queryMessages(
+ "instance-a", "TopicA", null, null, null, 100L, 200L);
+
+
assertThat(messages).extracting(MessageRecordVO::getMsgId).containsExactly("msg-after-correction");
+ DefaultMQPullConsumer consumer =
mockedConsumers.constructed().get(0);
+ verify(consumer).pull(queue, "*", 10L, 32);
+ verify(consumer).pull(queue, "*", 20L, 32);
+ verify(consumer).pull(queue, "*", 40L, 32);
+ }
+ verify(queryHistoryService).recordMessageQuery("instance-a", "TOPIC",
"TopicA", null, null, null,
+ 100L, 200L, 1);
+ }
+
@Test
void toRecordVOBoundsMessageBodyAndProperties() {
MessageExt message = new MessageExt();
@@ -298,6 +367,54 @@ class RocketMQMessageProviderTest {
assertThat(record.getConsumerStatus()).isEmpty();
}
+ @Test
+ void getMessageTraceShouldDeriveWindowFromTopicMessageStoreTimestamp()
throws Exception {
+ MessageExt original = new MessageExt();
+ original.setMsgId("msg-with-topic");
+ original.setStoreTimestamp(10_000_000L);
+ when(adminExt.viewMessage("orders",
"msg-with-topic")).thenReturn(original);
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenReturn(new QueryResult(0L, List.of()));
+
+ provider.getMessageTrace("instance-a", "msg-with-topic", "orders");
+
+ ArgumentCaptor<Long> beginCaptor = ArgumentCaptor.forClass(Long.class);
+ ArgumentCaptor<Long> endCaptor = ArgumentCaptor.forClass(Long.class);
+ verify(adminExt).queryMessage(eq("RMQ_SYS_TRACE_TOPIC"),
eq("msg-with-topic"), eq(64),
+ beginCaptor.capture(), endCaptor.capture());
+ assertThat(beginCaptor.getValue()).isEqualTo(10_000_000L - 5 *
60_000L);
+ assertThat(endCaptor.getValue()).isGreaterThanOrEqualTo(10_000_000L +
24 * 3600_000L);
+ verify(queryHistoryService).recordTraceQuery(eq("instance-a"),
eq("msg-with-topic"), eq(null), eq(0), eq(0));
+ }
+
+ @Test
+ void
getMessageTraceShouldUseFallbackOneHourWindowWhenMessageTimestampCannotBeResolved()
throws Exception {
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenReturn(new QueryResult(0L, List.of()));
+
+ provider.getMessageTrace("instance-a", "invalid-offset-id", "orders");
+
+ ArgumentCaptor<Long> beginCaptor = ArgumentCaptor.forClass(Long.class);
+ ArgumentCaptor<Long> endCaptor = ArgumentCaptor.forClass(Long.class);
+ verify(adminExt).queryMessage(eq("RMQ_SYS_TRACE_TOPIC"),
eq("invalid-offset-id"), eq(64),
+ beginCaptor.capture(), endCaptor.capture());
+ assertThat(endCaptor.getValue() -
beginCaptor.getValue()).isBetween(3_660_000L, 3_670_000L);
+ verify(queryHistoryService).recordTraceQuery(eq("instance-a"),
eq("invalid-offset-id"), eq(null), eq(0), eq(0));
+ }
+
+ @Test
+ void
offsetMessageIdFixtureShouldDecodeToBrokerAddressForTraceWindowTests() throws
Exception {
+ String offsetMsgId = MessageDecoder.createMessageId(
+ new InetSocketAddress("127.0.0.1", 10911), 12345L);
+
+ MessageId decoded = MessageDecoder.decodeMessageId(offsetMsgId);
+
+ assertThat(decoded.getAddress()).isInstanceOf(InetSocketAddress.class);
+ InetSocketAddress address = (InetSocketAddress) decoded.getAddress();
+ assertThat(address.getHostString()).isEqualTo("127.0.0.1");
+ assertThat(address.getPort()).isEqualTo(10911);
+ }
+
@Test
void getMessageTraceSurfacesAdminFailure() throws Exception {
when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
index 4c5d07ba..f233111c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
@@ -54,6 +54,7 @@ import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -111,6 +112,31 @@ class RocketMQMetadataProviderTest {
assertThat(groups.get(0).getConsumeType()).isEqualTo(ConsumeType.CLUSTERING);
}
+ @Test
+ void listConsumerGroupsShouldNotFetchLiveAdminInfoForEachGroup() {
+ RmqGroup first = new RmqGroup();
+ first.setName("group-a");
+ first.setClusterId("cluster-1");
+ first.setConsumeType("CLUSTERING");
+ first.setMaxRetry(16);
+ RmqGroup second = new RmqGroup();
+ second.setName("group-b");
+ second.setClusterId("cluster-1");
+ second.setConsumeType("BROADCASTING");
+ second.setMaxRetry(3);
+ when(groupMapper.selectList(any())).thenReturn(List.of(first, second));
+
+ RocketMQMetadataProvider provider = newProvider();
+
+ List<ConsumerGroupVO> groups =
provider.listConsumerGroups("cluster-1", null);
+
+
assertThat(groups).extracting(ConsumerGroupVO::getName).containsExactly("group-a",
"group-b");
+
assertThat(groups).extracting(ConsumerGroupVO::getOnlineInstances).containsExactly(0,
0);
+
assertThat(groups).extracting(ConsumerGroupVO::getTotalLag).containsExactly(0L,
0L);
+ verify(groupMapper).selectList(any());
+ verifyNoInteractions(runtimeAdminClientResolver);
+ }
+
@Test
void getTopicRoutesShouldUseSelectedInstanceRuntimeClient() {
List<BrokerRouteVO> routes =
List.of(BrokerRouteVO.builder().brokerName("broker-a").build());
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 76dc7abc..d14736d6 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
@@ -514,6 +514,19 @@ class SettingsServiceTest {
assertThat(result.getMessage()).contains("local or private address");
}
+ @Test
+ void testConnectionShouldRejectLoopbackIpv4Address() {
+ DataSourceTestDTO request = DataSourceTestDTO.builder()
+ .url("http://127.0.0.1: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()
