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 9e2bb75b4 fix: stop fabricating or losing state in four backend paths
(#5449)
9e2bb75b4 is described below
commit 9e2bb75b41c9aeb16cf56058b1eec6e070c3cd28
Author: 烤化の初雪 <[email protected]>
AuthorDate: Fri Oct 9 17:48:28 2026 +0800
fix: stop fabricating or losing state in four backend paths (#5449)
Four unrelated small backend defects, batched because each is a few lines
with its own test.
- `ApacheRocketMqClusterMetricsCollector` emitted no samples at all when a
broker's runtime stats
could not be read (no master address, null stats table, failed fetch),
and alert reconciliation
reads a missing sample as "the value cleared" - which resolved active
disk / JVM / send-queue
incidents on the collection failure itself. Those branches now emit
explicit unavailable
samples. (#5449)
- `LlmConfigService.saveConfig` rebuilt the persisted settings without
`clearApiKey`, so clearing the
API key in the UI did not survive the save. (#5436)
- `AliyunConverters.toTopicConsumerVO` left `metricsAvailable` at its
default, reporting lag and TPS
as real zeros when the Aliyun subscription API does not return them; it
is now explicitly
false. (#5423)
- `rmqctl/scripts/build-windows.ps1` defaulted `$Version` to `1.0.0` while
the CLI declares `3.0.0`,
so Windows builds stamped the wrong version. (#5432)
`AliyunConvertersTest` + `LlmConfigServiceTest` +
`ApacheRocketMqClusterMetricsCollectorTest`: 48
tests green, 0 checkstyle violations; `go test ./scripts/...` ok and `make
fmt` clean. The two new
collector cases were mutation-checked - removing either unavailable-sample
call reddens its own case.
Folded in #5423, #5432 and #5436.
---
rmqctl/cmd/version_metadata_test.go | 64 ++++++++++++++++++
rmqctl/scripts/build-windows.ps1 | 5 +-
.../ApacheRocketMqClusterMetricsCollector.java | 16 +++++
.../rocketmq/studio/ops/ai/LlmConfigService.java | 4 ++
.../studio/provider/alibaba/AliyunConverters.java | 5 ++
.../ApacheRocketMqClusterMetricsCollectorTest.java | 78 ++++++++++++++++++++++
.../studio/ops/ai/LlmConfigServiceTest.java | 24 +++++++
.../provider/alibaba/AliyunConvertersTest.java | 20 ++++++
8 files changed, 214 insertions(+), 2 deletions(-)
diff --git a/rmqctl/cmd/version_metadata_test.go
b/rmqctl/cmd/version_metadata_test.go
new file mode 100644
index 000000000..7abf63291
--- /dev/null
+++ b/rmqctl/cmd/version_metadata_test.go
@@ -0,0 +1,64 @@
+/*
+ * 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 cmd
+
+import (
+ "os"
+ "path/filepath"
+ "regexp"
+ "testing"
+)
+
+// The binary's reported version comes from three build entry points: `make
build` injects the
+// Makefile's VERSION, a plain `go build` falls back to CLIVersion, and the
Windows script injects
+// its own default. They had drifted (the script still defaulted to 1.0.0
after the CLI moved to
+// 3.0.0), so a Windows build reported a version no other build did. Pin them
together: a release
+// bump has to update all three or this fails.
+func TestVersionMetadataAgreesAcrossBuildEntryPoints(t *testing.T) {
+ makefile := readRepoFile(t, "Makefile")
+ script := readRepoFile(t, filepath.Join("scripts", "build-windows.ps1"))
+
+ makefileVersion := firstCapture(t, `(?m)^VERSION \?= (\S+)\s*$`,
makefile, "Makefile VERSION")
+ scriptVersion := firstCapture(t, `(?m)^\s*\[string\]\$Version =
"([^"]+)"`, script,
+ "build-windows.ps1 $Version default")
+
+ if CLIVersion != makefileVersion {
+ t.Errorf("cmd.CLIVersion = %q but Makefile VERSION ?= %q",
CLIVersion, makefileVersion)
+ }
+ if scriptVersion != CLIVersion {
+ t.Errorf("build-windows.ps1 default = %q but cmd.CLIVersion =
%q", scriptVersion, CLIVersion)
+ }
+}
+
+func readRepoFile(t *testing.T, relative string) string {
+ t.Helper()
+ path := filepath.Join("..", relative)
+ content, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatalf("reading %s: %v", path, err)
+ }
+ return string(content)
+}
+
+func firstCapture(t *testing.T, pattern, content, what string) string {
+ t.Helper()
+ matches := regexp.MustCompile(pattern).FindStringSubmatch(content)
+ if matches == nil {
+ t.Fatalf("could not read %s", what)
+ }
+ return matches[1]
+}
diff --git a/rmqctl/scripts/build-windows.ps1 b/rmqctl/scripts/build-windows.ps1
index 275cd0f95..af67aa4ed 100644
--- a/rmqctl/scripts/build-windows.ps1
+++ b/rmqctl/scripts/build-windows.ps1
@@ -29,7 +29,8 @@
right toolchain automatically.
.PARAMETER Version
- Value injected into cmd.CLIVersion (default: 1.0.0).
+ Value injected into cmd.CLIVersion (default: 3.0.0, the version the
Makefile and
+ cmd.CLIVersion already declare).
.PARAMETER Output
Output binary path (default: bin\rmqctl.exe).
@@ -42,7 +43,7 @@
.\scripts\build-windows.ps1 -Version 1.2.3 -RunTests
#>
param(
- [string]$Version = "1.0.0",
+ [string]$Version = "3.0.0",
[string]$Output = "bin\rmqctl.exe",
[switch]$RunTests
)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
index 9196ca196..3bedeb43e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
@@ -104,12 +104,14 @@ public class ApacheRocketMqClusterMetricsCollector
implements ClusterMetricsColl
: Map.of("brokerName", brokerName);
if (!StringUtils.hasText(address)) {
samples.add(unavailable(BROKER_AVAILABILITY, instance, clusterId,
labels, collectedAt));
+ addUnavailableRuntimeMetrics(instance, clusterId, labels,
collectedAt, samples);
return;
}
try {
KVTable runtime = admin.fetchBrokerRuntimeStats(address);
if (runtime == null || runtime.getTable() == null) {
samples.add(unavailable(BROKER_AVAILABILITY, instance,
clusterId, labels, collectedAt));
+ addUnavailableRuntimeMetrics(instance, clusterId, labels,
collectedAt, samples);
return;
}
samples.add(available(BROKER_AVAILABILITY, instance, clusterId,
labels, 1D, collectedAt));
@@ -125,9 +127,23 @@ public class ApacheRocketMqClusterMetricsCollector
implements ClusterMetricsColl
log.warn("Failed to collect runtime metrics for broker {} on
instance {}: {}", brokerName,
instance.getName(), error.getMessage());
samples.add(unavailable(BROKER_AVAILABILITY, instance, clusterId,
labels, collectedAt));
+ addUnavailableRuntimeMetrics(instance, clusterId, labels,
collectedAt, samples);
}
}
+ /**
+ * A broker whose runtime stats cannot be read at all must still report an
unavailable sample for
+ * each metric those stats carry: alert reconciliation treats a missing
sample as "the value
+ * cleared" and would resolve an active disk/JVM/send-queue incident on
the collection failure
+ * itself. The parse-miss branches already do this through {@code
metricOrUnavailable}.
+ */
+ private void addUnavailableRuntimeMetrics(InstanceVO instance, String
clusterId,
+ Map<String, String> labels, Instant collectedAt,
List<MetricSample> samples) {
+ samples.add(unavailable(BROKER_DISK_USAGE_RATIO, instance, clusterId,
labels, collectedAt));
+ samples.add(unavailable(BROKER_JVM_HEAP_USAGE_RATIO, instance,
clusterId, labels, collectedAt));
+ samples.add(unavailable(BROKER_SEND_QUEUE_USAGE_RATIO, instance,
clusterId, labels, collectedAt));
+ }
+
private static java.util.Optional<Double> parseDiskUsage(String raw) {
if (!StringUtils.hasText(raw)) {
return java.util.Optional.empty();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmConfigService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmConfigService.java
index 07e8ce423..138087787 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmConfigService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/LlmConfigService.java
@@ -132,6 +132,10 @@ public class LlmConfigService {
.llmProvider(normalized.getProvider())
.llmEngine(normalized.getEngine())
.apiKey(persistedApiKey)
+ // Carry the clear intent into the persisted settings: without
it a blank key plus
+ // clearApiKey=false means "keep the stored key" to
SettingsService's merge, so the
+ // secret the operator just removed would be written straight
back.
+ .clearApiKey(normalized.isClearApiKey())
.model(normalized.getModel())
.baseUrl(normalized.getApiBase())
.deploymentName(normalized.getDeploymentName())
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
index 4c8595eda..708bea0b9 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
@@ -164,6 +164,11 @@ final class AliyunConverters {
.group(data.getConsumerGroupId())
.consumeType(toConsumeType(data.getMessageModel()))
.messageModel(data.getMessageModel())
+ // ListTopicSubscriptions carries no lag or TPS field, so the
VO defaults (0/0)
+ // would render as measured "0 backlog / 0 TPS" in the console
and in the AI tool
+ // output. Flag them unavailable, the same way the Apache
provider does when
+ // consume stats cannot be read.
+ .metricsAvailable(false)
.build();
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollectorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollectorTest.java
index 2ffaf2a4c..79f3cfedb 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollectorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollectorTest.java
@@ -123,6 +123,74 @@ class ApacheRocketMqClusterMetricsCollectorTest {
.isEmpty();
}
+ @Test
+ void
emitsUnavailableSamplesForEveryRuntimeMetricWhenTheStatsCallFailsTest() throws
Exception {
+ RuntimeAdminClientResolver resolver =
mock(RuntimeAdminClientResolver.class);
+ MQAdminExt admin = mock(MQAdminExt.class);
+ InstanceVO instance = apacheInstance();
+ ClusterInfo topology = new ClusterInfo();
+ topology.setBrokerAddrTable(Map.of("broker-a", new
BrokerData("cluster-a", "broker-a",
+ new HashMap<>(Map.of(0L, "broker-a:10911")))));
+ when(admin.examineBrokerClusterInfo()).thenReturn(topology);
+ when(admin.fetchBrokerRuntimeStats("broker-a:10911"))
+ .thenThrow(new IllegalStateException("broker restarting"));
+ when(resolver.execute(eq(instance),
any(MqAdminExtFactory.AdminAction.class)))
+ .thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1)
+ .apply(admin));
+
+ List<MetricSample> samples = new
ApacheRocketMqClusterMetricsCollector(resolver).collect(instance);
+
+ // Without a sample of its own, an active disk/JVM/send-queue alert is
reconciled as cleared
+ // by the very collection that failed to read the value.
+ assertUnavailableBrokerMetric(samples, "broker.disk.usage_ratio");
+ assertUnavailableBrokerMetric(samples, "broker.jvm.heap.usage_ratio");
+ assertUnavailableBrokerMetric(samples,
"broker.send_queue.usage_ratio");
+ assertThat(samples).filteredOn(sample ->
sample.metricKey().equals("broker.availability"))
+ .singleElement().satisfies(sample ->
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.UNAVAILABLE));
+ }
+
+ @Test
+ void emitsUnavailableRuntimeMetricsWhenTheBrokerHasNoMasterAddressTest()
throws Exception {
+ RuntimeAdminClientResolver resolver =
mock(RuntimeAdminClientResolver.class);
+ MQAdminExt admin = mock(MQAdminExt.class);
+ InstanceVO instance = apacheInstance();
+ ClusterInfo topology = new ClusterInfo();
+ // A BrokerData with no 0L entry has no master address to read runtime
stats from.
+ topology.setBrokerAddrTable(Map.of("broker-a", new
BrokerData("cluster-a", "broker-a", new HashMap<>())));
+ when(admin.examineBrokerClusterInfo()).thenReturn(topology);
+ when(resolver.execute(eq(instance),
any(MqAdminExtFactory.AdminAction.class)))
+ .thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1)
+ .apply(admin));
+
+ List<MetricSample> samples = new
ApacheRocketMqClusterMetricsCollector(resolver).collect(instance);
+
+ assertUnavailableBrokerMetricWithoutAddress(samples,
"broker.disk.usage_ratio");
+ assertUnavailableBrokerMetricWithoutAddress(samples,
"broker.jvm.heap.usage_ratio");
+ assertUnavailableBrokerMetricWithoutAddress(samples,
"broker.send_queue.usage_ratio");
+ }
+
+ @Test
+ void emitsUnavailableRuntimeMetricsWhenTheStatsTableIsNullTest() throws
Exception {
+ RuntimeAdminClientResolver resolver =
mock(RuntimeAdminClientResolver.class);
+ MQAdminExt admin = mock(MQAdminExt.class);
+ InstanceVO instance = apacheInstance();
+ ClusterInfo topology = new ClusterInfo();
+ topology.setBrokerAddrTable(Map.of("broker-a", new
BrokerData("cluster-a", "broker-a",
+ new HashMap<>(Map.of(0L, "broker-a:10911")))));
+ when(admin.examineBrokerClusterInfo()).thenReturn(topology);
+ when(admin.fetchBrokerRuntimeStats("broker-a:10911")).thenReturn(null);
+ when(resolver.execute(eq(instance),
any(MqAdminExtFactory.AdminAction.class)))
+ .thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1)
+ .apply(admin));
+
+ List<MetricSample> samples = new
ApacheRocketMqClusterMetricsCollector(resolver).collect(instance);
+
+ assertUnavailableBrokerMetric(samples, "broker.disk.usage_ratio");
+ assertUnavailableBrokerMetric(samples, "broker.jvm.heap.usage_ratio");
+ assertUnavailableBrokerMetric(samples,
"broker.send_queue.usage_ratio");
+ }
+
private static void assertUnavailableBrokerMetric(List<MetricSample>
samples, String metricKey) {
assertThat(samples).filteredOn(sample ->
sample.metricKey().equals(metricKey))
.singleElement().satisfies(sample -> {
@@ -133,6 +201,16 @@ class ApacheRocketMqClusterMetricsCollectorTest {
});
}
+ private static void
assertUnavailableBrokerMetricWithoutAddress(List<MetricSample> samples, String
metricKey) {
+ assertThat(samples).filteredOn(sample ->
sample.metricKey().equals(metricKey))
+ .singleElement().satisfies(sample -> {
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.UNAVAILABLE);
+ assertThat(sample.value()).isNull();
+ assertThat(sample.labels()).containsEntry("brokerName",
"broker-a")
+ .doesNotContainKey("brokerAddr");
+ });
+ }
+
private static InstanceVO apacheInstance() {
return
InstanceVO.builder().name("local").endpoint("localhost:9876").vendor(InstanceVendor.APACHE).build();
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/LlmConfigServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/LlmConfigServiceTest.java
index 04064a476..4c5f7e5ef 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/LlmConfigServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/LlmConfigServiceTest.java
@@ -308,11 +308,35 @@ class LlmConfigServiceTest {
ArgumentCaptor<GeneralSettingsVO> captor =
ArgumentCaptor.forClass(GeneralSettingsVO.class);
verify(settingsService).saveGeneralSettings(captor.capture());
assertThat(captor.getValue().getApiKey()).isBlank();
+ // A blank key alone is not the clear signal: SettingsService's merge
reads a blank key
+ // with clearApiKey=false as "keep the stored key" and persists it
again, so the flag has
+ // to survive the rebuild of the settings VO.
+ assertThat(captor.getValue().isClearApiKey()).isTrue();
when(settingsService.getGeneralSettings()).thenReturn(captor.getValue());
assertThat(llmConfigService.getConfig().getApiKey()).isBlank();
assertThat(llmConfigService.getConfig().isApiKeyConfigured()).isFalse();
}
+ @Test
+ void saveConfigShouldNotAskToClearTheStoredApiKeyOnAnOrdinarySaveTest() {
+ llmConfigService.saveConfig(LlmConfigVO.builder()
+ .provider("deepseek")
+ .apiKey("sk-new")
+ .apiBase("https://api.deepseek.com/v1";)
+ .model("deepseek-chat")
+ .maxTokens(8192)
+ .temperature(0.2)
+ .enabled(true)
+ .build());
+
+ ArgumentCaptor<GeneralSettingsVO> captor =
ArgumentCaptor.forClass(GeneralSettingsVO.class);
+ verify(settingsService).saveGeneralSettings(captor.capture());
+ // The merge must keep filling in a key the form left blank, so an
ordinary save must not
+ // carry a clear flag.
+ assertThat(captor.getValue().isClearApiKey()).isFalse();
+ assertThat(captor.getValue().getApiKey()).isEqualTo("sk-new");
+ }
+
@Test
void saveConfigShouldNormalizeChatCompletionsEndpointToApiBase() {
llmConfigService.saveConfig(LlmConfigVO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConvertersTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConvertersTest.java
index 148624940..4629bf38e 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConvertersTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConvertersTest.java
@@ -18,8 +18,10 @@ package org.apache.rocketmq.studio.provider.alibaba;
import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupSubscriptionsResponseBody;
import
com.aliyun.sdk.service.rocketmq20220801.models.ListInstancesResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListTopicSubscriptionsResponseBody;
import com.aliyun.sdk.service.rocketmq20220801.models.ListTopicsResponseBody;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
+import org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
import org.apache.rocketmq.studio.instance.topic.TopicVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.junit.jupiter.api.Test;
@@ -61,6 +63,24 @@ class AliyunConvertersTest {
assertThat(vo.getType()).isEqualTo(TopicType.LITE);
}
+ @Test
+ void toTopicConsumerVoShouldMarkTheUngettableLagAndTpsUnavailableTest() {
+ // ListTopicSubscriptionsResponseBody.Data carries only consistency,
consumerGroupId,
+ // filterExpression, filterExpressionType, messageModel,
subscriptionStatus and topicName -
+ // no lag or TPS. Leaving the VO defaults would present 0 backlog / 0
TPS as measured
+ // values; the Apache provider flags the same gap with
metricsAvailable(false), which is
+ // the only value the console renders as unavailable.
+ ListTopicSubscriptionsResponseBody.Data data =
ListTopicSubscriptionsResponseBody.Data.builder()
+ .consumerGroupId("cg-orders")
+ .messageModel("CLUSTERING")
+ .build();
+
+ TopicConsumerVO consumer = AliyunConverters.toTopicConsumerVO(data);
+
+ assertThat(consumer.getGroup()).isEqualTo("cg-orders");
+ assertThat(consumer.isMetricsAvailable()).isFalse();
+ }
+
@Test
void toInstanceOptionShouldClampCountsOutsideTheIntegerRange() {
ListInstancesResponseBody.List data =
ListInstancesResponseBody.List.builder()
