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 3693dab57 fix(alert): renew collection lease during long passes (#2642)
3693dab57 is described below
commit 3693dab57c4a25de2a258c778763f152a07beeeb
Author: xdz997 <[email protected]>
AuthorDate: Fri Sep 4 15:48:47 2026 +0800
fix(alert): renew collection lease during long passes (#2642)
* fix: renew native alert collection lease during long passes
* docs: clarify native alert lease heartbeat safeguards
---
docs/studio-native-alerting-design.md | 4 +
.../cluster/metrics/AlertCollectionLease.java | 8 +
.../studio/cluster/metrics/AlertingProperties.java | 2 +
.../studio/cluster/metrics/CollectorScheduler.java | 162 +++++++++++++++++++--
.../metrics/MybatisPlusAlertCollectionLease.java | 9 ++
.../mapper/RmqAlertCollectionLeaseMapper.java | 5 +
server/src/main/resources/application.yml | 2 +
.../cluster/metrics/CollectorSchedulerTest.java | 68 +++++++++
.../MybatisPlusAlertCollectionLeaseTest.java | 56 +++++++
9 files changed, 301 insertions(+), 15 deletions(-)
diff --git a/docs/studio-native-alerting-design.md
b/docs/studio-native-alerting-design.md
index 06c0e69a6..344714231 100644
--- a/docs/studio-native-alerting-design.md
+++ b/docs/studio-native-alerting-design.md
@@ -239,6 +239,10 @@ Snapshots have short retention, initially 24 hours. Studio
is not a replacement
`CollectorScheduler` creates per-instance jobs at a default 30-second
interval. Each job has a bounded timeout and the scheduler uses bounded
concurrency, so one slow remote instance does not indefinitely delay every
other instance. Collection failure records an unavailable sample and health
diagnostic; it does not block other instances.
+The database lease is held for the entire collection pass, including passes in
which every collector returns an empty sample list. The active holder renews it
periodically (by default every 15 seconds, clamped to no more than one third of
the configured lease duration). Renewal only succeeds for the current holder
while its previous lease is still unexpired. If renewal fails, the scheduler
cancels outstanding collection jobs and refuses to persist later samples, so an
expired replica can [...]
+
+The lease duration and heartbeat interval can be configured with
`STUDIO_ALERTING_COLLECTION_LEASE_DURATION` and
`STUDIO_ALERTING_COLLECTION_LEASE_RENEWAL_INTERVAL`. The interval should be
comfortably shorter than the duration to tolerate transient database latency.
+
For multi-replica Studio deployment, the scheduler uses a database lease:
```text
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertCollectionLease.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertCollectionLease.java
index 2669a92b3..1d0a23d31 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertCollectionLease.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertCollectionLease.java
@@ -17,5 +17,13 @@
package org.apache.rocketmq.studio.cluster.metrics;
public interface AlertCollectionLease {
+ /** Claims the lease, or refreshes it when this replica already owns it. */
boolean tryAcquire();
+
+ /**
+ * Extends the current holder's lease without allowing an expired holder
to reclaim it.
+ *
+ * @return {@code true} when the current holder still owns the lease
+ */
+ boolean renew();
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertingProperties.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertingProperties.java
index a7773976c..c81854d2e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertingProperties.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertingProperties.java
@@ -32,6 +32,8 @@ public class AlertingProperties {
* native samples and emitting duplicate alert events.
*/
private String collectionLeaseDuration = "PT1M";
+ /** Maximum time between lease heartbeats; the scheduler clamps this below
the lease duration. */
+ private String collectionLeaseRenewalInterval = "PT15S";
/** Retain short-lived diagnostic samples without allowing the snapshot
table to grow indefinitely. */
private String snapshotRetention = "PT24H";
/** Retain terminal notification deliveries before deleting them from the
outbox. */
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
index 22fc26270..44301997b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
@@ -34,14 +34,18 @@ import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletionService;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.time.Duration;
import java.time.Instant;
import java.util.Set;
@@ -59,6 +63,7 @@ public class CollectorScheduler {
private final NativeAlertProcessor alertProcessor;
private final AlertCollectionLease collectionLease;
private final ExecutorService collectionExecutor;
+ private final ScheduledExecutorService leaseRenewalExecutor;
@Autowired
public CollectorScheduler(AlertingProperties properties,
InstanceRepository instanceRepository,
@@ -66,13 +71,22 @@ public class CollectorScheduler {
MetricSnapshotRepository snapshotRepository, NativeAlertProcessor
alertProcessor,
AlertCollectionLease collectionLease) {
this(properties, instanceRepository, clusterCollectors,
businessCollectors, snapshotRepository, alertProcessor,
- collectionLease, newCollectionExecutor(properties));
+ collectionLease, newCollectionExecutor(properties),
newLeaseRenewalExecutor());
}
CollectorScheduler(AlertingProperties properties, InstanceRepository
instanceRepository,
List<ClusterMetricsCollector> clusterCollectors,
List<BusinessMetricsCollector> businessCollectors,
MetricSnapshotRepository snapshotRepository, NativeAlertProcessor
alertProcessor,
AlertCollectionLease collectionLease, ExecutorService
collectionExecutor) {
+ this(properties, instanceRepository, clusterCollectors,
businessCollectors, snapshotRepository, alertProcessor,
+ collectionLease, collectionExecutor,
newLeaseRenewalExecutor());
+ }
+
+ CollectorScheduler(AlertingProperties properties, InstanceRepository
instanceRepository,
+ List<ClusterMetricsCollector> clusterCollectors,
List<BusinessMetricsCollector> businessCollectors,
+ MetricSnapshotRepository snapshotRepository, NativeAlertProcessor
alertProcessor,
+ AlertCollectionLease collectionLease, ExecutorService
collectionExecutor,
+ ScheduledExecutorService leaseRenewalExecutor) {
this.properties = properties;
this.instanceRepository = instanceRepository;
this.clusterCollectors = clusterCollectors;
@@ -81,6 +95,7 @@ public class CollectorScheduler {
this.alertProcessor = alertProcessor;
this.collectionLease = collectionLease;
this.collectionExecutor = collectionExecutor;
+ this.leaseRenewalExecutor = leaseRenewalExecutor;
}
@Scheduled(fixedDelayString =
"${studio.alerting.collection-interval:PT30S}")
@@ -89,8 +104,20 @@ public class CollectorScheduler {
log.debug("Skipping native alert collection because another Studio
replica holds the lease");
return;
}
+ List<Future<?>> leaseJobs = new CopyOnWriteArrayList<>();
+ LeaseHeartbeat heartbeat = new LeaseHeartbeat(collectionLease,
leaseRenewalExecutor,
+ leaseRenewalInterval(), leaseJobs);
+ heartbeat.start();
+ try {
+ collectWithPassBudget(heartbeat, leaseJobs);
+ } finally {
+ heartbeat.close();
+ }
+ }
+
+ private void collectWithPassBudget(LeaseHeartbeat heartbeat,
List<Future<?>> leaseJobs) {
List<InstanceVO> instances = instanceRepository.findAll();
- if (instances.isEmpty()) {
+ if (instances.isEmpty() || !heartbeat.isHeld()) {
return;
}
Duration timeout =
parsePositiveDuration(properties.getCollectionTimeout(),
Duration.ofSeconds(15));
@@ -101,11 +128,13 @@ public class CollectorScheduler {
ArrayDeque<InstanceVO> pending = new ArrayDeque<>(instances);
List<CollectionJob> running = new ArrayList<>();
try {
- submitAvailable(completionService, pending, running, parallelism,
timeout, passDeadline);
- while (!running.isEmpty()) {
+ submitAvailable(completionService, pending, running, parallelism,
timeout, passDeadline, heartbeat,
+ leaseJobs);
+ while (!running.isEmpty() && heartbeat.isHeld()) {
completeFinishedJobs(running);
cancelExpiredJobs(running, timeout, passDeadline, passTimeout);
- submitAvailable(completionService, pending, running,
parallelism, timeout, passDeadline);
+ submitAvailable(completionService, pending, running,
parallelism, timeout, passDeadline, heartbeat,
+ leaseJobs);
if (running.isEmpty()) {
break;
}
@@ -124,6 +153,10 @@ public class CollectorScheduler {
log.warn("Native metric collection executor rejected new work;
aborting pass with {} unstarted instance(s)",
pending.size());
}
+ if (!heartbeat.isHeld()) {
+ cancelRunningJobs(running);
+ return;
+ }
if (!pending.isEmpty()) {
log.warn("Native metric collection pass exceeded {}; {}
instance(s) were not started", passTimeout,
pending.size());
@@ -131,8 +164,9 @@ public class CollectorScheduler {
}
private void submitAvailable(CompletionService<InstanceVO>
completionService, ArrayDeque<InstanceVO> pending,
- List<CollectionJob> running, int parallelism, Duration timeout,
Instant passDeadline) {
- while (!pending.isEmpty() && running.size() < parallelism) {
+ List<CollectionJob> running, int parallelism, Duration timeout,
Instant passDeadline,
+ LeaseHeartbeat heartbeat, List<Future<?>> leaseJobs) {
+ while (!pending.isEmpty() && running.size() < parallelism &&
heartbeat.isHeld()) {
Instant now = Instant.now();
if (!passDeadline.isAfter(now)) {
return;
@@ -140,10 +174,13 @@ public class CollectorScheduler {
InstanceVO instance = pending.removeFirst();
try {
Future<InstanceVO> future = completionService.submit(() -> {
- collectClusterMetrics(instance);
- collectBusinessMetrics(instance);
+ if (heartbeat.isHeld()) {
+ collectClusterMetrics(instance, heartbeat);
+ collectBusinessMetrics(instance, heartbeat);
+ }
return instance;
});
+ leaseJobs.add(future);
running.add(new CollectionJob(instance, future,
now.plus(timeout)));
} catch (RejectedExecutionException error) {
pending.addFirst(instance);
@@ -232,12 +269,15 @@ public class CollectorScheduler {
snapshotRepository.deleteBefore(Instant.now().minus(retention));
}
- private void collectClusterMetrics(InstanceVO instance) {
+ private void collectClusterMetrics(InstanceVO instance, LeaseHeartbeat
heartbeat) {
for (ClusterMetricsCollector collector : clusterCollectors) {
+ if (!heartbeat.isHeld()) {
+ return;
+ }
try {
if (collector.supports(instance)) {
List<MetricSample> samples = collector.collect(instance);
- persist(AlertDomain.CLUSTER, instance,
collector.metricKeys(), samples);
+ persist(AlertDomain.CLUSTER, instance,
collector.metricKeys(), samples, heartbeat);
}
} catch (RuntimeException error) {
log.warn("Native metric collector failed for instance {}: {}",
instance.getName(), error.getMessage());
@@ -245,12 +285,15 @@ public class CollectorScheduler {
}
}
- private void collectBusinessMetrics(InstanceVO instance) {
+ private void collectBusinessMetrics(InstanceVO instance, LeaseHeartbeat
heartbeat) {
for (BusinessMetricsCollector collector : businessCollectors) {
+ if (!heartbeat.isHeld()) {
+ return;
+ }
try {
if (collector.supports(instance)) {
List<MetricSample> samples = collector.collect(instance);
- persist(AlertDomain.BUSINESS, instance,
collector.metricKeys(), samples);
+ persist(AlertDomain.BUSINESS, instance,
collector.metricKeys(), samples, heartbeat);
}
} catch (RuntimeException error) {
log.warn("Native metric collector failed for instance {}: {}",
instance.getName(), error.getMessage());
@@ -259,7 +302,7 @@ public class CollectorScheduler {
}
private void persist(AlertDomain domain, InstanceVO instance, Set<String>
declaredMetricKeys,
- List<MetricSample> samples) {
+ List<MetricSample> samples, LeaseHeartbeat heartbeat) {
List<MetricSample> collected = samples == null ? List.of() : samples;
Set<String> scopeMetricKeys = metricKeys(declaredMetricKeys,
collected);
if (scopeMetricKeys.isEmpty()) {
@@ -267,7 +310,9 @@ public class CollectorScheduler {
}
MetricCollectionScope actualScope = new MetricCollectionScope(domain,
instance.getName(), scopeMetricKeys);
// Refresh immediately before persisting so a slow remote collection
cannot write after lease loss.
- if (!collectionLease.tryAcquire()) {
+ // This second ownership check is defense-in-depth against a heartbeat
tick racing this write.
+ if (!heartbeat.isHeld() || !collectionLease.tryAcquire() ||
!heartbeat.isHeld()) {
+ heartbeat.markLost();
log.debug("Discarding native metric samples because the collection
lease was lost");
return;
}
@@ -287,6 +332,7 @@ public class CollectorScheduler {
@PreDestroy
void stopCollectionExecutor() {
collectionExecutor.shutdownNow();
+ leaseRenewalExecutor.shutdownNow();
}
private static ExecutorService newCollectionExecutor(AlertingProperties
properties) {
@@ -300,6 +346,92 @@ public class CollectorScheduler {
new ArrayBlockingQueue<>(parallelism), threadFactory, new
ThreadPoolExecutor.AbortPolicy());
}
+ private static ScheduledExecutorService newLeaseRenewalExecutor() {
+ ThreadFactory threadFactory = task -> {
+ Thread thread = new Thread(task,
"studio-native-metric-lease-renewer");
+ thread.setDaemon(true);
+ return thread;
+ };
+ return
java.util.concurrent.Executors.newSingleThreadScheduledExecutor(threadFactory);
+ }
+
+ private Duration leaseRenewalInterval() {
+ Duration leaseDuration =
parsePositiveDuration(properties.getCollectionLeaseDuration(),
Duration.ofMinutes(1));
+ Duration safeMaximum = leaseDuration.dividedBy(3);
+ if (!safeMaximum.isPositive()) {
+ safeMaximum = Duration.ofMillis(1);
+ }
+ Duration configured =
parsePositiveDuration(properties.getCollectionLeaseRenewalInterval(),
safeMaximum);
+ return configured.compareTo(safeMaximum) > 0 ? safeMaximum :
configured;
+ }
+
+ /**
+ * Renews the collection lease while broker calls are in flight and
cancels outstanding work
+ * as soon as ownership is lost, so an expired pass cannot publish late
samples.
+ */
+ private static final class LeaseHeartbeat implements AutoCloseable {
+ private final AlertCollectionLease lease;
+ private final ScheduledExecutorService executor;
+ private final Duration interval;
+ private final List<Future<?>> jobs;
+ private final AtomicBoolean held = new AtomicBoolean(true);
+ private volatile ScheduledFuture<?> scheduledTask;
+
+ private LeaseHeartbeat(AlertCollectionLease lease,
ScheduledExecutorService executor, Duration interval,
+ List<Future<?>> jobs) {
+ this.lease = lease;
+ this.executor = executor;
+ this.interval = interval;
+ this.jobs = jobs;
+ }
+
+ private void start() {
+ long intervalMillis = Math.max(1L, interval.toMillis());
+ try {
+ scheduledTask = executor.scheduleAtFixedRate(this::renewLease,
intervalMillis, intervalMillis,
+ TimeUnit.MILLISECONDS);
+ } catch (RuntimeException error) {
+ markLost();
+ log.warn("Unable to start native alert collection lease
heartbeat", error);
+ }
+ }
+
+ private void renewLease() {
+ if (!held.get()) {
+ return;
+ }
+ try {
+ if (!lease.renew()) {
+ markLost();
+ log.warn("Native alert collection lease renewal failed;
cancelling the active collection pass");
+ }
+ } catch (RuntimeException error) {
+ markLost();
+ log.warn("Native alert collection lease renewal raised an
exception; cancelling the active pass",
+ error);
+ }
+ }
+
+ private boolean isHeld() {
+ return held.get();
+ }
+
+ private void markLost() {
+ if (held.compareAndSet(true, false)) {
+ jobs.forEach(job -> job.cancel(true));
+ }
+ }
+
+ @Override
+ public void close() {
+ held.set(false);
+ ScheduledFuture<?> task = scheduledTask;
+ if (task != null) {
+ task.cancel(false);
+ }
+ }
+ }
+
private record CollectionJob(InstanceVO instance, Future<InstanceVO>
future, Instant deadline) {
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusAlertCollectionLease.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusAlertCollectionLease.java
index bca58e6b8..6bd12afa4 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusAlertCollectionLease.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusAlertCollectionLease.java
@@ -61,6 +61,15 @@ public class MybatisPlusAlertCollectionLease implements
AlertCollectionLease {
}
}
+ @Override
+ public boolean renew() {
+ Instant now = Instant.now();
+ Duration duration =
parseDuration(properties.getCollectionLeaseDuration());
+ LocalDateTime renewedAt = LocalDateTime.ofInstant(now, ZoneOffset.UTC);
+ LocalDateTime expiresAt = LocalDateTime.ofInstant(now.plus(duration),
ZoneOffset.UTC);
+ return mapper.renew(LEASE_NAME, holderId, renewedAt, expiresAt) > 0;
+ }
+
private static Duration parseDuration(String configured) {
try {
Duration duration = Duration.parse(configured);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqAlertCollectionLeaseMapper.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqAlertCollectionLeaseMapper.java
index 023074702..a932c9bc5 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqAlertCollectionLeaseMapper.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqAlertCollectionLeaseMapper.java
@@ -29,4 +29,9 @@ public interface RmqAlertCollectionLeaseMapper extends
BaseMapper<RmqAlertCollec
+ "AND (expires_at <= #{now} OR holder_id = #{holderId})")
int acquire(@Param("leaseName") String leaseName, @Param("holderId")
String holderId,
@Param("now") LocalDateTime now, @Param("expiresAt") LocalDateTime
expiresAt);
+
+ @Update("UPDATE rmq_alert_collection_lease SET expires_at = #{expiresAt},
gmt_modified = #{now} "
+ + "WHERE lease_name = #{leaseName} AND holder_id = #{holderId} AND
expires_at > #{now}")
+ int renew(@Param("leaseName") String leaseName, @Param("holderId") String
holderId,
+ @Param("now") LocalDateTime now, @Param("expiresAt") LocalDateTime
expiresAt);
}
diff --git a/server/src/main/resources/application.yml
b/server/src/main/resources/application.yml
index 9a586acbb..6efbf2bb5 100644
--- a/server/src/main/resources/application.yml
+++ b/server/src/main/resources/application.yml
@@ -85,6 +85,8 @@ studio:
collection-interval: ${STUDIO_ALERTING_COLLECTION_INTERVAL:PT30S}
collection-parallelism: ${STUDIO_ALERTING_COLLECTION_PARALLELISM:4}
collection-timeout: ${STUDIO_ALERTING_COLLECTION_TIMEOUT:PT15S}
+ collection-lease-duration:
${STUDIO_ALERTING_COLLECTION_LEASE_DURATION:PT1M}
+ collection-lease-renewal-interval:
${STUDIO_ALERTING_COLLECTION_LEASE_RENEWAL_INTERVAL:PT15S}
snapshot-retention: ${STUDIO_ALERTING_SNAPSHOT_RETENTION:PT24H}
snapshot-cleanup-interval:
${STUDIO_ALERTING_SNAPSHOT_CLEANUP_INTERVAL:PT1H}
notification-dispatch-interval:
${STUDIO_ALERTING_NOTIFICATION_DISPATCH_INTERVAL:PT10S}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
index ecf956412..33fa2ed72 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
@@ -30,6 +30,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
@@ -38,6 +39,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -331,6 +333,72 @@ class CollectorSchedulerTest {
verify(processor).processSuccessfulCollection(any(MetricCollectionScope.class),
org.mockito.ArgumentMatchers.eq(List.of()));
}
+ @Test
+ void renewsLeaseWhileAnEmptyCollectionPassIsStillRunningTest() throws
Exception {
+ AlertingProperties properties = new AlertingProperties();
+ properties.setCollectionLeaseDuration("PT0.2S");
+ properties.setCollectionLeaseRenewalInterval("PT1S");
+ properties.setCollectionTimeout("PT1S");
+ InstanceVO instance = InstanceVO.builder().name("slow-empty").build();
+ InstanceRepository instances = mock(InstanceRepository.class);
+ when(instances.findAll()).thenReturn(List.of(instance));
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ when(collector.supports(instance)).thenReturn(true);
+ when(collector.collect(instance)).thenAnswer(invocation -> {
+ Thread.sleep(350);
+ return List.of();
+ });
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ when(lease.tryAcquire()).thenReturn(true);
+ when(lease.renew()).thenReturn(true);
+ ExecutorService collectionExecutor = Executors.newFixedThreadPool(1);
+ ScheduledExecutorService renewalExecutor =
Executors.newSingleThreadScheduledExecutor();
+ CollectorScheduler scheduler = new CollectorScheduler(properties,
instances, List.of(collector), List.of(),
+ mock(MetricSnapshotRepository.class),
mock(NativeAlertProcessor.class), lease, collectionExecutor,
+ renewalExecutor);
+
+ try {
+ scheduler.collect();
+ verify(lease, atLeast(3)).renew();
+ } finally {
+ scheduler.stopCollectionExecutor();
+ }
+ }
+
+ @Test
+ void stopsAnActivePassWhenLeaseRenewalFailsTest() throws Exception {
+ AlertingProperties properties = new AlertingProperties();
+ properties.setCollectionLeaseDuration("PT0.2S");
+ properties.setCollectionLeaseRenewalInterval("PT1S");
+ properties.setCollectionTimeout("PT1S");
+ InstanceVO instance = InstanceVO.builder().name("lease-lost").build();
+ InstanceRepository instances = mock(InstanceRepository.class);
+ when(instances.findAll()).thenReturn(List.of(instance));
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ when(collector.supports(instance)).thenReturn(true);
+ when(collector.collect(instance)).thenAnswer(invocation -> {
+ Thread.sleep(350);
+ return List.of(sampleFor(instance));
+ });
+ MetricSnapshotRepository snapshots =
mock(MetricSnapshotRepository.class);
+ NativeAlertProcessor processor = mock(NativeAlertProcessor.class);
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ when(lease.tryAcquire()).thenReturn(true);
+ when(lease.renew()).thenReturn(false);
+ ExecutorService collectionExecutor = Executors.newFixedThreadPool(1);
+ ScheduledExecutorService renewalExecutor =
Executors.newSingleThreadScheduledExecutor();
+ CollectorScheduler scheduler = new CollectorScheduler(properties,
instances, List.of(collector), List.of(),
+ snapshots, processor, lease, collectionExecutor,
renewalExecutor);
+
+ try {
+ scheduler.collect();
+ verify(snapshots, never()).saveAll(any());
+ verify(processor, never()).processSuccessfulCollection(any(),
any());
+ } finally {
+ scheduler.stopCollectionExecutor();
+ }
+ }
+
private static MetricSample sampleFor(InstanceVO instance) {
return new MetricSample("nameserver.availability",
AlertDomain.CLUSTER, instance.getName(), null,
null, 1D, MetricAvailability.AVAILABLE, Instant.now());
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusAlertCollectionLeaseTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusAlertCollectionLeaseTest.java
new file mode 100644
index 000000000..ea17e6a84
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusAlertCollectionLeaseTest.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.cluster.metrics;
+
+import
org.apache.rocketmq.studio.persistence.mapper.RmqAlertCollectionLeaseMapper;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class MybatisPlusAlertCollectionLeaseTest {
+
+ @Test
+ void renewsAnUnexpiredLeaseHeldByThisReplicaTest() {
+ AlertingProperties properties = new AlertingProperties();
+ properties.setCollectionLeaseDuration("PT30S");
+ RmqAlertCollectionLeaseMapper mapper =
mock(RmqAlertCollectionLeaseMapper.class);
+ when(mapper.renew(eq("native-alert-collection"), anyString(), any(),
any())).thenReturn(1);
+
+ MybatisPlusAlertCollectionLease lease = new
MybatisPlusAlertCollectionLease(properties, mapper);
+
+ assertThat(lease.renew()).isTrue();
+ verify(mapper).renew(eq("native-alert-collection"), anyString(),
any(), any());
+ }
+
+ @Test
+ void reportsLeaseLossWhenTheDatabaseNoLongerMatchesThisHolderTest() {
+ AlertingProperties properties = new AlertingProperties();
+ RmqAlertCollectionLeaseMapper mapper =
mock(RmqAlertCollectionLeaseMapper.class);
+ when(mapper.renew(eq("native-alert-collection"), anyString(), any(),
any())).thenReturn(0);
+
+ MybatisPlusAlertCollectionLease lease = new
MybatisPlusAlertCollectionLease(properties, mapper);
+
+ assertThat(lease.renew()).isFalse();
+ verify(mapper).renew(eq("native-alert-collection"), anyString(),
any(), any());
+ }
+}
