RockteMQ-AI commented on code in PR #2642:
URL:
https://github.com/apache/rocketmq-dashboard/pull/2642#discussion_r3875654579
##########
server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java:
##########
@@ -120,30 +170,39 @@ public void cleanUpSnapshots() {
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 {
- persist(collector.supports(instance) ?
collector.collect(instance) : List.of());
+ List<MetricSample> samples = collector.supports(instance) ?
collector.collect(instance) : List.of();
+ persist(samples, heartbeat);
} catch (RuntimeException error) {
log.warn("Native metric collector failed for instance {}: {}",
instance.getName(), error.getMessage());
}
}
}
- private void collectBusinessMetrics(InstanceVO instance) {
+ private void collectBusinessMetrics(InstanceVO instance, LeaseHeartbeat
heartbeat) {
for (BusinessMetricsCollector collector : businessCollectors) {
+ if (!heartbeat.isHeld()) {
+ return;
+ }
try {
- persist(collector.supports(instance) ?
collector.collect(instance) : List.of());
+ List<MetricSample> samples = collector.supports(instance) ?
collector.collect(instance) : List.of();
+ persist(samples, heartbeat);
} catch (RuntimeException error) {
log.warn("Native metric collector failed for instance {}: {}",
instance.getName(), error.getMessage());
}
Review Comment:
**[Info]** The `tryAcquire()` re-check inside `persist()` is a good
defensive measure, but now that the `LeaseHeartbeat` already handles periodic
renewal and `isHeld()` is checked before calling `persist()`, this double-check
is somewhat redundant in the happy path. Consider adding a brief comment
explaining that this is intentional defense-in-depth (in case the heartbeat
interval happens to align poorly with a slow persist call).
##########
server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java:
##########
@@ -168,13 +228,101 @@ private static ExecutorService
newCollectionExecutor(AlertingProperties properti
new ArrayBlockingQueue<>(parallelism), threadFactory, new
ThreadPoolExecutor.AbortPolicy());
}
- private static Duration parsePositiveDuration(String value, Duration
fallback) {
+ 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),
+ "collection lease duration");
+ Duration safeMaximum = leaseDuration.dividedBy(3);
+ if (safeMaximum.isZero() || safeMaximum.isNegative()) {
+ safeMaximum = Duration.ofMillis(1);
+ }
+ Duration configured =
parsePositiveDuration(properties.getCollectionLeaseRenewalInterval(),
safeMaximum,
Review Comment:
**[Info]** The `LeaseHeartbeat` inner class is well-designed. A short
Javadoc on the class would help future readers understand the heartbeat
lifecycle at a glance.
##########
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}")
Review Comment:
**[Info]** The `renew` SQL correctly enforces `expires_at > now` to prevent
an expired holder from reclaiming the lease. Since both sides use the
Java-computed `now` parameter consistently, this is fine for now, but worth
keeping in mind if the DB ever switches to using its own `NOW()` for expiry
checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
