You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@unomi.apache.org by sh...@apache.org on 2026/08/17 08:29:07 UTC
(unomi) branch UNOMI-979-scheduler-lock-lease created (now 4de1679e3)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch UNOMI-979-scheduler-lock-lease
in repository https://gitbox.apache.org/repos/asf/unomi.git
at 4de1679e3 UNOMI-979: Deflake the scheduler test suites
This branch includes the following new commits:
new d3963edfc UNOMI-979: Judge scheduler lock expiry by the owner's recorded lease
new 4de1679e3 UNOMI-979: Deflake the scheduler test suites
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(unomi) 01/02: UNOMI-979: Judge scheduler lock expiry by the owner's recorded lease
Posted by sh...@apache.org.
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 d3963edfcb50b3a40e479e784f5e30e4b96739c4
Author: Serge Huber <sh...@apache.org>
AuthorDate: Mon Aug 17 10:26:21 2026 +0200
UNOMI-979: Judge scheduler lock expiry by the owner's recorded lease
A lock's renewal cadence is derived from its owner's configured lock timeout
(lockTimeout/3), but expiry was judged against the OBSERVER's timeout. A node
configured with a shorter timeout than a peer's renewal cadence therefore saw
every renewal gap as a dead lock: it marked the live execution CRASHED, cleared
the lock, and the next peer tick re-dispatched the task while the original
execution was still running. Reproduced deterministically with a 1s-timeout
observer against 10s-timeout workers; in production the same double execution
follows from configuration drift or a rolling upgrade. It was also the root
cause of the CI flakes in SchedulerServiceImplTest ("expected: <1> but was:
<2>").
Locks now record the lease the owner granted itself (ScheduledTask
lockLeaseMillis, stamped on every acquire and renewal, cleared on release), and
isLockExpired() judges against that lease. Documents written before lease
recording carry no lease and fall back to the observer's timeout - the exact
pre-change behaviour - so a rolling upgrade changes nothing for existing locks.
Corrupt (negative) leases fall back the same way rather than widening or
wedging the lock; an absurdly large lease is honoured, because the owner
declared it and stealing early is what causes double execution.
Also in this change:
- ScheduledTask now tolerates unknown JSON properties. Jackson's default
rejects the first unrecognized field, so during a rolling upgrade an older
node would lose the ability to read ANY task document a newer node had
written the moment a field is added - this field or any future one.
- startLockRenewal() warns when the configured lock timeout is at or below the
minimum renewal interval, the one configuration where a node cannot keep its
own lease alive and peers may legitimately recover its live work.
- ES and OpenSearch scheduledTask mappings gain the lockLeaseMillis field.
The divergent-timeout steal is pinned end to end by
testShortTimeoutObserverCannotRecoverLiveRenewedLock (fails as
"expected: <RUNNING> but was: <CRASHED>" with the fix reverted), the recovery
direction by testDeadOwnersShortLeaseDrivesPromptRecoveryByPatientSurvivor
(a patient survivor recovers a dead owner's task as soon as the OWNER's lease
expires - faster than before, and the case that proves failover still works).
Unit coverage in TaskLockManagerTest exercises lease stamping on all three
acquire paths, re-stamping on renewal after a runtime timeout change, clearing
on release, both override directions, the legacy/corrupt fallbacks, boundary
equality and overflow. ScheduledTaskLeaseSerializationTest pins the
persistence format through both real read paths, the legacy-document fallback,
and the newer-version-document case (fails with UnrecognizedPropertyException
without the annotation).
Co-Authored-By: Claude Opus 5 (1M context) <no...@anthropic.com>
---
.../org/apache/unomi/api/tasks/ScheduledTask.java | 36 +++-
.../META-INF/cxs/mappings/scheduledTask.json | 3 +
.../META-INF/cxs/mappings/scheduledTask.json | 3 +
.../impl/scheduler/SchedulerServiceImpl.java | 3 +
.../impl/scheduler/TaskExecutionManager.java | 12 ++
.../services/impl/scheduler/TaskLockManager.java | 29 +++-
.../impl/scheduler/TaskRecoveryManager.java | 2 +
.../services/impl/scheduler/TaskStateManager.java | 2 +
.../ScheduledTaskLeaseSerializationTest.java | 124 ++++++++++++++
.../scheduler/SchedulerServiceClusterRaceTest.java | 144 ++++++++++++++++
.../impl/scheduler/TaskLockManagerTest.java | 182 +++++++++++++++++++++
11 files changed, 535 insertions(+), 5 deletions(-)
diff --git a/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java b/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java
index c5d698e77..5a2f52ace 100644
--- a/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java
+++ b/api/src/main/java/org/apache/unomi/api/tasks/ScheduledTask.java
@@ -16,6 +16,7 @@
*/
package org.apache.unomi.api.tasks;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.unomi.api.Item;
import java.io.Serializable;
@@ -40,6 +41,11 @@ import java.util.HashSet;
* @see org.apache.unomi.api.services.SchedulerService
* @see TaskExecutor
*/
+// Tolerate unknown properties so a node running THIS version can still deserialize task
+// documents written by a NEWER version that has added fields (rolling upgrade window).
+// Without this, Jackson's default rejects the first unrecognized field and the older node
+// loses access to all scheduler state until it is upgraded.
+@JsonIgnoreProperties(ignoreUnknown = true)
public class ScheduledTask extends Item implements Serializable {
/**
@@ -86,6 +92,7 @@ public class ScheduledTask extends Item implements Serializable {
private boolean enabled;
private String lockOwner;
private Date lockDate;
+ private long lockLeaseMillis;
private boolean oneShot;
private boolean allowParallelExecution;
private TaskStatus status;
@@ -343,13 +350,40 @@ public class ScheduledTask extends Item implements Serializable {
/**
* Sets the date when the current lock was acquired.
- *
+ *
* @param lockDate the lock acquisition date
*/
public void setLockDate(Date lockDate) {
this.lockDate = lockDate;
}
+ /**
+ * Duration in milliseconds for which the current lock is valid, as declared by the node that
+ * acquired or last renewed it.
+ * <p>
+ * A lock's lifetime is a lease granted by its <em>owner</em>: the owner renews it on a cadence
+ * derived from its own configured lock timeout, so only the owner's timeout describes when a
+ * missing renewal actually means the owner is dead. Observers must judge expiry against this
+ * recorded lease, never against their own configured timeout — a node configured with a shorter
+ * timeout than the owner's renewal cadence would otherwise "recover" a lock whose owner is alive
+ * and mid-execution, and the task would run twice.
+ *
+ * @return the lease duration in milliseconds, or {@code 0} when the lock predates lease
+ * recording (legacy documents) and the observer's own timeout is the only guide
+ */
+ public long getLockLeaseMillis() {
+ return lockLeaseMillis;
+ }
+
+ /**
+ * Sets the lease duration granted with the current lock.
+ *
+ * @param lockLeaseMillis the lease duration in milliseconds, {@code 0} when unlocked or unknown
+ */
+ public void setLockLeaseMillis(long lockLeaseMillis) {
+ this.lockLeaseMillis = lockLeaseMillis;
+ }
+
/**
* Determines whether this task should execute only once.
* Tasks with period=0 are automatically marked as one-shot tasks.
diff --git a/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json b/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
index f36fc297c..030305e8e 100644
--- a/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
+++ b/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
@@ -75,6 +75,9 @@
"lockDate": {
"type": "date"
},
+ "lockLeaseMillis": {
+ "type": "long"
+ },
"lastExecutionDate": {
"type": "date"
},
diff --git a/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json b/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
index 9c1541d96..a251eebf4 100644
--- a/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
+++ b/persistence-opensearch/core/src/main/resources/META-INF/cxs/mappings/scheduledTask.json
@@ -78,6 +78,9 @@
"lockDate": {
"type": "date"
},
+ "lockLeaseMillis": {
+ "type": "long"
+ },
"lastExecutionDate": {
"type": "date"
},
diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
index 3c98963fc..5e84267da 100644
--- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
+++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
@@ -493,6 +493,7 @@ public class SchedulerServiceImpl implements SchedulerService {
if (newStatus == TaskStatus.COMPLETED || newStatus == TaskStatus.FAILED) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
task.setWaitingForTaskType(null);
task.setCurrentStep(null);
// Update last execution date for completed/failed tasks
@@ -511,6 +512,7 @@ public class SchedulerServiceImpl implements SchedulerService {
} else if (newStatus == TaskStatus.WAITING) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
} else if (newStatus == TaskStatus.RUNNING) {
// Update status details for running tasks
Map<String, Object> details = task.getStatusDetails();
@@ -899,6 +901,7 @@ public class SchedulerServiceImpl implements SchedulerService {
// and PersistenceSchedulerProvider.preDestroy need not unlock RUNNING.
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
if (task.isPersistent() && persistenceProvider != null) {
if (!persistenceProvider.saveTask(task)) {
LOGGER.warn("Failed to persist CRASHED state for task {} during shutdown; "
diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java
index 8f3bd41a4..cde83c97f 100644
--- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java
+++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskExecutionManager.java
@@ -502,6 +502,7 @@ public class TaskExecutionManager {
task.setExecutingNodeId(null);
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
schedulerService.saveTask(task, true);
} catch (Exception e) {
LOGGER.warn("Failed to abort prepared task {} during shutdown: {}",
@@ -532,6 +533,16 @@ public class TaskExecutionManager {
return;
}
long interval = Math.max(MIN_LOCK_RENEWAL_INTERVAL_MS, lockManager.getLockTimeout() / 3);
+ if (interval >= lockManager.getLockTimeout()) {
+ // The renewal floor exceeds the configured timeout, so this node cannot renew its own
+ // lease fast enough to keep it alive: peers may legitimately treat its live locks as
+ // expired between two renewals and recover mid-execution tasks. Surface the
+ // misconfiguration instead of leaving sporadic double executions to be diagnosed.
+ LOGGER.warn("Lock timeout {}ms is at or below the minimum renewal interval {}ms: "
+ + "this node's live locks can expire between renewals and be recovered by peers. "
+ + "Configure a lock timeout of at least {}ms.",
+ lockManager.getLockTimeout(), interval, MIN_LOCK_RENEWAL_INTERVAL_MS * 3);
+ }
LockRenewalHandle handle = new LockRenewalHandle();
activeLockRenewals.put(task.getItemId(), handle);
try {
@@ -668,6 +679,7 @@ public class TaskExecutionManager {
private boolean persistTerminalState(ScheduledTask task) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
if (!task.isPersistent()) {
boolean saved = schedulerService.saveTask(task);
if (!saved) {
diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java
index 978d97fa6..fbe1517c0 100644
--- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java
+++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskLockManager.java
@@ -161,6 +161,7 @@ public class TaskLockManager {
// Just set lock info but don't enforce exclusivity
task.setLockOwner(nodeId);
task.setLockDate(new Date());
+ task.setLockLeaseMillis(lockTimeout);
metricsManager.updateMetric(TaskMetricsManager.METRIC_TASKS_LOCK_ACQUIRED);
return true;
}
@@ -194,8 +195,10 @@ public class TaskLockManager {
latest.setLockOwner(nodeId);
latest.setLockDate(new Date());
+ latest.setLockLeaseMillis(lockTimeout);
task.setLockOwner(nodeId);
task.setLockDate(latest.getLockDate());
+ task.setLockLeaseMillis(lockTimeout);
metricsManager.updateMetric(TaskMetricsManager.METRIC_TASKS_LOCK_ACQUIRED);
// For non-persistent tasks, we just update the in-memory map
@@ -247,9 +250,12 @@ public class TaskLockManager {
task.setSystemMetadata(SEQ_NO, latestTask.getSystemMetadata(SEQ_NO));
task.setSystemMetadata(PRIMARY_TERM, latestTask.getSystemMetadata(PRIMARY_TERM));
- // Step 6: Set lock information
+ // Step 6: Set lock information. The lease records THIS node's timeout with the lock:
+ // renewal cadence is derived from the owner's timeout, so only the owner's timeout says
+ // when a missing renewal means the owner is dead (see isLockExpired()).
task.setLockOwner(nodeId);
task.setLockDate(new Date());
+ task.setLockLeaseMillis(lockTimeout);
LOGGER.debug("LOCK-DIAG [{}] node {} : attempting CAS write - if_seq_no={}, if_primary_term={}, "
+ "writing lockOwner={}",
@@ -391,6 +397,7 @@ public class TaskLockManager {
if (latestOwner == null) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
LOGGER.debug("LOCK-DIAG [{}] node {} : releaseLock() no-op, store already unlocked",
task.getItemId(), nodeId);
return true;
@@ -406,8 +413,10 @@ public class TaskLockManager {
toSave.setLockOwner(null);
toSave.setLockDate(null);
+ toSave.setLockLeaseMillis(0);
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
// Compare-and-set on the freshly loaded seq_no/primary_term, not a blind overwrite:
// a peer may win a legitimate CAS-based lock acquisition in the window between our
@@ -474,6 +483,7 @@ public class TaskLockManager {
}
latest.setLockDate(new Date());
+ latest.setLockLeaseMillis(lockTimeout);
// Compare-and-set on the fresh store view: if a peer stole the lock between the
// read above and this write, renewal fails closed instead of resurrecting our lock.
@@ -486,6 +496,7 @@ public class TaskLockManager {
// the executing thread's later compare-and-set writes are checked against the
// store's current version, not the pre-renewal one.
task.setLockDate(latest.getLockDate());
+ task.setLockLeaseMillis(latest.getLockLeaseMillis());
copyOccMetadata(latest, task);
LOGGER.debug("LOCK-DIAG [{}] node {} : renewLock() succeeded, new lockDate={}",
task.getItemId(), nodeId, latest.getLockDate());
@@ -537,12 +548,22 @@ public class TaskLockManager {
return true;
}
+ // Judge expiry against the lease the OWNER recorded with the lock, not this node's own
+ // configured timeout. The owner renews on a cadence derived from its own timeout
+ // (lockTimeout/3, see TaskExecutionManager#startLockRenewal), so a node configured with a
+ // shorter timeout than the owner's renewal cadence would otherwise declare a live,
+ // renewed lock dead in the gap between two renewals and "recover" a task that is still
+ // executing — observed as double execution under divergent per-node configuration.
+ // Locks written before lease recording carry no lease (0); only for those does this
+ // node's own timeout remain the best available guess.
+ long lease = task.getLockLeaseMillis() > 0 ? task.getLockLeaseMillis() : lockTimeout;
long now = System.currentTimeMillis();
long lockAge = now - task.getLockDate().getTime();
- boolean expired = lockAge > lockTimeout;
+ boolean expired = lockAge > lease;
LOGGER.debug("LOCK-DIAG isLockExpired() : task={}, lockDate={} ({}), now={}, lockAge={}ms, "
- + "lockTimeout={}ms -> expired={}",
- task.getItemId(), task.getLockDate(), task.getLockDate().getTime(), now, lockAge, lockTimeout, expired);
+ + "lease={}ms (recorded={}ms, own timeout={}ms) -> expired={}",
+ task.getItemId(), task.getLockDate(), task.getLockDate().getTime(), now, lockAge,
+ lease, task.getLockLeaseMillis(), lockTimeout, expired);
return expired;
}
}
diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java
index 43b29e368..d41c34c30 100644
--- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java
+++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskRecoveryManager.java
@@ -255,6 +255,7 @@ public class TaskRecoveryManager {
}
latest.setLockOwner(null);
latest.setLockDate(null);
+ latest.setLockLeaseMillis(0);
// Record the crash in execution history
recordCrash(latest, previousOwner);
@@ -273,6 +274,7 @@ public class TaskRecoveryManager {
task.setStatus(latest.getStatus());
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
task.setStatusDetails(latest.getStatusDetails());
task.setCurrentStep(latest.getCurrentStep());
task.setLastError(latest.getLastError());
diff --git a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java
index c8877bd72..d8bf91266 100644
--- a/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java
+++ b/services/src/main/java/org/apache/unomi/services/impl/scheduler/TaskStateManager.java
@@ -148,6 +148,7 @@ public class TaskStateManager {
private void clearTaskExecution(ScheduledTask task) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
task.setWaitingForTaskType(null);
task.setCurrentStep(null);
}
@@ -162,6 +163,7 @@ public class TaskStateManager {
private void clearLockInfo(ScheduledTask task) {
task.setLockOwner(null);
task.setLockDate(null);
+ task.setLockLeaseMillis(0);
}
private void updateRunningState(ScheduledTask task, String nodeId) {
diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java
new file mode 100644
index 000000000..126cee701
--- /dev/null
+++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.unomi.services.impl.scheduler;
+
+import org.apache.unomi.api.Item;
+import org.apache.unomi.api.tasks.ScheduledTask;
+import org.apache.unomi.persistence.spi.CustomObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Date;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Persistence-format coverage for {@link ScheduledTask#getLockLeaseMillis()}.
+ * <p>
+ * The lock lease is a cross-node security decision (it decides who may declare a peer dead), so
+ * its survival through the production serializer is not an implementation detail: a field that
+ * silently fails to round-trip would degrade every observer to the legacy observer-timeout
+ * fallback and quietly reintroduce the divergent-timeout double-execution bug. Both store read
+ * paths are exercised: direct class binding, and the {@code Item}-dispatched path the persistence
+ * services actually use ({@code readValue(json, Item.class)} via {@code ItemDeserializer}).
+ */
+public class ScheduledTaskLeaseSerializationTest {
+
+ private CustomObjectMapper mapper;
+
+ @BeforeEach
+ public void setUp() {
+ mapper = CustomObjectMapper.getCustomInstance();
+ mapper.registerBuiltInItemTypeClass(ScheduledTask.ITEM_TYPE, ScheduledTask.class);
+ }
+
+ private ScheduledTask lockedTask() {
+ ScheduledTask task = new ScheduledTask();
+ task.setItemId("lease-serialization-test");
+ task.setTaskType("lease-serialization-test");
+ task.setStatus(ScheduledTask.TaskStatus.RUNNING);
+ task.setLockOwner("node-a");
+ task.setLockDate(new Date());
+ task.setLockLeaseMillis(12345);
+ return task;
+ }
+
+ @Test
+ public void leaseSurvivesRoundTripViaDirectClassBinding() throws Exception {
+ String json = mapper.writeValueAsString(lockedTask());
+ assertTrue(json.contains("\"lockLeaseMillis\":12345"), "lease must be serialized: " + json);
+
+ ScheduledTask back = mapper.readValue(json, ScheduledTask.class);
+ assertEquals(12345, back.getLockLeaseMillis());
+ assertEquals("node-a", back.getLockOwner());
+ }
+
+ @Test
+ public void leaseSurvivesRoundTripViaItemDispatchedPath() throws Exception {
+ // This is the path the persistence services use when loading store documents.
+ String json = mapper.writeValueAsString(lockedTask());
+ Item item = mapper.readValue(json, Item.class);
+ assertTrue(item instanceof ScheduledTask, "itemType dispatch should yield a ScheduledTask");
+ assertEquals(12345, ((ScheduledTask) item).getLockLeaseMillis());
+ }
+
+ /**
+ * A document written BEFORE lease recording (no {@code lockLeaseMillis} field) must load with
+ * lease 0, which {@code TaskLockManager#isLockExpired} treats as "fall back to the observer's
+ * own timeout" — i.e. exactly the pre-lease behaviour, so a rolling upgrade cannot make old
+ * locks unexpirable or instantly expired.
+ */
+ @Test
+ public void legacyDocumentWithoutLeaseLoadsAsZero() throws Exception {
+ String legacyJson = "{" +
+ "\"itemId\":\"legacy-task\"," +
+ "\"itemType\":\"scheduledTask\"," +
+ "\"taskType\":\"legacy-task\"," +
+ "\"status\":\"RUNNING\"," +
+ "\"lockOwner\":\"old-node\"," +
+ "\"lockDate\":\"2026-01-01T00:00:00Z\"" +
+ "}";
+ Item item = mapper.readValue(legacyJson, Item.class);
+ ScheduledTask task = (ScheduledTask) item;
+ assertEquals(0, task.getLockLeaseMillis(), "missing lease must read as 0 (legacy fallback)");
+ assertEquals("old-node", task.getLockOwner());
+ }
+
+ /**
+ * A document written by a NEWER version carrying a field this version does not know must
+ * still deserialize (rolling upgrade window: older binaries keep reading scheduler state
+ * written by upgraded peers). Pinned by {@code @JsonIgnoreProperties(ignoreUnknown = true)}
+ * on ScheduledTask — without it, Jackson's default rejects the first unknown field and the
+ * older node loses access to every task document the newer node has touched.
+ */
+ @Test
+ public void documentFromNewerVersionWithUnknownFieldStillLoads() throws Exception {
+ String futureJson = "{" +
+ "\"itemId\":\"future-task\"," +
+ "\"itemType\":\"scheduledTask\"," +
+ "\"taskType\":\"future-task\"," +
+ "\"status\":\"SCHEDULED\"," +
+ "\"lockLeaseMillis\":5000," +
+ "\"someFieldAddedInAFutureVersion\":\"whatever\"" +
+ "}";
+ Item item = mapper.readValue(futureJson, Item.class);
+ assertNotNull(item);
+ assertEquals(5000, ((ScheduledTask) item).getLockLeaseMillis());
+ }
+}
diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java
index 62ae062bd..c967de791 100644
--- a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java
+++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceClusterRaceTest.java
@@ -331,6 +331,150 @@ public class SchedulerServiceClusterRaceTest {
assertEquals(1, executors.size(), "Exactly one node should have executed");
}
+ /**
+ * A node configured with a SHORTER lock timeout than a peer must not "recover" that peer's
+ * live, renewed lock.
+ * <p>
+ * The owner renews its lock every {@code lockTimeout/3} — a cadence derived from its OWN
+ * timeout. Before lock leases were recorded ({@link ScheduledTask#getLockLeaseMillis()}),
+ * expiry was judged against the <em>observer's</em> timeout, so an observer whose timeout was
+ * shorter than the owner's renewal cadence saw every renewal gap as an expired lock: it marked
+ * the live execution CRASHED and cleared the lock, and the next peer tick re-dispatched the
+ * task while the original execution was still running. This reproduced deterministically as
+ * {@code maxConcurrent=2} with a 1s-timeout observer against 10s-timeout workers, and is also a
+ * production hazard under config drift or rolling upgrades. The recorded lease makes expiry
+ * owner-relative, so the divergent observer becomes harmless.
+ */
+ @Test
+ public void testShortTimeoutObserverCannotRecoverLiveRenewedLock() throws Exception {
+ SchedulerServiceImpl worker1 = createNode("lease-worker1", true, 10000);
+ SchedulerServiceImpl worker2 = createNode("lease-worker2", true, 10000);
+ // Divergent config: this node judges everything with a 500ms timeout. It registers no
+ // executor for the task type, so any double execution must come via a worker re-dispatch.
+ SchedulerServiceImpl watchdog = createNode("lease-watchdog", true, 500);
+ seedActiveNodes("lease-worker1", "lease-worker2", "lease-watchdog");
+
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger executions = new AtomicInteger(0);
+
+ TaskExecutor executor = new TaskExecutor() {
+ @Override
+ public String getTaskType() {
+ return "lease-liveness-test";
+ }
+
+ @Override
+ public void execute(ScheduledTask task, TaskStatusCallback callback) throws Exception {
+ executions.incrementAndGet();
+ started.countDown();
+ assertTrue(release.await(TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS));
+ callback.complete();
+ }
+ };
+ worker1.registerTaskExecutor(executor);
+ worker2.registerTaskExecutor(executor);
+
+ ScheduledTask task = worker1.newTask("lease-liveness-test")
+ .disallowParallelExecution()
+ .asOneShot()
+ .schedule();
+
+ assertTrue(started.await(TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS), "One worker should start the task");
+
+ // Let the lock age past the watchdog's 500ms timeout while staying far inside the owner's
+ // 10s lease (the owner's renewal cadence is 10s/3, so the age check below cannot be
+ // satisfied by a renewal racing us — any observed age > 600ms is a genuine renewal gap).
+ long deadline = System.currentTimeMillis() + TEST_TIMEOUT_MS;
+ while (System.currentTimeMillis() < deadline) {
+ ScheduledTask stored = persistenceService.load(task.getItemId(), ScheduledTask.class);
+ if (stored != null && stored.getLockDate() != null
+ && System.currentTimeMillis() - stored.getLockDate().getTime() > 600) {
+ break;
+ }
+ Thread.sleep(50);
+ }
+
+ // Force the divergent observer's recovery pass repeatedly — the deterministic version of
+ // the background tick that used to steal the lock.
+ for (int i = 0; i < 3; i++) {
+ watchdog.recoverCrashedTasks();
+ }
+
+ ScheduledTask observed = persistenceService.load(task.getItemId(), ScheduledTask.class);
+ assertEquals(ScheduledTask.TaskStatus.RUNNING, observed.getStatus(),
+ "A live, renewed lock must not be marked CRASHED by a shorter-timeout observer");
+ assertNotNull(observed.getLockOwner(), "The owner's lock must not be cleared");
+
+ release.countDown();
+
+ ScheduledTask done = waitForStatus(worker1, task.getItemId(), ScheduledTask.TaskStatus.COMPLETED, TEST_TIMEOUT_MS);
+ assertEquals(ScheduledTask.TaskStatus.COMPLETED, done.getStatus());
+ assertEquals(1, executions.get(),
+ "The task must execute exactly once despite the divergent-timeout observer");
+ }
+
+ /**
+ * The recovery-enabling direction of lease-based expiry: a genuinely DEAD owner must still be
+ * recovered, and the moment that happens is decided by the lease the dead owner recorded, not
+ * by the survivor's own (here much longer) timeout. This is the guarantee that keeps crash
+ * failover working after the lease change — and it is now faster when the dead node ran with
+ * a short timeout, because peers no longer wait out their own longer opinion.
+ */
+ @Test
+ public void testDeadOwnersShortLeaseDrivesPromptRecoveryByPatientSurvivor() throws Exception {
+ SchedulerServiceImpl survivor = createNode("lease-survivor", true, 30_000);
+ seedActiveNodes("lease-survivor");
+
+ CountDownLatch recovered = new CountDownLatch(1);
+ TaskExecutor executor = new TaskExecutor() {
+ @Override
+ public String getTaskType() {
+ return "dead-owner-lease-test";
+ }
+
+ @Override
+ public void execute(ScheduledTask task, TaskStatusCallback callback) {
+ recovered.countDown();
+ callback.complete();
+ }
+ };
+ survivor.registerTaskExecutor(executor);
+
+ // Manufacture what a crashed node leaves behind: RUNNING, locked, lease recorded from a
+ // short timeout, and silent (no renewal will ever come). lockDate is backdated past the
+ // lease so the very first recovery pass can act.
+ ScheduledTask ghost = new ScheduledTask();
+ ghost.setItemId("ghost-owned-task");
+ ghost.setTaskType("dead-owner-lease-test");
+ ghost.setEnabled(true);
+ ghost.setPersistent(true);
+ ghost.setOneShot(true);
+ ghost.setStatus(ScheduledTask.TaskStatus.RUNNING);
+ ghost.setExecutingNodeId("ghost-node");
+ ghost.setLockOwner("ghost-node");
+ ghost.setLockDate(new Date(System.currentTimeMillis() - 2000));
+ ghost.setLockLeaseMillis(500);
+ persistenceService.save(ghost);
+ persistenceService.refreshIndex(ScheduledTask.class);
+ persistenceService.refresh();
+
+ // Force recovery passes rather than waiting for background ticks. The survivor's own
+ // timeout is 30s: pre-lease it would have refused to touch this lock for 30s, and this
+ // latch (10s) would time out. The recorded 500ms lease is what lets it act now.
+ long deadline = System.currentTimeMillis() + TEST_TIMEOUT_MS;
+ while (recovered.getCount() > 0 && System.currentTimeMillis() < deadline) {
+ survivor.recoverCrashedTasks();
+ recovered.await(250, TimeUnit.MILLISECONDS);
+ }
+
+ assertTrue(recovered.getCount() == 0,
+ "a patient survivor must recover a dead owner's task as soon as the OWNER's lease expires");
+ ScheduledTask done = waitForStatus(survivor, "ghost-owned-task", ScheduledTask.TaskStatus.COMPLETED, TEST_TIMEOUT_MS);
+ assertEquals(ScheduledTask.TaskStatus.COMPLETED, done.getStatus(),
+ "the recovered task must run to completion on the survivor");
+ }
+
@Test
public void testAffinityOpenFieldAfterBackupWindowsWhenPrimaryDead() throws Exception {
SchedulerServiceImpl backup1 = createNode("aff-backup1", true, 10000);
diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java
index 249783884..25fe94bbc 100644
--- a/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java
+++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/TaskLockManagerTest.java
@@ -216,6 +216,188 @@ public class TaskLockManagerTest {
assertFalse(lockManager.isLockExpired(task));
}
+ /**
+ * Expiry must be judged against the lease the OWNER recorded with the lock, not this
+ * observer's own timeout. An observer configured shorter than the owner's renewal cadence
+ * would otherwise "recover" a live, renewed lock between two renewals and double-run the
+ * task (see SchedulerServiceClusterRaceTest#testShortTimeoutObserverCannotRecoverLiveRenewedLock
+ * for the end-to-end version).
+ */
+ @Test
+ public void testIsLockExpiredHonoursRecordedLeaseOverObserverTimeout() {
+ ScheduledTask task = TaskTestFixtures.baseTask("lease");
+ task.setLockDate(new Date(System.currentTimeMillis() - 5000));
+
+ // 5s-old lock, observer timeout 1s: expired by observer maths, but the owner granted 10s.
+ task.setLockLeaseMillis(10000);
+ assertFalse(lockManager.isLockExpired(task),
+ "a lock inside its owner-recorded lease must not expire under a shorter observer timeout");
+
+ // The reverse also holds: an owner that granted itself a SHORT lease is expired even
+ // when the observer's own timeout would still consider it live.
+ task.setLockDate(new Date(System.currentTimeMillis() - 500));
+ task.setLockLeaseMillis(100);
+ assertTrue(lockManager.isLockExpired(task),
+ "a lock past its owner-recorded lease is expired regardless of the observer timeout");
+ }
+
+ /** Locks written before lease recording (lease 0) fall back to the observer's own timeout. */
+ @Test
+ public void testIsLockExpiredFallsBackToObserverTimeoutForLegacyLocks() {
+ ScheduledTask task = TaskTestFixtures.baseTask("legacy");
+ task.setLockDate(new Date(System.currentTimeMillis() - 5000));
+ task.setLockLeaseMillis(0);
+ assertTrue(lockManager.isLockExpired(task));
+
+ task.setLockDate(new Date());
+ assertFalse(lockManager.isLockExpired(task));
+ }
+
+ /** A corrupt negative lease must not wedge or widen the lock: treat it like a legacy lock. */
+ @Test
+ public void testIsLockExpiredNegativeLeaseFallsBackToObserverTimeout() {
+ ScheduledTask task = TaskTestFixtures.baseTask("corrupt");
+ task.setLockDate(new Date(System.currentTimeMillis() - 5000));
+ task.setLockLeaseMillis(-1);
+ assertTrue(lockManager.isLockExpired(task), "negative lease + old lock: observer timeout applies");
+
+ task.setLockDate(new Date());
+ assertFalse(lockManager.isLockExpired(task), "negative lease + fresh lock: observer timeout applies");
+ }
+
+ /** Boundary parity with the observer-timeout path: age == lease is NOT yet expired. */
+ @Test
+ public void testIsLockExpiredFalseWhenAgeEqualsRecordedLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("edge");
+ task.setLockLeaseMillis(2000);
+ task.setLockDate(new Date(System.currentTimeMillis() - 2000));
+ assertFalse(lockManager.isLockExpired(task));
+ }
+
+ /**
+ * An absurd lease (misconfigured or corrupt owner) must not overflow the arithmetic. The lock
+ * is honoured as unexpired — the owner declared it, and stealing it risks double execution;
+ * a genuinely wedged task from a dead misconfigured node is an operator decision, not one a
+ * peer may take unilaterally with a shorter opinion.
+ */
+ @Test
+ public void testIsLockExpiredHugeLeaseIsHonouredWithoutOverflow() {
+ ScheduledTask task = TaskTestFixtures.baseTask("huge");
+ task.setLockLeaseMillis(Long.MAX_VALUE);
+ task.setLockDate(new Date(System.currentTimeMillis() - 100_000));
+ assertFalse(lockManager.isLockExpired(task));
+ }
+
+ // ------------------------------------------------------------------ lease stamping
+ // Every path that writes a lock must record the owner's lease with it, and every path that
+ // clears a lock must clear the lease: a cleared owner with a leftover lease (or the reverse)
+ // would make expiry decisions against a lock that no longer exists.
+
+ @Test
+ public void testParallelAcquireStampsLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("parallel-lease");
+ task.setAllowParallelExecution(true);
+ assertTrue(lockManager.acquireLock(task));
+ assertEquals(1000, task.getLockLeaseMillis(), "parallel marker must record the owner's lease");
+ }
+
+ @Test
+ public void testInMemoryAcquireStampsLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("mem-lease");
+ task.setPersistent(false);
+ assertTrue(lockManager.acquireLock(task));
+ assertEquals(1000, task.getLockLeaseMillis(), "in-memory lock must record the owner's lease");
+ }
+
+ @Test
+ public void testDistributedAcquireStampsLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("dist-lease");
+ task.setNextScheduledExecution(new Date(System.currentTimeMillis() - 10_000));
+ ScheduledTask latest = TaskTestFixtures.baseTask("dist-lease");
+ latest.setItemId(task.getItemId());
+ latest.setSystemMetadata("seq_no", 3L);
+ latest.setSystemMetadata("primary_term", 1L);
+ when(schedulerService.getTask(task.getItemId())).thenReturn(latest);
+ when(schedulerService.saveTaskWithRefresh(any(ScheduledTask.class))).thenReturn(true);
+
+ assertTrue(lockManager.acquireLock(task));
+ assertEquals(1000, task.getLockLeaseMillis(), "distributed lock must record the owner's lease");
+ }
+
+ /**
+ * Renewal re-stamps the lease from the owner's CURRENT timeout, so a runtime configuration
+ * change (ConfigAdmin update) propagates to the store within one renewal interval instead of
+ * peers judging against a stale grant for the rest of the execution.
+ */
+ @Test
+ public void testRenewLockRestampsLeaseFromCurrentTimeout() {
+ ScheduledTask task = TaskTestFixtures.baseTask("renew-lease");
+ task.setLockOwner(NODE);
+ task.setLockDate(new Date());
+ task.setLockLeaseMillis(1000);
+
+ ScheduledTask storeView = TaskTestFixtures.baseTask("renew-lease");
+ storeView.setItemId(task.getItemId());
+ storeView.setLockOwner(NODE);
+ storeView.setLockDate(task.getLockDate());
+ storeView.setLockLeaseMillis(1000);
+ when(schedulerService.getTask(task.getItemId())).thenReturn(storeView);
+ when(schedulerService.saveTaskWithRefresh(storeView)).thenReturn(true);
+
+ lockManager.setLockTimeout(5000);
+ assertTrue(lockManager.renewLock(task));
+ assertEquals(5000, storeView.getLockLeaseMillis(), "store must carry the current lease");
+ assertEquals(5000, task.getLockLeaseMillis(), "caller's view must be synced to the current lease");
+ }
+
+ @Test
+ public void testReleaseLockClearsLease() {
+ ScheduledTask task = TaskTestFixtures.baseTask("release-lease");
+ task.setLockOwner(NODE);
+ task.setLockDate(new Date());
+ task.setLockLeaseMillis(1000);
+
+ ScheduledTask stored = TaskTestFixtures.baseTask("release-lease");
+ stored.setItemId(task.getItemId());
+ stored.setLockOwner(NODE);
+ stored.setLockDate(task.getLockDate());
+ stored.setLockLeaseMillis(1000);
+ when(schedulerService.getTask(eq(task.getItemId()), eq(true))).thenReturn(stored);
+
+ assertTrue(lockManager.releaseLock(task));
+ assertEquals(0, task.getLockLeaseMillis(), "release must clear the caller's lease");
+ assertEquals(0, stored.getLockLeaseMillis(), "release must clear the persisted lease");
+ }
+
+ /**
+ * The recovery-enabling direction: a dead owner that granted itself a SHORT lease is
+ * recoverable by an observer configured with a much longer timeout — the observer must not
+ * impose its own, slower opinion on a lock whose owner promised to renew far sooner.
+ */
+ @Test
+ public void testNonOwnerCanReleaseLockPastItsShortRecordedLease() {
+ lockManager.setLockTimeout(60_000); // observer is very patient by its own config
+
+ ScheduledTask stored = TaskTestFixtures.baseTask("dead-short-lease");
+ stored.setLockOwner("dead-node");
+ stored.setLockDate(new Date(System.currentTimeMillis() - 2000));
+ stored.setLockLeaseMillis(500); // owner promised renewal every ~166ms and is silent for 2s
+
+ ScheduledTask callerView = TaskTestFixtures.baseTask("dead-short-lease");
+ callerView.setItemId(stored.getItemId());
+ callerView.setLockOwner("dead-node");
+ callerView.setLockDate(stored.getLockDate());
+ callerView.setLockLeaseMillis(500);
+ when(schedulerService.getTask(eq(callerView.getItemId()), eq(true))).thenReturn(stored);
+
+ assertTrue(lockManager.isLockExpired(callerView),
+ "a lock silent past its own lease is expired even for a patient observer");
+ assertTrue(lockManager.releaseLock(callerView),
+ "recovery must be able to clear a dead owner's expired-by-lease lock");
+ assertNull(stored.getLockOwner());
+ assertEquals(0, stored.getLockLeaseMillis());
+ }
+
@Test
public void testAffinityBlocksBackupDuringPrimaryWindow() {
List<String> nodes = Arrays.asList("aaa-node", NODE, "zzz-node");
(unomi) 02/02: UNOMI-979: Deflake the scheduler test suites
Posted by sh...@apache.org.
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 <sh...@apache.org>
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) <no...@anthropic.com>
---
.../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));
}