youngkermit8-coder opened a new issue, #7155: URL: https://github.com/apache/rocketmq-dashboard/issues/7155
### Before Creating the Bug Report - [x] I searched existing issues and PRs for collector cancellation, checkpoint and lease-loss work; no equivalent post-checkpoint fix was found in these bounded searches. - [x] This concerns Studio's native metric scheduler, not a live broker usage question. - [x] The exact unchanged source and controlled reproduction are specified below. ### Studio Version `rocketmq-studio@127c47b06a268cc74224d5f34c56ec7c36ac0d55`, built from source. No production patch, POM change, dependency override or unrelated candidate was applied. Only the diagnostic test below was added. ### Runtime Environment Windows / Temurin 21.0.12+8 / Maven 3.9.9. The fixture runs the real scheduler and worker executor. Collectors, snapshot/lease repositories, alert processor and the heartbeat timer are mocks. No application context, MySQL, real cloud, NameServer or broker is started. ### Connected RocketMQ Cluster None. This reproduces worker admission after cancellation independently of the remote metric source. It does not demonstrate a live cluster incident or real notification delivery. ### Build Toolchain JDK 21, Maven 3.9.9; Mockito 5.23.0 from this version's test dependencies. The targeted command below compiled and executed all four parameterized invocations. Checkstyle reported zero violations. The command deliberately ends with Maven exit 1 because three assertions about unwanted side effects fail on the unchanged production code; it is not a passing fixed regression, full-suite run or package validation. ### Describe the Bug The per-job cancellation flag added by [#6690](https://github.com/apache/rocketmq-dashboard/pull/6690) correctly rejects a collector result cancelled **before entry** to `persist()`. This report is a follow-up about work cancelled **while a later checkpoint call is in flight**: 1. The job passes the entry cancellation check, then blocks inside `collectionLease.tryAcquire()`. The real scheduler expires the job and marks its flag. If that call absorbs the interrupt and later returns true while a sibling keeps the pass alive, `persist()` still starts snapshot persistence and alert processing. 2. If cancellation instead happens during an already-started `saveAll()`, that write may finish, but `persist()` still begins alert processing afterward. 3. Lease-loss admission has a related ordering: a worker can run before `submit()` returns and its Future is registered in the heartbeat's jobs list. If a renewal fails during its snapshot call, the real heartbeat marks the pass lost, but there may be no registered Future to cancel yet. After the call returns, the old flow begins alert processing without checking the heartbeat again. These are cooperative checkpoint gaps, not a claim that Java can forcibly cancel JDBC, that already-started transactions must roll back, or that a post-checkpoint check supplies atomic database fencing. ### Steps to Reproduce At the pinned version, add this exact fixture as `server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorCancellationCheckpointTest.java`: ```java /* * 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.instance.InstanceRepository; import org.apache.rocketmq.studio.instance.InstanceVO; import org.apache.rocketmq.studio.ops.alert.AlertDomain; import org.apache.rocketmq.studio.ops.alert.NativeAlertProcessor; import org.junit.jupiter.api.Timeout; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.ValueSource; import java.time.Instant; import java.util.List; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @Timeout(20) class CollectorCancellationCheckpointTest { enum Checkpoint { LEASE, SNAPSHOT } @ParameterizedTest @ValueSource(booleans = {false, true}) void snapshotCompletionChecksLeaseLossBeforeAlertProcessingTest(boolean leaseLost) throws Exception { AlertingProperties properties = new AlertingProperties(); properties.setCollectionParallelism(1); properties.setCollectionTimeout("PT30S"); InstanceVO instance = instance("snapshot-lease"); MetricSample sample = sample(instance); 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.metricKeys()).thenReturn(Set.of("nameserver.availability")); when(collector.collect(instance)).thenReturn(List.of(sample)); AlertCollectionLease lease = mock(AlertCollectionLease.class); when(lease.tryAcquire()).thenReturn(true); when(lease.renew()).thenReturn(!leaseLost); ScheduledExecutorService renewals = mock(ScheduledExecutorService.class); ScheduledFuture<?> renewalTask = mock(ScheduledFuture.class); AtomicReference<Runnable> renewal = new AtomicReference<>(); when(renewals.scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), eq(TimeUnit.MILLISECONDS))) .thenAnswer(invocation -> { renewal.set(invocation.getArgument(0)); return renewalTask; }); MetricSnapshotRepository snapshots = mock(MetricSnapshotRepository.class); doAnswer(invocation -> { // Invoke the real heartbeat callback while a snapshot call is in flight. renewal.get().run(); return null; }).when(snapshots).saveAll(List.of(sample)); NativeAlertProcessor processor = mock(NativeAlertProcessor.class); CountDownLatch workerFinished = new CountDownLatch(1); ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()) { @Override public void execute(Runnable command) { super.execute(command); // A worker may run before submit() returns and registers its Future with the pass. // Hold that valid ordering deterministically, rather than racing the scheduler cleanup. try { assertTrue(workerFinished.await(5, TimeUnit.SECONDS), "worker must finish before registration"); } catch (InterruptedException error) { Thread.currentThread().interrupt(); throw new IllegalStateException(error); } } @Override protected void afterExecute(Runnable task, Throwable error) { try { super.afterExecute(task, error); } finally { workerFinished.countDown(); } } }; CollectorScheduler scheduler = new CollectorScheduler(properties, instances, List.of(collector), List.of(), snapshots, processor, lease, executor, renewals); try { scheduler.collect(); verify(lease).renew(); verify(snapshots).saveAll(List.of(sample)); if (leaseLost) { verify(processor, never()).processSuccessfulCollection(any(), any()); } else { verify(processor).processSuccessfulCollection(any(), eq(List.of(sample))); } } finally { scheduler.stopCollectionExecutor(); assertTrue(executor.awaitTermination(3, TimeUnit.SECONDS), "test worker must terminate"); } } @ParameterizedTest @EnumSource(Checkpoint.class) void cancellationDuringBlockingWorkStopsSubsequentSideEffectsTest(Checkpoint checkpoint) throws Exception { AlertingProperties properties = new AlertingProperties(); properties.setCollectionParallelism(2); properties.setCollectionTimeout("PT2S"); InstanceVO slow = instance("slow"); InstanceVO healthy = instance("healthy"); InstanceVO blocker = instance("blocker"); InstanceVO lastBlocker = instance("last-blocker"); InstanceRepository instances = mock(InstanceRepository.class); when(instances.findAll()).thenReturn(List.of(slow, healthy, blocker, lastBlocker)); MetricSample slowSample = sample(slow); MetricSample healthySample = sample(healthy); BlockingCheckpoint gate = new BlockingCheckpoint(); CountDownLatch lastBlockerStarted = new CountDownLatch(1); CountDownLatch releaseBlockers = new CountDownLatch(1); CountDownLatch slowWorkerFinished = new CountDownLatch(1); AtomicReference<Thread> slowThread = new AtomicReference<>(); AtomicReference<Throwable> passFailure = new AtomicReference<>(); ClusterMetricsCollector collector = mock(ClusterMetricsCollector.class); when(collector.supports(any())).thenReturn(true); when(collector.metricKeys()).thenReturn(Set.of("nameserver.availability")); when(collector.collect(slow)).thenAnswer(invocation -> { slowThread.set(Thread.currentThread()); return List.of(slowSample); }); when(collector.collect(healthy)).thenReturn(List.of(healthySample)); when(collector.collect(blocker)).thenAnswer(invocation -> { awaitBlocker(releaseBlockers); return List.of(); }); when(collector.collect(lastBlocker)).thenAnswer(invocation -> { lastBlockerStarted.countDown(); awaitBlocker(releaseBlockers); return List.of(); }); AlertCollectionLease lease = mock(AlertCollectionLease.class); when(lease.tryAcquire()).thenAnswer(invocation -> { if (checkpoint == Checkpoint.LEASE && Thread.currentThread() == slowThread.get()) { gate.awaitReleaseAbsorbingCancellation(); } return true; }); MetricSnapshotRepository snapshots = mock(MetricSnapshotRepository.class); doAnswer(invocation -> { if (checkpoint == Checkpoint.SNAPSHOT && Thread.currentThread() == slowThread.get()) { gate.awaitReleaseAbsorbingCancellation(); } return null; }).when(snapshots).saveAll(any()); NativeAlertProcessor processor = mock(NativeAlertProcessor.class); ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 2, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()) { @Override protected void afterExecute(Runnable task, Throwable error) { super.afterExecute(task, error); if (Thread.currentThread() == slowThread.get()) { slowWorkerFinished.countDown(); } } }; CollectorScheduler scheduler = new CollectorScheduler(properties, instances, List.of(collector), List.of(), snapshots, processor, lease, executor); Thread pass = new Thread(() -> { try { scheduler.collect(); } catch (Throwable error) { passFailure.set(error); } }); try { pass.start(); assertTrue(gate.entered.await(3, TimeUnit.SECONDS), "slow job must reach the blocking checkpoint"); assertTrue(gate.cancelled.await(5, TimeUnit.SECONDS), "the real scheduler must cancel the slow job"); assertTrue(lastBlockerStarted.await(3, TimeUnit.SECONDS), "a younger job must keep the pass alive"); gate.release.countDown(); assertTrue(slowWorkerFinished.await(2, TimeUnit.SECONDS), "wait for actual worker exit, not cancelled Future"); if (checkpoint == Checkpoint.LEASE) { verify(snapshots, never()).saveAll(List.of(slowSample)); } else { // The snapshot call already began before cancellation; no rollback is promised. verify(snapshots).saveAll(List.of(slowSample)); } verify(processor, never()).processSuccessfulCollection(any(), argThat(rows -> rows.contains(slowSample))); verify(snapshots).saveAll(List.of(healthySample)); verify(processor).processSuccessfulCollection(any(), argThat(rows -> rows.contains(healthySample))); } finally { gate.release.countDown(); releaseBlockers.countDown(); pass.join(TimeUnit.SECONDS.toMillis(3)); scheduler.stopCollectionExecutor(); assertTrue(executor.awaitTermination(3, TimeUnit.SECONDS), "test workers must terminate"); } assertFalse(pass.isAlive(), "collection pass must finish"); assertNull(passFailure.get(), "collection pass must not fail"); } private static void awaitBlocker(CountDownLatch release) { try { release.await(); } catch (InterruptedException cancelled) { // Let the older blocker finish so the younger one can start. } } private static InstanceVO instance(String name) { return InstanceVO.builder().name(name).build(); } private static MetricSample sample(InstanceVO instance) { return new MetricSample("nameserver.availability", AlertDomain.CLUSTER, instance.getName(), null, null, 1D, MetricAvailability.AVAILABLE, Instant.now()); } private static final class BlockingCheckpoint { private final CountDownLatch entered = new CountDownLatch(1); private final CountDownLatch cancelled = new CountDownLatch(1); private final CountDownLatch release = new CountDownLatch(1); private void awaitReleaseAbsorbingCancellation() { entered.countDown(); while (true) { try { release.await(); return; } catch (InterruptedException interrupted) { cancelled.countDown(); } } } } } ``` Run from `server/`, with the Java 21 toolchain: ```sh mvn -B -ntp -Dtest=CollectorCancellationCheckpointTest test ``` My current Windows run used offline dependencies and bounded Maven/test JVM heaps, with the existing Mockito agent supplied explicitly. The critical ordering uses latches, not sleeps: the cancellation cases wait until the actual scheduler interrupts the blocked worker and a younger job has started; the lease-loss case waits for the worker to finish before Future registration, and invokes the real registered heartbeat callback inside the snapshot mock. That timer invocation is controlled, not a measurement of the actual production heartbeat period. ### What Did You Expect to See? After a blocked lease refresh returns, a job already known to be cancelled should not start snapshot persistence. After a snapshot call returns, known per-job cancellation or known pass lease loss should prevent admission to alert processing. Healthy collection should still proceed. An already-started snapshot may finish and no rollback is required. ### What Did You See Instead? Four actual invocations: three `NeverWantedButInvoked` assertion failures, one passing healthy-renewal control, zero errors and zero skips: - cancelled during lease refresh: unexpected `saveAll`; - cancelled during snapshot: unexpected `processSuccessfulCollection`; - failed heartbeat renewal during the forced pre-registration ordering: unexpected `processSuccessfulCollection`; - the same ordering with a successful renewal: the healthy control passes. A separate local audit verified all 1,142 original exported server files against the pinned Git objects and disk, verified the added fixture's exact source, and compared the compiled scheduler to a separately built unchanged package at the same base. This audit is provenance checking, not additional behavioral cases or full-regression evidence. ### Additional Context The inspected [current scheduler](https://github.com/apache/rocketmq-dashboard/blob/127c47b06a268cc74224d5f34c56ec7c36ac0d55/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java) blob is `ba4d2e18d4a4a9b7b5e73f855cb4667ee99e6ff4`. Its `persist()` entry guard is present; there is no cancellation recheck after the lease call or cancellation/heartbeat recheck after the snapshot call. [#5152](https://github.com/apache/rocketmq-dashboard/pull/5152) changes the snapshot query's evaluation-time bound, not these scheduler checkpoints; its actual body and changed-file list were inspected. #6690's existing entry guard should be preserved and credited. Could maintainers confirm the intended cooperative checkpoint policy before a code PR? The narrow local proposal is to recheck the job flag after lease refresh and the job flag plus heartbeat after snapshot persistence, preserving APIs, schema, dependencies, timeouts and already-started write semantics. It does not eliminate a lease change racing the next call. An earlier local prototype and separately pinned tests/JDBC studies predate this Issue. They are not claimed as current-base implementation validation or maintainer agreement. The current reproduction above uses mocked storage, not those historical real-JDBC studies, and proves processor admission rather than actual alert state changes or notification delivery. No code PR, binaries, database, credentials or raw authenticated logs are attached. AI assistance was used for investigation and verification. ### Are You Willing to Submit a Pull Request? - [x] Yes, after necessary design feedback and fresh implementation validation. -- 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]
