This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch UNOMI-979-scheduler-lock-lease in repository https://gitbox.apache.org/repos/asf/unomi.git
commit 4de1679e3643c99d11769b4552466fc3dc0fdde8 Author: Serge Huber <[email protected]> AuthorDate: Mon Aug 17 10:26:21 2026 +0200 UNOMI-979: Deflake the scheduler test suites The scheduler unit tests failed sporadically on loaded CI runners (SchedulerServiceImplTest.testConcurrentLockAcquisition "expected: <1> but was: <2>", testClusteringSupport "runOnAllNodes task should execute on every node"). Diagnosed from the scheduler's own LOCK-DIAG traces and reproduced on demand under CPU oversubscription; the companion commit fixes the production half (lock-lease expiry). This commit fixes what the tests themselves got wrong, and encodes the rules in the class javadoc: - The setUp() scheduler ran with a 1s lock timeout while multi-node tests created nodes with the 10s default, and it kept polling in the background as an uninvited extra cluster node for the whole test. It now uses the production-default timeout, and the multi-node tests that do not use it stop it first (the testNodeFailure pattern). - testClusteringSupport demanded that one runOnAllNodes task execute on all three nodes within the timeout. That is not a property the implementation promises: all nodes share the task's single schedule document, each period has one phase-dependent winner, and there is no fairness. The test now asserts what IS promised, and the regression it was really protecting - non-executor nodes must poll and run runOnAllNodes tasks - is pinned deterministically in a new test where the non-executor is the only node. - Exact execution counts were asserted while periodic tasks could still fire (fixed-delay, metrics/history, restart, dedicated-executor tests): now cancel-and-quiesce first or assert lower bounds. - Thread.sleep policy, also documented in the class javadoc: sleep-then-assert-positive waits are converted to bounded polls (awaitStatus) or Mockito timeout() verifies; "keep the executor busy" sleeps become latches the test releases; deliberate quiet windows for NEGATIVE assertions keep their sleeps (a poll cannot confirm that nothing happened, and a short window can only miss a violation, never fail a healthy run); poll intervals and genuine workload durations stay. - configureDebugLogging() was dead code: it set slf4j-simple properties while logback-test.xml binds logback. Removed; -DTEST_LOG_LEVEL=DEBUG is the real switch and is now documented, so the next CI failure arrives with LOCK-DIAG traces instead of a bare assertion message. - The stalled-execution recovery test's premise stopped firing when lock renewal was introduced (a live execution's lock no longer expires naturally; verified from traces: renewal succeeds throughout the stall and no expiry verdict fires). Its javadoc now records what it still pins and where the reclaim path keeps unit coverage. Validated by running the four scheduler suites repeatedly under ~2.5x CPU oversubscription - the protocol that reproduced both original failures on demand - with seven consecutive green runs after these changes. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> --- .../impl/scheduler/SchedulerServiceImplTest.java | 277 ++++++++++++++++----- .../impl/scheduler/TaskExecutionManagerTest.java | 50 ++-- 2 files changed, 250 insertions(+), 77 deletions(-) diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java index d3e91d56c..5f9e8ee38 100644 --- a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java +++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImplTest.java @@ -74,6 +74,26 @@ import static org.mockito.Mockito.when; * - RetryTests: Task retry behavior and delay * - MaintenanceTests: Task cleanup and maintenance * - QueryTests: Task querying and filtering + * + * <h3>Debugging</h3> + * Logging is configured by {@code src/test/resources/logback-test.xml}; run with + * {@code -DTEST_LOG_LEVEL=DEBUG} to see the scheduler's {@code LOCK-DIAG} traces, which record + * every lock acquisition, renewal, expiry verdict and recovery decision. A bare assertion + * failure from this suite is rarely diagnosable without them. + * + * <h3>Timing rules for this suite</h3> + * The {@code setUp()} scheduler keeps polling in the background for the whole test, so: + * <ul> + * <li>Multi-node tests that do not use the setUp scheduler must {@code preDestroy()} it first — + * otherwise it participates in the shared persistence store as an extra, unaccounted node + * (see {@code testNodeFailure} for the pattern).</li> + * <li>Never assert an exact execution count while the task can still fire: cancel the task or + * stop the scheduler first, or assert a lower bound.</li> + * <li>Prefer latches the test releases over {@code Thread.sleep(N)} for "keep the executor busy + * while I check something" — a fixed sleep is a bet on scheduler timing that loaded CI + * runners lose. Sleeps are acceptable as poll intervals inside bounded retry loops and as + * genuine workload where the duration itself is the test subject.</li> + * </ul> */ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) @@ -89,7 +109,17 @@ public class SchedulerServiceImplTest { private static final long TEST_TIMEOUT = 15000; // 15 seconds — extra margin for loaded CI runners /** Time unit for test timeouts */ private static final TimeUnit TEST_TIME_UNIT = TimeUnit.MILLISECONDS; - /** Lock timeout for testing lock expiration */ + /** + * Lock timeout for the setUp scheduler and for multi-node tests, matching + * {@code TaskLockManager}'s production default. Deliberately NOT short: the setUp scheduler + * keeps polling in the background during every test, and a node whose lock timeout is shorter + * than a peer's renewal cadence (peer timeout / 3) declares that peer's live locks expired in + * the gap between renewals — before lock leases this stole locks from mid-execution tasks and + * double-ran them (the CI flake in testConcurrentLockAcquisition). All nodes sharing one store + * must agree on this value unless lock expiry itself is the behaviour under test. + */ + private static final long DEFAULT_LOCK_TIMEOUT = 10000; // 10 seconds + /** Short lock timeout for tests that exercise lock expiration; set it explicitly per test. */ private static final long TEST_LOCK_TIMEOUT = 1000; // 1 second /** Thread pool size for parallel execution */ private static final int TEST_THREAD_POOL_SIZE = 4; @@ -113,18 +143,12 @@ public class SchedulerServiceImplTest { // Test categories with documentation // JUnit 5 provides tags; marker interfaces removed - - private static void configureDebugLogging() { - // Enable debug logging for scheduler package - System.setProperty("org.slf4j.simpleLogger.log.org.apache.unomi.services.impl.scheduler", "DEBUG"); - System.setProperty("org.slf4j.simpleLogger.showDateTime", "true"); - System.setProperty("org.slf4j.simpleLogger.dateTimeFormat", "yyyy-MM-dd HH:mm:ss.SSS"); - System.setProperty("org.slf4j.simpleLogger.showThreadName", "true"); - } + // (An earlier configureDebugLogging() helper set org.slf4j.simpleLogger.* properties here; + // it was dead code — logback-test.xml binds logback, which ignores those. Use + // -DTEST_LOG_LEVEL=DEBUG instead, see the class javadoc.) @BeforeEach public void setUp() throws IOException { - configureDebugLogging(); CustomObjectMapper.getCustomInstance().registerBuiltInItemTypeClass(ScheduledTask.ITEM_TYPE, ScheduledTask.class); securityService = TestHelper.createSecurityService(); @@ -155,9 +179,10 @@ public class SchedulerServiceImplTest { false, 0); // Set TTL to 0 for immediate purging in tests - // Configure scheduler for testing + // Configure scheduler for testing. The lock timeout matches the production default and + // the multi-node tests' nodes; tests exercising expiry shorten it themselves. schedulerService.setThreadPoolSize(TEST_THREAD_POOL_SIZE); - schedulerService.setLockTimeout(TEST_LOCK_TIMEOUT); + schedulerService.setLockTimeout(DEFAULT_LOCK_TIMEOUT); schedulerService.postConstruct(); } @@ -284,7 +309,10 @@ public class SchedulerServiceImplTest { .schedule(); assertTrue(executionLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), "Task should execute three times"); - assertEquals(3, executionCount.get(), "Task should execute exactly three times"); + // Lower bound, not equality: the periodic task keeps firing between the latch release + // and this line, so an exact count is a race against the next period (cf. the fixed-rate + // test above, which already asserts >= for the same reason). + assertTrue(executionCount.get() >= 3, "Task should execute at least three times"); if (workerError.get() != null) { throw new AssertionError("Assertion failed in worker thread", workerError.get()); } @@ -452,7 +480,10 @@ public class SchedulerServiceImplTest { @Test @Tag("ClusterTests") public void testClusteringSupport() throws Exception { - // Test clustering behavior with multiple nodes + // Test clustering behavior with multiple nodes. The setUp scheduler is not part of this + // cluster: stop it so it cannot interfere with the three nodes' tasks or the node + // detection markers below (testNodeFailure pattern). + schedulerService.preDestroy(); SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = TestHelper.createSchedulerService("node2", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl nonExecutorNode = TestHelper.createSchedulerService("node3", persistenceService, executionContextManager, bundleContext, clusterService, -1, false, true); @@ -470,7 +501,10 @@ public class SchedulerServiceImplTest { persistenceService.refresh(); CountDownLatch exclusiveLatch = new CountDownLatch(1); - CountDownLatch allNodesLatch = new CountDownLatch(3); // one execution observed per node + // Opens on the FIRST runOnAllNodes execution, on whichever node wins the first + // round; allNodesNodes records the distinct winners (see the comment below on why + // "all three nodes" is not a property the implementation promises). + CountDownLatch allNodesLatch = new CountDownLatch(1); Set<String> exclusiveNodes = ConcurrentHashMap.newKeySet(); Set<String> allNodesNodes = ConcurrentHashMap.newKeySet(); @@ -496,9 +530,8 @@ public class SchedulerServiceImplTest { @Override public void execute(ScheduledTask task, TaskStatusCallback callback) { - if (allNodesNodes.add(task.getExecutingNodeId())) { - allNodesLatch.countDown(); - } + allNodesNodes.add(task.getExecutingNodeId()); + allNodesLatch.countDown(); callback.complete(); } }; @@ -531,13 +564,22 @@ public class SchedulerServiceImplTest { exclusiveNodes.contains("node3"), "Exclusive task must not execute on a non-executor node"); + // What runOnAllNodes actually promises, as implemented: ANY node - including a + // non-executor - may poll and run the task. It does NOT promise that every node runs + // it: all nodes share the task's single schedule (one lastExecutionDate / + // nextScheduledExecution on one document), so each period has ONE phase-dependent + // winner and there is no fairness across nodes. This test used to demand an execution + // from all three nodes within the timeout, which made it a lottery over checker-tick + // phases - the "runOnAllNodes task should execute on every node" CI flake. The + // non-executor half of the guarantee is pinned deterministically in + // testRunOnAllNodesExecutesOnNonExecutorNode, where the non-executor is the only node. assertTrue( allNodesLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), - "runOnAllNodes task should execute on every node including non-executors"); - assertTrue(allNodesNodes.contains("node1"), "runOnAllNodes should run on node1"); - assertTrue(allNodesNodes.contains("node2"), "runOnAllNodes should run on node2"); - assertTrue(allNodesNodes.contains("node3"), "runOnAllNodes should run on non-executor node3"); + "runOnAllNodes task should execute on at least one node"); + // Keep the lock-inspection task's execution alive until this test has finished + // inspecting its lock, instead of betting on a fixed sleep outlasting the checks. + CountDownLatch lockTaskRelease = new CountDownLatch(1); TaskExecutor clusterLockTestExecutor = new TaskExecutor() { @Override public String getTaskType() { @@ -546,7 +588,7 @@ public class SchedulerServiceImplTest { @Override public void execute(ScheduledTask task, TaskStatusCallback callback) { try { - Thread.sleep(5000); + lockTaskRelease.await(TEST_TIMEOUT, TEST_TIME_UNIT); callback.complete(); } catch (InterruptedException e) { callback.fail(e.getMessage()); @@ -554,39 +596,103 @@ public class SchedulerServiceImplTest { } }; - schedulerService.registerTaskExecutor(clusterLockTestExecutor); + // Register on the cluster's own executor nodes (the setUp scheduler is stopped). + node1.registerTaskExecutor(clusterLockTestExecutor); + node2.registerTaskExecutor(clusterLockTestExecutor); - // Test lock management - ScheduledTask lockTask = node1.newTask("cluster-lock-test") - .disallowParallelExecution() - .schedule(); + try { + // Test lock management + ScheduledTask lockTask = node1.newTask("cluster-lock-test") + .disallowParallelExecution() + .schedule(); + + // Refresh persistence to ensure task updates are available (handles refresh delay) + persistenceService.refresh(); + // Wait until the task has a lock owner. Deadline-based rather than + // TestHelper.retryUntil's fixed 20x100ms budget, which a loaded runner exceeds + // (dispatch needs a checker tick plus the simulated refresh delay). + ScheduledTask lockedTask = null; + long lockDeadline = System.currentTimeMillis() + TEST_TIMEOUT; + while (System.currentTimeMillis() < lockDeadline) { + lockedTask = persistenceService.load(lockTask.getItemId(), ScheduledTask.class); + if (lockedTask != null && lockedTask.getLockOwner() != null) { + break; + } + Thread.sleep(100); + } + assertNotNull(lockedTask, "Lock task should be persisted"); + assertNotNull(lockedTask.getLockOwner(), "Task should have lock owner"); + assertNotNull(lockedTask.getLockDate(), "Task should have lock date"); + + // Test lock release - directly update task in persistence + lockedTask.setLockOwner(null); + lockedTask.setLockDate(null); + lockedTask.setLockLeaseMillis(0); + persistenceService.save(lockedTask); + + // Refresh index to ensure changes are visible + persistenceService.refreshIndex(ScheduledTask.class); + + // Get latest state and verify lock release + ScheduledTask releasedTask = persistenceService.load(lockTask.getItemId(), ScheduledTask.class); + assertNull(releasedTask.getLockOwner(), "Lock should be released"); + } finally { + lockTaskRelease.countDown(); + } - // Refresh persistence to ensure task updates are available (handles refresh delay) - persistenceService.refresh(); - // Retry until task has lock owner (handles refresh delay for updates) - ScheduledTask lockedTask = TestHelper.retryUntil( - () -> persistenceService.load(lockTask.getItemId(), ScheduledTask.class), - t -> t != null && t.getLockOwner() != null - ); - assertNotNull(lockedTask.getLockOwner(), "Task should have lock owner"); - assertNotNull(lockedTask.getLockDate(), "Task should have lock date"); + } finally { + node1.preDestroy(); + node2.preDestroy(); + nonExecutorNode.preDestroy(); + } + } - // Test lock release - directly update task in persistence - lockedTask.setLockOwner(null); - lockedTask.setLockDate(null); - persistenceService.save(lockedTask); + /** + * The non-executor half of the runOnAllNodes guarantee, pinned deterministically: a node + * with {@code executorNode=false} must still poll for and execute runOnAllNodes tasks. + * <p> + * testClusteringSupport cannot assert this reliably — with executor nodes present, all nodes + * race on the task's single shared schedule and there is no fairness, so whether the + * non-executor ever wins a round within the timeout is checker-phase luck. Here the + * non-executor is the ONLY node, so if it does not poll runOnAllNodes work (the regression + * this pins), nothing executes and the latch times out. + */ + @Test + @Tag("ClusterTests") + public void testRunOnAllNodesExecutesOnNonExecutorNode() throws Exception { + schedulerService.preDestroy(); + SchedulerServiceImpl nonExecutorOnly = TestHelper.createSchedulerService( + "solo-non-executor", persistenceService, executionContextManager, bundleContext, clusterService, -1, false, true); - // Refresh index to ensure changes are visible - persistenceService.refreshIndex(ScheduledTask.class); + try { + CountDownLatch executed = new CountDownLatch(1); + AtomicReference<String> executingNode = new AtomicReference<>(); + + TaskExecutor executor = new TaskExecutor() { + @Override + public String getTaskType() { + return "all-nodes-solo-test"; + } - // Get latest state and verify lock release - ScheduledTask releasedTask = persistenceService.load(lockTask.getItemId(), ScheduledTask.class); - assertNull(releasedTask.getLockOwner(), "Lock should be released"); + @Override + public void execute(ScheduledTask task, TaskStatusCallback callback) { + executingNode.set(task.getExecutingNodeId()); + executed.countDown(); + callback.complete(); + } + }; + nonExecutorOnly.registerTaskExecutor(executor); + + nonExecutorOnly.newTask("all-nodes-solo-test") + .runOnAllNodes() + .withPeriod(100, TimeUnit.MILLISECONDS) + .schedule(); + assertTrue(executed.await(TEST_TIMEOUT, TEST_TIME_UNIT), + "a non-executor node must poll for and run runOnAllNodes tasks"); + assertEquals("solo-non-executor", executingNode.get()); } finally { - node1.preDestroy(); - node2.preDestroy(); - nonExecutorNode.preDestroy(); + nonExecutorOnly.preDestroy(); } } @@ -748,17 +854,29 @@ public class SchedulerServiceImplTest { failureLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), "Task should fail once"); - // Verify metrics and history + // The 100ms-period task keeps executing (and failing) after the latches fire, so exact + // counts are a race against the next period. Cancel it and wait for the cancellation to + // land before reading anything. + schedulerService.cancelTask(task.getItemId()); + TestHelper.retryUntil( + () -> schedulerService.getTask(task.getItemId()), + t -> t != null && t.getStatus() != ScheduledTask.TaskStatus.RUNNING + && t.getStatus() != ScheduledTask.TaskStatus.SCHEDULED); + + // Verify metrics and history. Successes are exact (the executor only ever completes the + // first two); failures are a lower bound (every later period failed until the cancel won). ScheduledTask finalTask = schedulerService.getTask(task.getItemId()); @SuppressWarnings("unchecked") List<Map<String, Object>> history = (List<Map<String, Object>>) finalTask.getStatusDetails().get("executionHistory"); assertNotNull(history, "Should have execution history"); - assertEquals(3, history.size(), "Should have 3 history entries"); + assertTrue(history.size() >= 3, "Should have at least 3 history entries, had " + history.size()); assertEquals(2, finalTask.getSuccessCount(), "Should have 2 successful executions"); - assertEquals(1, finalTask.getFailureCount(), "Should have 1 failed execution"); - assertEquals(3, finalTask.getSuccessCount() + finalTask.getFailureCount(), "Total executions should be 3"); + assertTrue(finalTask.getFailureCount() >= 1, + "Should have at least 1 failed execution, had " + finalTask.getFailureCount()); + // No history-size == successCount+failureCount equality here: an execution in flight + // while the cancel lands may or may not get its failure recorded, by design. // Verify history entries int successEntries = 0; @@ -774,7 +892,7 @@ public class SchedulerServiceImplTest { } assertEquals(2, successEntries, "Should have 2 successful executions"); - assertEquals(1, failureEntries, "Should have 1 failed execution"); + assertTrue(failureEntries >= 1, "Should have at least 1 failed execution"); // Verify metrics assertTrue(schedulerService.getMetric("tasks.completed") > 0, "Should have completed tasks metric"); @@ -1023,13 +1141,24 @@ public class SchedulerServiceImplTest { * tasks that already executed, stranding the task in CRASHED state forever. The * execution manager must recognize that the execution it owns is still alive, reclaim * the task and process the failure (and its retry) normally. + * + * <p>NOTE: since lock renewal was introduced (the lock is re-stamped every lockTimeout/3 + * while the executor runs), a stalled-but-live execution's lock no longer expires from + * natural timing, so the CRASH-mark this test was written around does not fire anymore - + * verified from the LOCK-DIAG traces: renewal succeeds throughout the stall and no expiry + * verdict ever triggers. The test remains valuable as a pin on the surviving behaviour + * (a failure reported after a stall longer than the lock timeout still schedules its + * retries and completes), and its assertions were already written to tolerate both worlds + * (>= 3 executions). The reclaim path itself is now only reachable when renewal genuinely + * stops (e.g. a GC pause longer than the full lease) and is covered at unit level in + * TaskExecutionManagerTest. */ @Test @Tag("RetryTests") public void testOneShotRetryAfterRecoveryMarksLiveExecutionCrashed() throws Exception { - // setUp() only sets the lock timeout on the scheduler service; the lock manager - // created by TestHelper keeps its 10s default. Shorten it here so a stalled - // execution's lock actually expires within this test's stall window. + // Shorten the lock timeout so the stall below dwarfs it. (setLockTimeout on the service + // propagates to the lock manager as well; setting the lock manager directly is + // equivalent and kept for clarity about what the timeout is FOR here.) schedulerService.getLockManager().setLockTimeout(TEST_LOCK_TIMEOUT); CountDownLatch completionLatch = new CountDownLatch(1); @@ -1320,6 +1449,7 @@ public class SchedulerServiceImplTest { schedulerService.setLockTimeout(TEST_LOCK_TIMEOUT); CountDownLatch executionLatch = new CountDownLatch(1); + CountDownLatch holdRelease = new CountDownLatch(1); AtomicBoolean taskStarted = new AtomicBoolean(false); TaskExecutor executor = new TaskExecutor() { @@ -1335,8 +1465,9 @@ public class SchedulerServiceImplTest { taskStarted.set(true); executionLatch.countDown(); - // Hold the lock longer than timeout - Thread.sleep(TEST_LOCK_TIMEOUT * 2); + // Hold the lock until the test has finished inspecting it - a latch the + // test releases, not a fixed sleep the test hopes is long enough. + holdRelease.await(TEST_TIMEOUT, TEST_TIME_UNIT); callback.complete(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -1359,12 +1490,15 @@ public class SchedulerServiceImplTest { // Directly update task to simulate lock expiration runningTask.setLockOwner(null); runningTask.setLockDate(null); + runningTask.setLockLeaseMillis(0); persistenceService.save(runningTask); persistenceService.refreshIndex(ScheduledTask.class); // Check lock status after manual release ScheduledTask updatedTask = persistenceService.load(task.getItemId(), ScheduledTask.class); assertNull(updatedTask.getLockOwner(), "Lock should be released after manual update"); + + holdRelease.countDown(); } /** @@ -1631,6 +1765,12 @@ public class SchedulerServiceImplTest { @Test @Tag("ClusterTests") public void testConcurrentLockAcquisition() throws Exception { + // This test is about node1/node2 only: stop the setUp scheduler so the "two-node" cluster + // really has two nodes (testNodeFailure pattern). It used to stay up with a 1s lock + // timeout against these nodes' 10s, declare their live locks expired between renewals, + // and mark the running task CRASHED — which a peer then re-dispatched concurrently + // (the "expected: <1> but was: <2>" CI flake). + schedulerService.preDestroy(); SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = TestHelper.createSchedulerService("node2", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); @@ -1738,6 +1878,8 @@ public class SchedulerServiceImplTest { @Test @Tag("ClusterTests") public void testTaskRebalancing() throws Exception { + // Two-node test: stop the setUp scheduler so it is not a hidden third participant. + schedulerService.preDestroy(); SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = null; try { @@ -1861,6 +2003,8 @@ public class SchedulerServiceImplTest { @Test @Tag("ClusterTests") public void testLockStealing() throws Exception { + // Two-node test: stop the setUp scheduler so it is not a hidden third participant. + schedulerService.preDestroy(); SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = TestHelper.createSchedulerService("node2", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); @@ -1946,6 +2090,10 @@ public class SchedulerServiceImplTest { @Test public void testNodeAffinity() throws Exception { + // Three-node test. Stop the setUp scheduler: getActiveNodes() falls back to scanning + // tasks with recent locks, and a foreign recovery pass that clears the detection tasks' + // locks below would silently shrink the cluster this test asserts on. + schedulerService.preDestroy(); // Create test nodes with cluster service SchedulerServiceImpl node1 = TestHelper.createSchedulerService("node1", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); SchedulerServiceImpl node2 = TestHelper.createSchedulerService("node2", persistenceService, executionContextManager, bundleContext, clusterService, -1, true, true); @@ -2425,7 +2573,9 @@ public class SchedulerServiceImplTest { newSchedulerService.preDestroy(); assertTrue(executed, "Task should execute after scheduler restart"); - assertEquals(2, executionCount.get(), "Task should have executed twice"); + // Lower bound: the 500ms-period task may legitimately fire again between the latch + // release and preDestroy() completing on a slow runner. + assertTrue(executionCount.get() >= 2, "Task should have executed at least twice"); // Verify the reloaded task has same ID ScheduledTask reloadedTask = persistenceService.load(persistentTask.getItemId(), ScheduledTask.class); @@ -2757,10 +2907,13 @@ public class SchedulerServiceImplTest { assertTrue( secondExecutionLatch.await(TEST_TIMEOUT * 2, TEST_TIME_UNIT), "Task should execute after restart with dedicated executor"); - assertEquals(2, executionCount.get(), "Task should execute twice"); - // Clean up + // Stop the scheduler before asserting the count, exactly like the first assertion above: + // the task runs at 100ms fixed rate, so a third tick can fire between the latch release + // and the assert. After preDestroy() the count is stable; >= tolerates a tick that + // squeezed in before shutdown took effect. newSchedulerService.preDestroy(); + assertTrue(executionCount.get() >= 2, "Task should execute at least twice"); } /** diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java index bafcc5356..9c045bf45 100644 --- a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java +++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManagerTest.java @@ -128,6 +128,19 @@ public class TaskExecutionManagerTest { assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus()); } + /** + * Waits for the wrapper's asynchronous terminal transition to land on the shared task object. + * The executor's callback returns before the wrapper finishes its bookkeeping, so asserting + * the final status right after the latch (or after a fixed sleep) races the wrapper thread. + */ + private static void awaitStatus(ScheduledTask task, ScheduledTask.TaskStatus expected, long timeoutMs) + throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMs; + while (task.getStatus() != expected && System.currentTimeMillis() < deadline) { + Thread.sleep(20); + } + } + @Test public void testExecuteTaskDuplicateDispatchIsSkipped() throws Exception { CountDownLatch started = new CountDownLatch(1); @@ -148,6 +161,9 @@ public class TaskExecutionManagerTest { // Second dispatch while claim held executionManager.executeTask(task, executor); release.countDown(); + // Deliberate quiet window for a NEGATIVE assertion: a wrongly accepted duplicate + // dispatch would start within milliseconds. Too short can only miss a violation + // (false green), never fail a healthy run. Thread.sleep(200); assertEquals(1, runs.get()); } @@ -273,7 +289,7 @@ public class TaskExecutionManagerTest { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.FAILED, 5000); assertEquals(ScheduledTask.TaskStatus.FAILED, task.getStatus()); assertEquals(1, task.getFailureCount()); assertTrue(task.isEnabled()); @@ -298,7 +314,7 @@ public class TaskExecutionManagerTest { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.SCHEDULED, 5000); assertEquals(0, task.getFailureCount()); assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus()); assertNotNull(task.getNextScheduledExecution()); @@ -340,7 +356,7 @@ public class TaskExecutionManagerTest { releaser.start(); executionManager.shutdown(); releaser.join(2000); - Thread.sleep(200); + awaitStatus(task, ScheduledTask.TaskStatus.SCHEDULED, 5000); assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus()); assertEquals(1, task.getFailureCount()); // No second attempt — retry schedule skipped after scheduler shutdown @@ -362,7 +378,7 @@ public class TaskExecutionManagerTest { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.COMPLETED, 5000); assertEquals(ScheduledTask.TaskStatus.COMPLETED, task.getStatus()); assertFalse(task.isEnabled()); assertNull(task.getNextScheduledExecution()); @@ -384,7 +400,7 @@ public class TaskExecutionManagerTest { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.SCHEDULED, 5000); assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus()); assertNotNull(task.getNextScheduledExecution()); assertTrue(task.getNextScheduledExecution().getTime() >= before + 5_000); @@ -406,7 +422,7 @@ public class TaskExecutionManagerTest { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.COMPLETED, 5000); assertEquals(ScheduledTask.TaskStatus.COMPLETED, task.getStatus()); } @@ -427,6 +443,8 @@ public class TaskExecutionManagerTest { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); + // Deliberate quiet window for a NEGATIVE assertion (callbacks must have been ignored); + // a poll cannot confirm that nothing happened. Thread.sleep(100); assertEquals(ScheduledTask.TaskStatus.CANCELLED, task.getStatus()); assertEquals(completedBefore, metricsManager.getMetric(TaskMetricsManager.METRIC_TASKS_COMPLETED)); @@ -448,7 +466,7 @@ public class TaskExecutionManagerTest { executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); + awaitStatus(task, ScheduledTask.TaskStatus.COMPLETED, 5000); assertEquals(ScheduledTask.TaskStatus.COMPLETED, task.getStatus()); assertFalse(task.isEnabled()); } @@ -545,12 +563,12 @@ public class TaskExecutionManagerTest { ScheduledTask task = TaskTestFixtures.baseTask("cancel-race"); executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); - assertEquals(ScheduledTask.TaskStatus.CANCELLED, task.getStatus()); // persistTerminalState() is skipped (terminal transition correctly bailed out above), but // the wrapper's cleanup still CAS-clears executingNodeId once; that write is expected to // fail harmlessly against a real store since the document moved on to CANCELLED. - verify(schedulerService, times(1)).saveTaskWithRefresh(any()); + // timeout() waits for the asynchronous cleanup instead of betting a fixed sleep on it. + verify(schedulerService, timeout(5000).times(1)).saveTaskWithRefresh(any()); + assertEquals(ScheduledTask.TaskStatus.CANCELLED, task.getStatus()); assertEquals(0, metricsManager.getMetric(TaskMetricsManager.METRIC_TASKS_COMPLETED)); } @@ -572,12 +590,12 @@ public class TaskExecutionManagerTest { ScheduledTask task = TaskTestFixtures.baseTask("peer-lock"); executionManager.executeTask(task, executor); assertTrue(done.await(5, TimeUnit.SECONDS)); - Thread.sleep(100); - assertEquals(ScheduledTask.TaskStatus.RUNNING, task.getStatus()); // persistTerminalState() is skipped (peer holds the lock), but the wrapper's cleanup still // CAS-clears executingNodeId once; that write is expected to fail harmlessly against a real // store since the peer is the authoritative owner. - verify(schedulerService, times(1)).saveTaskWithRefresh(any()); + // timeout() waits for the asynchronous cleanup instead of betting a fixed sleep on it. + verify(schedulerService, timeout(5000).times(1)).saveTaskWithRefresh(any()); + assertEquals(ScheduledTask.TaskStatus.RUNNING, task.getStatus()); } @Test @@ -600,9 +618,11 @@ public class TaskExecutionManagerTest { }; ScheduledTask task = TaskTestFixtures.baseTask("abort-prep"); executionManager.executeTask(task, executor); - Thread.sleep(300); - assertEquals(1, executed.getCount(), "executor must not run after shutdown-abort"); + // Positive half: wait for the asynchronous abort to land instead of a fixed sleep. + awaitStatus(task, ScheduledTask.TaskStatus.CRASHED, 5000); assertEquals(ScheduledTask.TaskStatus.CRASHED, task.getStatus()); + // Negative half: the executor must never have run (green-direction check). + assertEquals(1, executed.getCount(), "executor must not run after shutdown-abort"); assertNull(task.getLockOwner()); verify(schedulerService, atLeastOnce()).saveTask(any(ScheduledTask.class), eq(true)); }
