From d3963edfcb50b3a40e479e784f5e30e4b96739c4 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 17 Aug 2026 10:26:21 +0200 Subject: [PATCH 1/8] 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: but was: " 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) --- .../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 ++ .../impl/scheduler/TaskLockManager.java | 29 ++- .../impl/scheduler/TaskRecoveryManager.java | 2 + .../impl/scheduler/TaskStateManager.java | 2 + .../ScheduledTaskLeaseSerializationTest.java | 124 ++++++++++++ .../SchedulerServiceClusterRaceTest.java | 144 ++++++++++++++ .../impl/scheduler/TaskLockManagerTest.java | 182 ++++++++++++++++++ 11 files changed, 535 insertions(+), 5 deletions(-) create mode 100644 services/src/test/java/org/apache/unomi/services/impl/scheduler/ScheduledTaskLeaseSerializationTest.java 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 c5d698e771..5a2f52acef 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 @@ * @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 enum TaskStatus { 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 Date getLockDate() { /** * 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. + *

+ * A lock's lifetime is a lease granted by its owner: 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 f36fc297c2..030305e8e8 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 9c1541d968..a251eebf47 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 3c98963fc3..5e84267dac 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 @@ private void updateTaskState(ScheduledTask task, TaskStatus newStatus, String er 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 @@ private void updateTaskState(ScheduledTask task, TaskStatus newStatus, String er } 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 details = task.getStatusDetails(); @@ -899,6 +901,7 @@ public void preDestroy() { // 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 8f3bd41a48..cde83c97f4 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 @@ private void abortPreparedExecution(ScheduledTask task) { 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 @@ private void startLockRenewal(ScheduledTask task) { 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 @@ private boolean canCommitTerminalTransition(ScheduledTask task) { 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 978d97fa64..fbe1517c07 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 boolean acquireLock(ScheduledTask task) { // 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 @@ private boolean acquireInMemoryLock(ScheduledTask task) { 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 @@ private boolean acquireDistributedLock(ScheduledTask task) { 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 boolean releaseLock(ScheduledTask task) { 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 boolean releaseLock(ScheduledTask task) { 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 boolean renewLock(ScheduledTask task) { } 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 boolean renewLock(ScheduledTask task) { // 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 boolean isLockExpired(ScheduledTask task) { 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 43b29e3685..d41c34c309 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 @@ private void recoverCrashedTask(ScheduledTask task) { } latest.setLockOwner(null); latest.setLockDate(null); + latest.setLockLeaseMillis(0); // Record the crash in execution history recordCrash(latest, previousOwner); @@ -273,6 +274,7 @@ private void recoverCrashedTask(ScheduledTask task) { 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 c8877bd725..d8bf91266c 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 @@ private void updateStateSpecificFields(ScheduledTask task, TaskStatus newStatus, private void clearTaskExecution(ScheduledTask task) { task.setLockOwner(null); task.setLockDate(null); + task.setLockLeaseMillis(0); task.setWaitingForTaskType(null); task.setCurrentStep(null); } @@ -162,6 +163,7 @@ private void preserveCrashState(ScheduledTask task, String nodeId) { 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 0000000000..126cee7013 --- /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()}. + *

+ * 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 62ae062bd8..c967de791f 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 void execute(ScheduledTask task, TaskStatusCallback callback) throws Exce 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. + *

+ * 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 observer's 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 2497838845..25fe94bbc3 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 void testIsLockExpiredFalseWhenAgeEqualsTimeout() { 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 nodes = Arrays.asList("aaa-node", NODE, "zzz-node"); From 4de1679e3643c99d11769b4552466fc3dc0fdde8 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 17 Aug 2026 10:26:21 +0200 Subject: [PATCH 2/8] 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) --- .../scheduler/SchedulerServiceImplTest.java | 277 ++++++++++++++---- .../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 d3e91d56ce..5f9e8ee38a 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 @@ * - RetryTests: Task retry behavior and delay * - MaintenanceTests: Task cleanup and maintenance * - QueryTests: Task querying and filtering + * + *

Debugging

+ * 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. + * + *

Timing rules for this suite

+ * The {@code setUp()} scheduler keeps polling in the background for the whole test, so: + *
    + *
  • 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).
    • + *
  • 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.
    • + *
  • 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.
    • + *
*/ @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 void setUp() throws IOException { 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 void testFixedDelayExecution() throws Exception { .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 void execute(ScheduledTask task, TaskStatusCallback callback) throws Exce @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 void testClusteringSupport() throws Exception { 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 exclusiveNodes = ConcurrentHashMap.newKeySet(); Set allNodesNodes = ConcurrentHashMap.newKeySet(); @@ -496,9 +530,8 @@ public String getTaskType() { @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 void execute(ScheduledTask task, TaskStatusCallback callback) { 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 String getTaskType() { @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 void execute(ScheduledTask task, TaskStatusCallback callback) { } }; - 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. + *

+ * 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 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 void execute(ScheduledTask task, TaskStatusCallback callback) { 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> history = (List>) 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 void execute(ScheduledTask task, TaskStatusCallback callback) { } 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 void execute(ScheduledTask task, TaskStatusCallback callback) throws Exce * 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. + * + *

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 void testLockTimeout() throws Exception { 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 void execute(ScheduledTask task, TaskStatusCallback callback) { 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 void execute(ScheduledTask task, TaskStatusCallback callback) { // 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 boolean canResume(ScheduledTask task) { @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 @@ private static class ExecutionInfo { @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 void execute(ScheduledTask task, TaskStatusCallback callback) { @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 void testNewTaskDefaultsToExclusiveLocking() { @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 void execute(ScheduledTask task, TaskStatusCallback callback) { 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 void execute(ScheduledTask task, TaskStatusCallback callback) { 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 bafcc53565..9c045bf45e 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 void testPrepareForExecutionFailsWhenLockDenied() { 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 void testExecuteTaskDuplicateDispatchIsSkipped() throws Exception { // 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 void testHandleTaskErrorOneShotExhaustsRetriesStaysFailed() throws Except 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 void testHandleTaskErrorPeriodicResetsFailureCountAfterExhaustion() throw 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 void testHandleTaskErrorSkipsRetryScheduleDuringShutdown() throws Excepti 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 void testHandleTaskCompletionOneShotDisablesAndClearsNext() throws Except 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 void testHandleTaskCompletionPeriodicReschedules() throws Exception { 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 void testHandleTaskCompletionPeriodZeroDoesNotReschedule() throws Excepti 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 void testCompletionAndErrorIgnoredWhenNotRunning() throws Exception { 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 void testReclaimPrematureCrashBeforeCompletion() throws Exception { 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 void testTerminalCompleteSkippedWhenCancelledInStore() throws Exception { 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 void testTerminalCompleteSkippedWhenPeerHoldsLock() throws Exception { 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 void testAbortPreparedExecutionOnShutdownReleasesLockAndMarksCrashed() th }; 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)); } From 4d0ba8be5715a7fcf9da36f504ec468a708ae607 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 17 Aug 2026 13:11:11 +0200 Subject: [PATCH 3/8] UNOMI-979: Rebase task counters on the store before a terminal transition CI reported SchedulerServiceImplTest.testMetricsAndHistory failing with "Should have 2 successful executions ==> expected: <2> but was: <1>": a periodic task recorded one success after two successful executions. canCommitTerminalTransition() loads the authoritative document by id and carries only the OCC tokens onto the executing task instance. The counters and execution history stay as the dispatched copy had them - and that copy comes from findEnabledScheduledOrWaitingTasks(), a search query that lags the store by up to the index refresh interval. The compare-and-set in persistTerminalState() then protects only the document version, not those values, so incrementing a stale base and CAS-writing it succeeds and silently discards the newer count. Two successful executions dispatched from the same lagged view therefore both write successCount=1. Accumulators are now rebased on the fresh read before the terminal handler increments them: successCount, failureCount, and the append-only execution history (taken from the store when it is ahead, so other statusDetails keys such as checkpoint and crash markers survive). Status, lock fields and scheduling are untouched - those belong to the execution's own outcome. This is a data-integrity fix independent of the lock-lease change: success and failure counts, and the execution history a UI or operator reads, were under-reported whenever a dispatch raced the refresh interval. It affects Elasticsearch and OpenSearch equally, both having real refresh lag. testTerminalCompletionRebasesCountersOnStoreValues pins it and is mutation-validated: with the rebase removed it fails "expected: <2> but was: <1>", the exact CI symptom, deterministically. Also: testOneShotRetryBehavior's retry-delay assertion now reports the full gap sequence, execution count and persistence mode on failure instead of a bare boolean. It is deliberately not relaxed - an execution landing sooner than the retry delay is a real contract violation. The suspected mechanism is the same staleness family on the dispatch side (prepareForExecution() checks due-ness against the instance it is handed, so a lagged copy still carrying an already past nextScheduledExecution passes the check), but that is unproven and fixing it means changing the dispatch path, so the next occurrence is made decisive rather than silenced. Co-Authored-By: Claude Opus 5 (1M context) --- .../impl/scheduler/TaskExecutionManager.java | 50 ++++++++++++++++ .../scheduler/SchedulerServiceImplTest.java | 32 ++++++++-- .../scheduler/TaskExecutionManagerTest.java | 60 +++++++++++++++++++ 3 files changed, 137 insertions(+), 5 deletions(-) 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 cde83c97f4..98b98a2e30 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 @@ -23,6 +23,8 @@ import java.util.ArrayList; import java.util.Date; +import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.*; @@ -668,9 +670,57 @@ private boolean canCommitTerminalTransition(ScheduledTask task) { // Carry OCC tokens from the fresh load so persistTerminalState can CAS. TaskLockManager.copyOccMetadata(latest, task); + rebaseAccumulatorsFromStore(task, latest); return true; } + /** + * Rebases the running task's accumulating fields on the authoritative store document before a + * terminal handler increments them. + *

+ * The task instance a wrapper carries comes from the dispatch path, whose discovery query + * ({@code findEnabledScheduledOrWaitingTasks}) is search-based and therefore lags the store by + * up to the index refresh interval. Its {@code successCount}, {@code failureCount} and + * execution history can predate writes that have already landed. The compare-and-set in + * {@link #persistTerminalState} protects only the document version, not these values: + * incrementing a stale base and then CAS-writing it succeeds and silently loses the newer + * count. Observed as a periodic task reporting one success after two successful executions. + *

+ * Only accumulators are taken from the store. Status, lock fields and scheduling are the + * terminal handler's business and are set from the execution's own outcome. + * + * @param task the executing task instance about to be mutated by a terminal handler + * @param latest the authoritative document, freshly loaded by id + */ + private static void rebaseAccumulatorsFromStore(ScheduledTask task, ScheduledTask latest) { + task.setSuccessCount(latest.getSuccessCount()); + task.setFailureCount(latest.getFailureCount()); + + // Execution history is append-only, so the longer list is the more current one. Other + // statusDetails keys stay as the execution left them (checkpoint markers, crash details). + Map latestDetails = latest.getStatusDetails(); + if (latestDetails == null) { + return; + } + Object latestHistory = latestDetails.get("executionHistory"); + if (!(latestHistory instanceof List)) { + return; + } + Map details = task.getStatusDetails(); + if (details == null) { + details = new HashMap<>(); + task.setStatusDetails(details); + } else if (!(details instanceof HashMap)) { + details = new HashMap<>(details); + task.setStatusDetails(details); + } + Object ourHistory = details.get("executionHistory"); + int ourSize = ourHistory instanceof List ? ((List) ourHistory).size() : 0; + if (((List) latestHistory).size() > ourSize) { + details.put("executionHistory", new ArrayList<>((List) latestHistory)); + } + } + /** * Persists a terminal task state. Persistent tasks use compare-and-set so a late * complete/fail cannot clobber CANCELLED or a peer's RUNNING document. Lock fields are 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 5f9e8ee38a..1be4eea88e 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 @@ -1050,12 +1050,34 @@ public void execute(ScheduledTask task, TaskStatusCallback callback) { executionLatch.await(TEST_TIMEOUT, TimeUnit.MILLISECONDS), "Task should complete all executions"); - // Verify retry delays + // Verify retry delays. Deliberately NOT relaxed: an execution landing sooner than the + // retry delay means a retry attempt was dispatched early, which is a real contract + // violation worth failing on. Suspected mechanism if this fires on CI and not locally: + // prepareForExecution() checks due-ness against the task instance it was handed, and the + // checker discovers tasks with a search query that lags the store, so a stale copy still + // carrying the pre-retry (already past) nextScheduledExecution passes the due check and + // executes immediately. Unproven - hence the diagnostics below rather than a weakened + // assertion, so the next occurrence is decisive instead of just a boolean. for (int i = 1; i < executionTimes.size(); i++) { - long delay = executionTimes.get(i) - executionTimes.get(i-1); - assertTrue( - delay >= TEST_RETRY_DELAY, - "Retry delay should be at least " + TEST_RETRY_DELAY + "ms"); + long delay = executionTimes.get(i) - executionTimes.get(i - 1); + if (delay < TEST_RETRY_DELAY) { + StringBuilder detail = new StringBuilder(); + detail.append("Retry delay should be at least ").append(TEST_RETRY_DELAY) + .append("ms but execution #").append(i + 1).append(" came ").append(delay) + .append("ms after #").append(i) + .append(". persistent=").append(persistent) + .append(", executions=").append(executionTimes.size()) + .append(" (expected ").append(TEST_MAX_RETRIES + 1).append("), gaps=["); + for (int j = 1; j < executionTimes.size(); j++) { + if (j > 1) { + detail.append(", "); + } + detail.append(executionTimes.get(j) - executionTimes.get(j - 1)).append("ms"); + } + detail.append("]. More executions than expected points at a duplicate dispatch; " + + "the right count with a short gap points at an early retry schedule."); + fail(detail.toString()); + } } // Wait for the task to transition from RUNNING to COMPLETED state 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 9c045bf45e..f81b9c6e47 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 @@ -28,8 +28,12 @@ import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; +import java.util.ArrayList; import java.util.Collections; import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -91,6 +95,62 @@ public void tearDown() { executionManager.shutdown(); } + /** + * A terminal handler must increment counters from the STORE's values, not from the possibly + * stale copy the wrapper is carrying. + *

+ * The dispatch path discovers tasks with a search query, which lags the store by up to the + * index refresh interval, so the executing instance can hold counters that predate writes + * already committed. {@code persistTerminalState}'s compare-and-set protects only the document + * version, so incrementing a stale base then CAS-writing it succeeds and silently loses the + * newer count. Observed in CI as a periodic task reporting one success after two successful + * executions ({@code SchedulerServiceImplTest.testMetricsAndHistory}). + */ + @Test + public void testTerminalCompletionRebasesCountersOnStoreValues() throws Exception { + CountDownLatch done = new CountDownLatch(1); + TaskExecutor executor = new TaskExecutor() { + @Override public String getTaskType() { return "stale-counters"; } + @Override public void execute(ScheduledTask task, TaskStatusCallback callback) { + callback.complete(); + done.countDown(); + } + }; + + // What the wrapper carries: a search-lagged view that has not seen the first success. + ScheduledTask stale = TaskTestFixtures.baseTask("stale-counters"); + stale.setOneShot(false); + stale.setPeriod(60_000); + stale.setSuccessCount(0); + stale.setFailureCount(0); + + // What the store actually holds: one success already recorded, with its history entry. + ScheduledTask store = TaskTestFixtures.baseTask("stale-counters"); + store.setItemId(stale.getItemId()); + store.setStatus(ScheduledTask.TaskStatus.RUNNING); + store.setExecutingNodeId(NODE); + store.setSuccessCount(1); + Map storeDetails = new HashMap<>(); + List> storeHistory = new ArrayList<>(); + storeHistory.add(Collections.singletonMap("status", "SUCCESS")); + storeDetails.put("executionHistory", storeHistory); + store.setStatusDetails(storeDetails); + when(schedulerService.getTask(eq(stale.getItemId()), eq(true))).thenReturn(store); + + executionManager.executeTask(stale, executor); + assertTrue(done.await(5, TimeUnit.SECONDS)); + awaitStatus(stale, ScheduledTask.TaskStatus.SCHEDULED, 5000); + + assertEquals(2, stale.getSuccessCount(), + "the second success must count from the store's value (1), not the stale copy's (0)"); + + @SuppressWarnings("unchecked") + List> history = + (List>) stale.getStatusDetails().get("executionHistory"); + assertEquals(2, history.size(), + "history must extend the store's entries rather than restart from the stale copy's"); + } + @Test public void testPrepareForExecutionRejectsDisabledAndWrongStatus() { ScheduledTask disabled = TaskTestFixtures.baseTask("p"); From 07d497569fe5d17abe226586cb948a8059355eb3 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 17 Aug 2026 14:13:23 +0200 Subject: [PATCH 4/8] UNOMI-979: Dump scheduler state and DEBUG trace when a scheduler test fails Every intermittent scheduler failure on CI so far has arrived as a bare assertion message. The evidence needed to diagnose one - which node held a lock, when it was renewed, who judged it expired, what the store actually contained - is already logged as LOCK-DIAG lines, but only at DEBUG, and CI does not run at DEBUG. Re-running with -DTEST_LOG_LEVEL=DEBUG rarely helps because an intermittent failure usually does not recur on demand. Both failures in the previous CI run had to be diagnosed by reading code, and neither could be reproduced locally even at matching core count under load. SchedulerDiagnosticsExtension captures DEBUG for the scheduler and cluster packages into a bounded in-memory ring buffer and dumps it, together with a snapshot of every task document in the store, at the moment a test fails: * The snapshot runs from TestExecutionExceptionHandler, which fires before the test's own @AfterEach, so it shows the state that caused the failure rather than what teardown left behind. It refreshes the index first, since the query is search-based and would otherwise report "(none)" purely because of refresh lag. Per task it reports status, enabled, executing node, lock owner, lock date, lease, success and failure counts, next execution, history size and last error - the fields these failures actually turn on. * Capture costs nothing on a passing test: the ring buffer is the only consumer of the DEBUG events, so nothing reaches the console unless a test fails. Verified both ways - a passing run prints zero LOCK-DIAG lines, and a failing one prints the full trace. * When -DTEST_LOG_LEVEL is set the extension stands aside entirely, so an explicit request for console output still gets console output. Wired into the four scheduler test classes with one annotation each. The extension finds the persistence service by reflection so adding it to a test class needs no other change. Note for anyone touching these tests: mvn compile and test-compile in this module report BUILD SUCCESS while skipping changed sources, and produced a class file with an unqualified annotation descriptor here (TypeNotPresentException at runtime). Use mvn clean when verifying a compile. Co-Authored-By: Claude Opus 5 (1M context) --- .../SchedulerDiagnosticsExtension.java | 271 ++++++++++++++++++ .../SchedulerServiceClusterRaceTest.java | 1 + .../scheduler/SchedulerServiceImplTest.java | 1 + .../scheduler/TaskExecutionManagerTest.java | 1 + .../impl/scheduler/TaskLockManagerTest.java | 1 + 5 files changed, 275 insertions(+) create mode 100644 services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerDiagnosticsExtension.java diff --git a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerDiagnosticsExtension.java b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerDiagnosticsExtension.java new file mode 100644 index 0000000000..73bf1eb5d3 --- /dev/null +++ b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerDiagnosticsExtension.java @@ -0,0 +1,271 @@ +/* + * 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 ch.qos.logback.classic.Level; +import ch.qos.logback.classic.LoggerContext; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.CyclicBufferAppender; +import org.apache.unomi.api.tasks.ScheduledTask; +import org.apache.unomi.persistence.spi.PersistenceService; +import org.junit.jupiter.api.extension.AfterEachCallback; +import org.junit.jupiter.api.extension.BeforeEachCallback; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.junit.jupiter.api.extension.TestExecutionExceptionHandler; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Field; +import java.text.SimpleDateFormat; +import java.util.Collection; +import java.util.Date; +import java.util.List; +import java.util.Map; + +/** + * Makes an intermittent scheduler test failure diagnosable from the CI log it failed in. + *

+ * Scheduler failures are almost always about state and ordering: which node held a lock, when it + * was renewed, who decided it had expired, what the store actually contained. The scheduler already + * logs all of that as {@code LOCK-DIAG} lines, but only at DEBUG, and CI does not run at DEBUG -- + * so every intermittent failure historically arrived as a bare assertion message with the evidence + * discarded. Re-running with {@code -DTEST_LOG_LEVEL=DEBUG} rarely helps, because an intermittent + * failure usually does not recur on demand. + *

+ * This extension therefore captures DEBUG for the scheduler packages into a bounded in-memory ring + * buffer that costs nothing on a passing test, and dumps it -- together with a snapshot of every + * task document in the store -- at the moment a test fails. The snapshot is taken from + * {@link TestExecutionExceptionHandler}, which runs before the test's own {@code @AfterEach} + * teardown, so the store is still alive and holds the state that caused the failure rather than + * whatever cleanup left behind. + *

+ * When {@code -DTEST_LOG_LEVEL} is set explicitly, the extension stays out of the way and leaves + * logback's configured behaviour alone: an explicit request for console output should get console + * output. + */ +public class SchedulerDiagnosticsExtension + implements BeforeEachCallback, AfterEachCallback, TestExecutionExceptionHandler { + + /** Enough lines to cover several checker ticks across a handful of nodes. */ + private static final int BUFFER_SIZE = 4000; + + /** Packages whose DEBUG output explains scheduler behaviour. */ + private static final String[] CAPTURED_LOGGERS = { + "org.apache.unomi.services.impl.scheduler", + "org.apache.unomi.services.impl.cluster" + }; + + private static final String APPENDER_NAME = "scheduler-diagnostics-ring-buffer"; + private static final ExtensionContext.Namespace NAMESPACE = + ExtensionContext.Namespace.create(SchedulerDiagnosticsExtension.class); + + private static boolean explicitLogLevelRequested() { + String requested = System.getProperty("TEST_LOG_LEVEL"); + return requested != null && !requested.trim().isEmpty(); + } + + @Override + public void beforeEach(ExtensionContext context) { + if (explicitLogLevelRequested()) { + return; + } + if (!(LoggerFactory.getILoggerFactory() instanceof LoggerContext)) { + return; // not logback (shaded/OSGi runs); nothing to attach to + } + LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); + + CyclicBufferAppender buffer = new CyclicBufferAppender<>(); + buffer.setContext(loggerContext); + buffer.setName(APPENDER_NAME); + buffer.setMaxSize(BUFFER_SIZE); + buffer.start(); + + for (String name : CAPTURED_LOGGERS) { + ch.qos.logback.classic.Logger logger = loggerContext.getLogger(name); + // additive=false keeps the captured DEBUG out of the console on passing runs; the + // buffer dump below is the only consumer, and it only fires on failure. + logger.setLevel(Level.DEBUG); + logger.setAdditive(false); + logger.addAppender(buffer); + } + context.getStore(NAMESPACE).put(APPENDER_NAME, buffer); + } + + @Override + public void afterEach(ExtensionContext context) { + @SuppressWarnings("unchecked") + CyclicBufferAppender buffer = + context.getStore(NAMESPACE).remove(APPENDER_NAME, CyclicBufferAppender.class); + if (buffer == null || !(LoggerFactory.getILoggerFactory() instanceof LoggerContext)) { + return; + } + LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); + for (String name : CAPTURED_LOGGERS) { + ch.qos.logback.classic.Logger logger = loggerContext.getLogger(name); + logger.detachAppender(buffer); + logger.setAdditive(true); + logger.setLevel(null); // inherit from root again + } + buffer.stop(); + } + + @Override + public void handleTestExecutionException(ExtensionContext context, Throwable throwable) + throws Throwable { + StringBuilder report = new StringBuilder(); + report.append("\n================ SCHEDULER DIAGNOSTICS for ") + .append(context.getRequiredTestClass().getSimpleName()).append('.') + .append(context.getRequiredTestMethod().getName()) + .append(" ================\n") + .append("Failure: ").append(throwable).append('\n'); + + appendTaskSnapshot(report, context); + appendBufferedLog(report, context); + + report.append("================ END SCHEDULER DIAGNOSTICS ================\n"); + // stdout, not a logger: this must survive whatever logging configuration is in force, and + // Surefire captures stdout into the report the CI log shows. + System.out.println(report); + + throw throwable; + } + + /** + * Dumps every task document the test's persistence service can see. Taken before teardown, so + * this is the state that produced the failure. + */ + private void appendTaskSnapshot(StringBuilder report, ExtensionContext context) { + report.append("\n-- task documents in the store at failure time --\n"); + PersistenceService persistenceService = findPersistenceService(context); + if (persistenceService == null) { + report.append(" (no PersistenceService field found on the test instance)\n"); + return; + } + try { + // getAllItems is search-based, and both the in-memory harness and a real cluster hold a + // refresh interval behind the store. Force visibility first: the test has already + // failed, so there is no state left worth preserving, and a snapshot that silently + // reports "(none)" because of refresh lag is worse than useless. + try { + persistenceService.refreshIndex(ScheduledTask.class); + persistenceService.refresh(); + } catch (Exception ignored) { + report.append(" (refresh before snapshot failed; list may lag the store)\n"); + } + List tasks = + persistenceService.getAllItems(ScheduledTask.class, 0, -1, null).getList(); + if (tasks.isEmpty()) { + report.append(" (none)\n"); + return; + } + SimpleDateFormat fmt = new SimpleDateFormat("HH:mm:ss.SSS"); + for (ScheduledTask task : tasks) { + report.append(" ").append(task.getItemId()) + .append(" type=").append(task.getTaskType()) + .append(" status=").append(task.getStatus()) + .append(" enabled=").append(task.isEnabled()) + .append(" execNode=").append(task.getExecutingNodeId()) + .append(" lockOwner=").append(task.getLockOwner()) + .append(" lockDate=") + .append(task.getLockDate() == null ? "null" : fmt.format(task.getLockDate())) + .append(" lease=").append(task.getLockLeaseMillis()).append("ms") + .append(" success=").append(task.getSuccessCount()) + .append(" failure=").append(task.getFailureCount()) + .append(" nextExec=") + .append(task.getNextScheduledExecution() == null + ? "null" : fmt.format(task.getNextScheduledExecution())) + .append(" history=").append(historySize(task)) + .append(" lastError=").append(task.getLastError()) + .append('\n'); + } + } catch (Exception e) { + report.append(" (failed to read tasks: ").append(e).append(")\n"); + } + } + + private static int historySize(ScheduledTask task) { + Map details = task.getStatusDetails(); + if (details == null) { + return 0; + } + Object history = details.get("executionHistory"); + return history instanceof Collection ? ((Collection) history).size() : 0; + } + + /** + * Finds a {@link PersistenceService} on the test instance. Reflection rather than an interface + * the tests must implement: the point is that adding this extension to a test class costs one + * annotation and no other change. + */ + private PersistenceService findPersistenceService(ExtensionContext context) { + Object testInstance = context.getTestInstance().orElse(null); + if (testInstance == null) { + return null; + } + for (Class type = testInstance.getClass(); type != null; type = type.getSuperclass()) { + for (Field field : type.getDeclaredFields()) { + if (!PersistenceService.class.isAssignableFrom(field.getType())) { + continue; + } + try { + field.setAccessible(true); + PersistenceService value = (PersistenceService) field.get(testInstance); + if (value != null) { + return value; + } + } catch (ReflectiveOperationException | RuntimeException ignored) { + // Not readable; keep looking. + } + } + } + return null; + } + + private void appendBufferedLog(StringBuilder report, ExtensionContext context) { + @SuppressWarnings("unchecked") + CyclicBufferAppender buffer = + context.getStore(NAMESPACE).get(APPENDER_NAME, CyclicBufferAppender.class); + if (buffer == null) { + report.append("\n-- captured scheduler DEBUG log --\n") + .append(" (not captured; -DTEST_LOG_LEVEL was set, so the log went to the console)\n"); + return; + } + int count = buffer.getLength(); + report.append("\n-- captured scheduler DEBUG log (last ").append(count) + .append(" events, newest last) --\n"); + if (count == 0) { + report.append(" (empty)\n"); + return; + } + SimpleDateFormat fmt = new SimpleDateFormat("HH:mm:ss.SSS"); + for (int i = 0; i < count; i++) { + ILoggingEvent event = buffer.get(i); + if (event == null) { + continue; + } + report.append(" ").append(fmt.format(new Date(event.getTimeStamp()))) + .append(" [").append(event.getThreadName()).append("] ") + .append(event.getLevel()).append(' ') + .append(shortLoggerName(event.getLoggerName())).append(" - ") + .append(event.getFormattedMessage()).append('\n'); + } + } + + private static String shortLoggerName(String loggerName) { + int lastDot = loggerName.lastIndexOf('.'); + return lastDot < 0 ? loggerName : loggerName.substring(lastDot + 1); + } +} 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 c967de791f..d2ce88f7f6 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 @@ -75,6 +75,7 @@ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) @Tag("ClusterTests") +@ExtendWith(SchedulerDiagnosticsExtension.class) public class SchedulerServiceClusterRaceTest { private static final Logger LOGGER = LoggerFactory.getLogger(SchedulerServiceClusterRaceTest.class); 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 1be4eea88e..0521f476bd 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 @@ -97,6 +97,7 @@ */ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) +@ExtendWith(SchedulerDiagnosticsExtension.class) public class SchedulerServiceImplTest { private static final Logger LOGGER = LoggerFactory.getLogger(SchedulerServiceImplTest.class); 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 f81b9c6e47..8aa3105b04 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 @@ -49,6 +49,7 @@ */ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) +@ExtendWith(SchedulerDiagnosticsExtension.class) public class TaskExecutionManagerTest { private static final String NODE = "exec-node"; 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 25fe94bbc3..57cec8d58e 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 @@ -44,6 +44,7 @@ */ @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) +@ExtendWith(SchedulerDiagnosticsExtension.class) public class TaskLockManagerTest { private static final String NODE = "lock-node"; From 753f084e0aaad3930dd1967b976d53647bdab719 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 17 Aug 2026 14:21:47 +0200 Subject: [PATCH 5/8] UNOMI-979: Report flaky unit tests and archive reports on CI Intermittent unit-test failures were invisible in two directions. A test that failed once and would have passed on a retry turned the whole build red, and a test that already passed on someone's manual re-run left no record at all - so nobody could tell which suites were unreliable, or whether a deflaking change had worked. Both of the scheduler failures investigated under this ticket had to be diagnosed from a bare assertion message. The unit-test job now: * Retries a failing test twice (-Dsurefire.rerunFailingTestsCount=2, passed via the existing MAVEN_EXTRA_OPTS hook). This runner has 2 vCPU and several suites are timing-sensitive, so one unlucky scheduling hiccup should not fail a build. * Surfaces every flake in the job summary as a table of test, retry count and first failure message. This is the half that keeps the retry honest: a test passing only on retry is a real intermittent failure, and without the summary the retry would simply convert a red build into an invisible green one. * Archives surefire reports when the build failed or anything flaked - the runs where the reports, including the scheduler state and DEBUG traces the new SchedulerDiagnosticsExtension writes into them, are worth keeping. Skipped on clean runs so artifacts do not accumulate on every push. * Publishes a JUnit check for unit tests, mirroring what the integration-test job already does, so per-test results are visible without opening the log. Job timeout raised 15 -> 20 minutes: retried failures add time on a red build, and a timeout is a worse signal than a clean failure. Verified by running a test rigged to fail on its first attempt only: Surefire reports "Flakes: 1", the build goes green, flakyFailure lands in the XML, and the detection script picks it up and writes the summary table. The script was also checked against synthetic reports to confirm it flags only the flaky case. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/unomi-ci-build-tests.yml | 71 +++++++++++++++++++++- 1 file changed, 70 insertions(+), 1 deletion(-) diff --git a/.github/workflows/unomi-ci-build-tests.yml b/.github/workflows/unomi-ci-build-tests.yml index 076ab448e2..ebb283fef0 100644 --- a/.github/workflows/unomi-ci-build-tests.yml +++ b/.github/workflows/unomi-ci-build-tests.yml @@ -21,7 +21,9 @@ jobs: unit-tests: name: Execute unit tests runs-on: ubuntu-latest - timeout-minutes: 15 + # Slightly above the previous 15: retried failures (rerunFailingTestsCount below) add time + # on a red build, and a timeout is a much worse signal than a clean failure. + timeout-minutes: 20 steps: - uses: actions/checkout@v5 - name: Set up JDK 17 @@ -36,12 +38,79 @@ jobs: sudo apt-get install -y graphviz dot -V - name: Build and Unit tests + env: + # Retry a failing test twice before calling the build red. Several suites (notably the + # scheduler ones) are timing-sensitive and this runner has 2 vCPU, so a single unlucky + # scheduling hiccup should not fail a whole build. A test that only passes on retry is + # NOT silently forgiven: Surefire records it as a flake, and the step below surfaces + # every one in the job summary so the flake rate stays visible instead of becoming + # invisible green. + MAVEN_EXTRA_OPTS: -Dsurefire.rerunFailingTestsCount=2 run: ./build.sh --ci # Keep only third-party dependencies in the post-job Maven cache: Unomi's own # snapshots are rebuilt every run and would only bloat the cache / risk staleness - name: Clean Unomi artifacts from Maven cache if: always() run: rm -rf ~/.m2/repository/org/apache/unomi + # A flake is a test that failed and then passed on retry. The build is green, so without + # this the signal is lost entirely — which is how the scheduler suites stayed unreliable + # for as long as they did. + - name: Detect flaky tests + id: flakes + if: always() + run: | + python3 - <<'PY' >> "$GITHUB_STEP_SUMMARY" + import glob, os, xml.etree.ElementTree as ET + flaky = [] + for path in glob.glob('**/target/surefire-reports/TEST-*.xml', recursive=True): + try: + root = ET.parse(path).getroot() + except ET.ParseError: + continue + for case in root.iter('testcase'): + reruns = case.findall('flakyFailure') + case.findall('flakyError') + if reruns: + msg = (reruns[0].get('message') or '').strip().replace('\n', ' ') + flaky.append((case.get('classname', '?'), case.get('name', '?'), + len(reruns), msg[:160])) + if flaky: + print('### :warning: Flaky tests detected\n') + print('These failed and then passed on retry. The build is green, but each one is') + print('a real intermittent failure worth investigating.\n') + print('| Test | Retries | First failure |') + print('| --- | --- | --- |') + for cls, name, n, msg in sorted(flaky): + print(f'| `{cls}.{name}` | {n} | {msg or "—"} |') + else: + print('### No flaky tests detected\n') + with open(os.environ['GITHUB_OUTPUT'], 'a') as out: + out.write(f'found={"true" if flaky else "false"}\n') + out.write(f'count={len(flaky)}\n') + PY + # Uploaded when the build failed OR when something only passed on retry: those are exactly + # the runs where the reports (and the scheduler diagnostics dumped into them) are worth + # keeping. Skipped on a clean green run so this does not accumulate on every push. + - name: Archive unit test reports + uses: actions/upload-artifact@v6 + if: always() && (job.status == 'failure' || steps.flakes.outputs.found == 'true') + with: + name: unit-test-reports-jdk17-${{ github.run_number }} + path: | + **/target/surefire-reports/** + if-no-files-found: ignore + retention-days: 14 + # Always publish so a later "re-run failed jobs" pass updates the check to green, matching + # the integration-test job's behaviour. + - name: Publish Test Report + uses: mikepenz/action-junit-report@v3 + if: always() + continue-on-error: true + with: + report_paths: '**/target/surefire-reports/TEST-*.xml' + check_name: 'JUnit Test Report (unit tests)' + update_check: true + fail_on_failure: false + require_tests: false integration-tests: name: Execute integration tests From 3ed33ce9805c9faf9e2bff53c73fa9335ecfa02f Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 17 Aug 2026 20:16:41 +0200 Subject: [PATCH 6/8] UNOMI-979: Skip unit tests in the integration-test CI jobs The unit-test job already runs the whole unit suite on the commit, and then both integration-test legs ran it again as part of their build before reaching the integration tests they exist for -- three executions of the same suite per CI cycle, two of them with no signal the first did not already give. Both legs now pass --skip-unit-tests, an option build.sh already supports and documents alongside --integration-tests. It activates the skip-unit-tests profile, which sets maven-surefire-plugin's skip only: maven-failsafe-plugin is untouched, so the integration tests themselves still run. Measured on the full reactor (same 60 modules, same stop point, only surefire differing): 215.6s with unit tests versus 48.8s without, so roughly 2.8 minutes of unit-test execution -- including the 806-test services module -- on a 16-core machine. The CI runners have 2 vCPU, so the saving there is larger; the unit-test job spends 10 minutes on build plus tests in total. Unit-test regressions still fail the build: they fail it in the unit-test job, on the same commit, which is where that signal belongs. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/unomi-ci-build-tests.yml | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/.github/workflows/unomi-ci-build-tests.yml b/.github/workflows/unomi-ci-build-tests.yml index ebb283fef0..febf15c73f 100644 --- a/.github/workflows/unomi-ci-build-tests.yml +++ b/.github/workflows/unomi-ci-build-tests.yml @@ -143,11 +143,15 @@ jobs: MAVEN_EXTRA_OPTS: >- -Dopensearch.port=${{ matrix.port }} -Delasticsearch.port=${{ matrix.port }} + # --skip-unit-tests: the unit-tests job above already ran them on this same commit, and + # both IT legs would otherwise run the whole suite a second and third time before getting + # to the integration tests they exist for. The flag activates the skip-unit-tests profile, + # which sets surefire's skip only -- failsafe, and therefore the ITs, still run. run: | if [ "${{ matrix.search-engine }}" = "opensearch" ]; then - ./build.sh --ci --integration-tests --use-opensearch + ./build.sh --ci --integration-tests --skip-unit-tests --use-opensearch else - ./build.sh --ci --integration-tests + ./build.sh --ci --integration-tests --skip-unit-tests fi # Keep only third-party dependencies in the post-job Maven cache: Unomi's own # snapshots are rebuilt every run and would only bloat the cache / risk staleness From 8d3fe677c947dc0d97a5e4d875cc9031c38619a5 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 17 Aug 2026 20:24:53 +0200 Subject: [PATCH 7/8] UNOMI-979: Skip Javadoc validation in the integration-test CI jobs --ci turns on Javadoc validation, which adds two extra full-reactor Maven invocations after the build: javadoc:javadoc and javadoc-tags-warn checkstyle:check. Measured on CI they cost 1.5 and 0.3 minutes. The integration-test job is gated on `needs: unit-tests`, so both have already passed on the same commit by the time it starts, and with max-parallel: 1 the two legs pay for the duplication one after the other. Adds --no-javadoc to build.sh and uses it in both legs. The flag is applied as an explicit veto after argument parsing, so it wins over --ci regardless of the order the two appear in; verified for --ci --no-javadoc and --no-javadoc --ci. Together with the unit-test skip in the previous commit, the CI breakdown for the 10-minute unit-test step was: mvn clean 0.1, mvn install (build + unit tests) 7.6, javadoc 1.5, checkstyle 0.3. Summed surefire execution across the reactor was 3.0 minutes. Each IT leg should therefore drop roughly 5 minutes, and since the legs run sequentially that is about 10 minutes off the total. Javadoc regressions still fail the build: they fail it in the unit-test job, where doclint errors are meant to be caught. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/unomi-ci-build-tests.yml | 16 ++++++++++------ build.sh | 14 ++++++++++++++ 2 files changed, 24 insertions(+), 6 deletions(-) diff --git a/.github/workflows/unomi-ci-build-tests.yml b/.github/workflows/unomi-ci-build-tests.yml index febf15c73f..07f2f58583 100644 --- a/.github/workflows/unomi-ci-build-tests.yml +++ b/.github/workflows/unomi-ci-build-tests.yml @@ -143,15 +143,19 @@ jobs: MAVEN_EXTRA_OPTS: >- -Dopensearch.port=${{ matrix.port }} -Delasticsearch.port=${{ matrix.port }} - # --skip-unit-tests: the unit-tests job above already ran them on this same commit, and - # both IT legs would otherwise run the whole suite a second and third time before getting - # to the integration tests they exist for. The flag activates the skip-unit-tests profile, - # which sets surefire's skip only -- failsafe, and therefore the ITs, still run. + # This job is gated on `needs: unit-tests`, so the unit suite and the Javadoc/checkstyle + # validation have already passed on this exact commit. Re-running either here is pure + # duplication before the integration tests this job exists for, and the legs run + # sequentially (max-parallel: 1), so it costs twice over. + # --skip-unit-tests activates the skip-unit-tests profile, which sets surefire's skip + # only: failsafe, and therefore the ITs, still run. + # --no-javadoc drops the two extra full-reactor invocations --ci adds + # (javadoc:javadoc and javadoc-tags-warn checkstyle:check). run: | if [ "${{ matrix.search-engine }}" = "opensearch" ]; then - ./build.sh --ci --integration-tests --skip-unit-tests --use-opensearch + ./build.sh --ci --integration-tests --skip-unit-tests --no-javadoc --use-opensearch else - ./build.sh --ci --integration-tests --skip-unit-tests + ./build.sh --ci --integration-tests --skip-unit-tests --no-javadoc fi # Keep only third-party dependencies in the post-job Maven cache: Unomi's own # snapshots are rebuilt every run and would only bloat the cache / risk staleness diff --git a/build.sh b/build.sh index c74fd5eb80..6c93ecb710 100755 --- a/build.sh +++ b/build.sh @@ -281,6 +281,7 @@ IT_SEARCH_ENGINE_LOGS=false IT_MEMORY_SAMPLER=true IT_MEMORY_INTERVAL=30 JAVADOC=false +NO_JAVADOC=false LOG_FILE="" LOG_FILE_ONLY=false @@ -329,6 +330,7 @@ EOF echo -e " ${CYAN}--no-memory-sampler${NC} Disable JVM/system memory sampling during integration tests" echo -e " ${CYAN}--memory-interval SEC${NC} Memory sample interval in seconds (default: 30)" echo -e " ${CYAN}--javadoc${NC} Build and validate Javadoc after install (doclint errors fail; public/protected tag gaps warn)" + echo -e " ${CYAN}--no-javadoc${NC} Skip Javadoc/checkstyle validation (overrides --ci; use when another job already ran it)" echo -e " ${CYAN}--ci${NC} CI mode: no Karaf, non-interactive, includes Javadoc" echo -e " ${CYAN}--log-file PATH${NC} Tee all output to PATH (console + file)" echo -e " ${CYAN}--log-file-only${NC} With --log-file: write to file only, suppress console" @@ -373,6 +375,7 @@ EOF echo " --no-memory-sampler Disable JVM/system memory sampling during integration tests" echo " --memory-interval SEC Memory sample interval in seconds (default: 30)" echo " --javadoc Build and validate Javadoc after install (doclint errors fail; public/protected tag gaps warn)" + echo " --no-javadoc Skip Javadoc/checkstyle validation (overrides --ci; use when another job already ran it)" echo " --ci CI mode: no Karaf, non-interactive, includes Javadoc" echo " --log-file PATH Tee all output to PATH (console + file)" echo " --log-file-only With --log-file: write to file only, suppress console" @@ -549,6 +552,11 @@ while [ "$1" != "" ]; do --javadoc) JAVADOC=true ;; + --no-javadoc) + # Explicit veto, applied after argument parsing so it wins regardless of whether it + # appears before or after --ci (which turns Javadoc on). + NO_JAVADOC=true + ;; --log-file) shift LOG_FILE="$1" @@ -1167,6 +1175,12 @@ echo "Estimated time: 3-5 minutes for build, 50-60 minutes with integration test start_timer # Build phases with enhanced output +# Apply the --no-javadoc veto now that all arguments are parsed, so it wins over --ci +# regardless of flag order. +if [ "$NO_JAVADOC" = true ]; then + JAVADOC=false +fi + [ "$JAVADOC" = true ] && total_steps=4 || total_steps=2 current_step=0 From cb25ad27b07b34facb83cd5adff76b057855f7fb Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 17 Aug 2026 21:09:29 +0200 Subject: [PATCH 8/8] UNOMI-979: Sample per-process CPU and disk I/O during integration tests The integration tests take about 59 minutes on OpenSearch against 40 on Elasticsearch, and the existing metrics could not explain it. They cover memory only: both engines showed ample heap headroom, no swap and no warnings, so the gap was clearly not memory. System load hinted at the answer - the *slower* engine ran at a *lower* load, median 0.15 against 0.77 on a 2-vCPU runner - but system load alone cannot say whether a run is computing, blocked on disk, or waiting on a remote call. Adds six columns to memory-samples.tsv: karaf_cpu_pct, karaf_io_read_mb_s, karaf_io_write_mb_s, search_cpu_pct, search_io_read_mb_s, search_io_write_mb_s. They are appended, so existing column positions are untouched and older sample files still summarize. The summary gains cpu.mean/peak per process, io.peak per process, cpu.idle.samples.pct, and cpu.warning.mostly.idle, which fires when neither process uses meaningful CPU and I/O is negligible - the signature of a run whose time goes on waiting rather than work. CPU is a true interval percentage, not ps(1)'s %cpu: that is an average over the whole process lifetime, so a JVM busy during startup reads as busy forever and is useless for spotting a stall. On Linux, which is what CI runs and the only place these numbers are compared between runs, it deltas /proc//stat between samples. macOS has no procfs and falls back to ps, which is not comparable with a Linux sample; the code says so where it matters. The sample interval drops from 30s to 10s. At 30s a single sample spans several integration tests, so a stall cannot be attributed to the test that caused it. To keep that from tripling the load the sampler puts on the machine it is measuring, the search engine's memory, CPU and block I/O now come from ONE docker stats call per sample rather than the two an obvious implementation would use. Also fixes a pre-existing bug found while testing: it_memory_find_karaf_pid used pgrep, which exits non-zero when nothing matches, and under set -euo pipefail that aborted the whole sample. Since the sampler starts before Karaf does, every sample taken during startup was silently discarded - exactly the window where the search engine is booting and its resource use is most interesting. Verified on macOS and in an ubuntu:24.04 container using mawk, as ubuntu-latest does. On Linux: 100.0% for a saturated process, 0.0% for an idle one, 0 for a dead pid, 94 MB/s for a real writer, correct summaries under mawk, and a clean start/stop cycle. Fault injection confirms a bug in this script cannot fail a build: a syntax error only warns, a broken helper still yields samples, a hung sampler is killed promptly by stop, and an unwritable directory, corrupt cache or absent docker CLI all degrade to zeros. Co-Authored-By: Claude Opus 5 (1M context) --- build.sh | 8 +- itests/lib/it-run-memory.sh | 184 +++++++++++++++++++++++++++++++++--- itests/sample-it-memory.sh | 9 +- 3 files changed, 184 insertions(+), 17 deletions(-) diff --git a/build.sh b/build.sh index 6c93ecb710..408768e320 100755 --- a/build.sh +++ b/build.sh @@ -279,7 +279,9 @@ RESOLVER_DEBUG=false KEEP_CONTAINER=false IT_SEARCH_ENGINE_LOGS=false IT_MEMORY_SAMPLER=true -IT_MEMORY_INTERVAL=30 +# 10s, matching itests/sample-it-memory.sh: at 30s a single sample spans several ITs, so a +# stall cannot be attributed to the test that caused it. +IT_MEMORY_INTERVAL=10 JAVADOC=false NO_JAVADOC=false LOG_FILE="" @@ -328,7 +330,7 @@ EOF echo -e " ${CYAN}--keep-container${NC} Keep search engine container running after tests (for post-failure inspection)" echo -e " ${CYAN}--search-engine-logs${NC} Stream search engine Docker logs to the Maven console during integration tests" echo -e " ${CYAN}--no-memory-sampler${NC} Disable JVM/system memory sampling during integration tests" - echo -e " ${CYAN}--memory-interval SEC${NC} Memory sample interval in seconds (default: 30)" + echo -e " ${CYAN}--memory-interval SEC${NC} Memory sample interval in seconds (default: 10)" echo -e " ${CYAN}--javadoc${NC} Build and validate Javadoc after install (doclint errors fail; public/protected tag gaps warn)" echo -e " ${CYAN}--no-javadoc${NC} Skip Javadoc/checkstyle validation (overrides --ci; use when another job already ran it)" echo -e " ${CYAN}--ci${NC} CI mode: no Karaf, non-interactive, includes Javadoc" @@ -373,7 +375,7 @@ EOF echo " --keep-container Keep search engine container running after tests (for post-failure inspection)" echo " --search-engine-logs Stream search engine Docker logs to the Maven console during integration tests" echo " --no-memory-sampler Disable JVM/system memory sampling during integration tests" - echo " --memory-interval SEC Memory sample interval in seconds (default: 30)" + echo " --memory-interval SEC Memory sample interval in seconds (default: 10)" echo " --javadoc Build and validate Javadoc after install (doclint errors fail; public/protected tag gaps warn)" echo " --no-javadoc Skip Javadoc/checkstyle validation (overrides --ci; use when another job already ran it)" echo " --ci CI mode: no Karaf, non-interactive, includes Javadoc" diff --git a/itests/lib/it-run-memory.sh b/itests/lib/it-run-memory.sh index 1c8f434668..8c61cdd5b5 100644 --- a/itests/lib/it-run-memory.sh +++ b/itests/lib/it-run-memory.sh @@ -27,7 +27,7 @@ IT_MEMORY_SAMPLER_LOG="memory-sampler.log" IT_MEMORY_SAMPLER_CACHE="memory-sampler.cache" IT_MEMORY_SWAP_PRESSURE_MB=2048 -IT_MEMORY_TSV_HEADER=$'timestamp_utc\tkaraf_pid\tkaraf_heap_used_mb\tkaraf_heap_max_mb\tkaraf_gct_s\tes_heap_used_mb\tes_heap_max_mb\tdocker_rss_mb\tsystem_mem_available_mb\tsystem_swap_used_mb\tsystem_load_1m' +IT_MEMORY_TSV_HEADER=$'timestamp_utc\tkaraf_pid\tkaraf_heap_used_mb\tkaraf_heap_max_mb\tkaraf_gct_s\tes_heap_used_mb\tes_heap_max_mb\tdocker_rss_mb\tsystem_mem_available_mb\tsystem_swap_used_mb\tsystem_load_1m\tkaraf_cpu_pct\tkaraf_io_read_mb_s\tkaraf_io_write_mb_s\tsearch_cpu_pct\tsearch_io_read_mb_s\tsearch_io_write_mb_s' _IT_MEMORY_OS="" @@ -350,7 +350,12 @@ it_memory_parse_docker_mem_to_mb() { } it_memory_find_karaf_pid() { - pgrep -f 'org.apache.karaf.main.Main' 2>/dev/null | head -1 + # `|| true`: pgrep exits non-zero when nothing matches, and with `set -euo pipefail` that + # aborted the whole sample. The sampler starts before Karaf does, so every sample taken + # during startup was discarded -- exactly the window where the search engine is booting and + # its resource use is most interesting. No match now yields an empty pid, which the callers + # and the summarizer already treat as "no Karaf yet" (guarded by `if ($2+0 > 0)`). + pgrep -f 'org.apache.karaf.main.Main' 2>/dev/null | head -1 || true } it_memory_karaf_max_mb_cached() { @@ -446,21 +451,139 @@ it_memory_search_engine_stats() { echo -e "$(it_memory_mb_from_bytes "${used_bytes:-0}")\t$(it_memory_mb_from_bytes "${max_bytes:-0}")" } -it_memory_docker_rss_mb() { +# --- CPU and disk I/O sampling ------------------------------------------------- +# +# Added to answer "is the run CPU-bound, I/O-bound, or waiting?". The memory columns alone +# could not distinguish a busy run from an idle one blocked on a remote call, which is exactly +# the question raised by the Elasticsearch/OpenSearch IT duration gap: system load was near +# idle on the slower engine, so the extra time was spent waiting rather than computing. +# +# CPU is measured as a TRUE INTERVAL PERCENTAGE, not ps(1)'s %cpu -- that is an average over the +# whole process lifetime, so a JVM that was busy at startup reads as busy forever and the number +# is useless for spotting a stall. On Linux -- which is what CI runs, and the only place these +# numbers are compared across runs -- we delta /proc//stat between samples for a true +# interval figure. macOS has no procfs, so it falls back to ps(1)'s lifetime average: good enough +# to see that a process is alive and roughly how hard it has worked, but NOT comparable with a +# Linux sample and not to be read as "CPU right now". Everything here is best-effort: a missing +# file, a dead pid or an absent docker CLI yields 0 and never fails a run. + +# Stores "value timestamp" pairs so the next sample can compute a delta. +_it_memory_counter_cache() { + local target_dir="$1" key="$2" + echo "$target_dir/.it-memory-counter-$key" +} + +# Echoes the per-second rate between this reading and the previous one, or 0 on the first call. +_it_memory_rate_per_sec() { + local target_dir="$1" key="$2" value="$3" + local cache prev_value prev_ts now delta_v delta_t + cache="$(_it_memory_counter_cache "$target_dir" "$key")" + now="$(date +%s)" + + if [ -r "$cache" ]; then + read -r prev_value prev_ts < "$cache" 2>/dev/null || true + fi + printf '%s %s\n' "$value" "$now" > "$cache" 2>/dev/null || true + + if [ -z "${prev_value:-}" ] || [ -z "${prev_ts:-}" ]; then + echo "0" + return + fi + delta_t=$((now - prev_ts)) + [ "$delta_t" -le 0 ] && { echo "0"; return; } + delta_v="$(awk -v a="$value" -v b="$prev_value" 'BEGIN { d = a - b; print (d > 0 ? d : 0) }')" + awk -v d="$delta_v" -v t="$delta_t" 'BEGIN { printf "%.2f", d / t }' +} + +# Interval CPU% for a pid. >100 is legitimate on multi-core (sum across threads). +it_memory_process_cpu_pct() { + local target_dir="$1" pid="${2:-}" + local ticks hz cpu_s rate + + if [ -z "$pid" ] || [ "$pid" = "0" ] || ! kill -0 "$pid" 2>/dev/null; then + echo "0" + return + fi + + if it_memory_is_linux && [ -r "/proc/$pid/stat" ]; then + # Fields 14 (utime) and 15 (stime), in clock ticks. comm (field 2) is parenthesised and + # may itself contain spaces AND parentheses, so split after the LAST ')' rather than the + # first: a process named e.g. "java (worker)" otherwise shifts every subsequent index. + ticks="$(awk '{ + i = length($0) + while (i > 0 && substr($0, i, 1) != ")") i-- + n = split(substr($0, i + 2), f, " ") + if (n >= 13) print f[12] + f[13]; else print 0 + }' "/proc/$pid/stat" 2>/dev/null)" + [ -z "$ticks" ] && { echo "0"; return; } + hz="$(getconf CLK_TCK 2>/dev/null || echo 100)" + cpu_s="$(awk -v t="$ticks" -v hz="$hz" 'BEGIN { printf "%.4f", t / hz }')" + rate="$(_it_memory_rate_per_sec "$target_dir" "cpu-$pid" "$cpu_s")" + awk -v r="$rate" 'BEGIN { printf "%.1f", r * 100 }' + return + fi + + # macOS / no procfs: lifetime average, better than nothing for a local run. + ps -o %cpu= -p "$pid" 2>/dev/null | tr -d ' ' | awk 'NF { printf "%.1f", $1; found = 1 } END { if (!found) print 0 }' +} + +# Interval disk read/write in MB/s for a pid (Linux only; /proc//io). +it_memory_process_io_mb_s() { + local target_dir="$1" pid="${2:-}" + local read_bytes write_bytes read_rate write_rate + + if [ -z "$pid" ] || [ "$pid" = "0" ] || ! it_memory_is_linux || [ ! -r "/proc/$pid/io" ]; then + echo -e "0\t0" + return + fi + + read_bytes="$(awk '/^read_bytes:/ { print $2 }' "/proc/$pid/io" 2>/dev/null)" + write_bytes="$(awk '/^write_bytes:/ { print $2 }' "/proc/$pid/io" 2>/dev/null)" + read_rate="$(_it_memory_rate_per_sec "$target_dir" "ior-$pid" "${read_bytes:-0}")" + write_rate="$(_it_memory_rate_per_sec "$target_dir" "iow-$pid" "${write_bytes:-0}")" + awk -v r="$read_rate" -v w="$write_rate" 'BEGIN { printf "%.2f\t%.2f", r / 1048576, w / 1048576 }' +} + +# One docker stats call per sample, returning RSS, CPU% and block I/O together. +# +# Deliberately a single invocation: `docker stats --no-stream` costs ~1s and briefly loads the +# daemon, and the sampler now runs 3x more often (10s rather than 30s). Two calls per sample +# would have meant six times the docker traffic of the original sampler, perturbing the very +# run being measured and tripling the exposure to a hung daemon. CPUPerc is already an interval +# measurement; BlockIO is cumulative and is deltaed here. +# +# Echoes: rss_mb \t cpu_pct \t io_read_mb_s \t io_write_mb_s +it_memory_docker_sample() { local target_dir="$1" - local container rss + local container stats mem cpu blockio read_raw write_raw read_b write_b read_rate write_rate if ! command -v docker >/dev/null 2>&1; then - echo "0" + echo -e "0\t0\t0\t0" return fi container="$(it_memory_resolve_docker_container "$target_dir")" - rss="$(docker stats --no-stream --format '{{.MemUsage}}' "$container" 2>/dev/null | head -1 | cut -d/ -f1 | tr -d ' ')" + stats="$(docker stats --no-stream --format '{{.MemUsage}}|{{.CPUPerc}}|{{.BlockIO}}' "$container" 2>/dev/null | head -1)" + if [ -z "$stats" ]; then + echo -e "0\t0\t0\t0" + return + fi + + mem="$(echo "$stats" | cut -d'|' -f1 | cut -d/ -f1 | tr -d ' ')" + cpu="$(echo "$stats" | cut -d'|' -f2 | tr -d ' %')" + blockio="$(echo "$stats" | cut -d'|' -f3)" + read_raw="$(echo "$blockio" | cut -d/ -f1 | tr -d ' ')" + write_raw="$(echo "$blockio" | cut -d/ -f2 | tr -d ' ')" + read_b="$(it_memory_parse_docker_mem_to_mb "$read_raw")" + write_b="$(it_memory_parse_docker_mem_to_mb "$write_raw")" + read_rate="$(_it_memory_rate_per_sec "$target_dir" "dior" "${read_b:-0}")" + write_rate="$(_it_memory_rate_per_sec "$target_dir" "diow" "${write_b:-0}")" - it_memory_parse_docker_mem_to_mb "$rss" + printf '%s\t%.1f\t%.2f\t%.2f\n' \ + "$(it_memory_parse_docker_mem_to_mb "$mem")" "${cpu:-0}" "${read_rate:-0}" "${write_rate:-0}" } + it_memory_system_stats() { local mem_available swap_used load_1m @@ -482,13 +605,22 @@ it_memory_sample_once() { es_line="$(it_memory_search_engine_stats "$port")" sys_line="$(it_memory_system_stats)" - printf '%s\t%s\t%s\t%s\t%s\t%s\n' \ + # One docker call per sample; split into the RSS column (8) and the CPU/IO columns (15-17). + local docker_line docker_rss docker_cpu_io + docker_line="$(it_memory_docker_sample "$target_dir")" + docker_rss="$(printf '%s' "$docker_line" | cut -f1)" + docker_cpu_io="$(printf '%s' "$docker_line" | cut -f2-4)" + + printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \ "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \ "${karaf_pid:-0}" \ "$karaf_line" \ "$es_line" \ - "$(it_memory_docker_rss_mb "$target_dir")" \ - "$sys_line" + "${docker_rss:-0}" \ + "$sys_line" \ + "$(it_memory_process_cpu_pct "$target_dir" "$karaf_pid")" \ + "$(it_memory_process_io_mb_s "$target_dir" "$karaf_pid")" \ + "${docker_cpu_io:-$(printf '0\t0\t0')}" } it_memory_write_samples_header() { @@ -508,7 +640,7 @@ it_memory_summarize_samples() { awk -F'\t' -v summary="$summary_file" -v swap_pressure_mb="$IT_MEMORY_SWAP_PRESSURE_MB" ' NR == 1 { next } - NF < 11 { next } + NF < 11 { next } # pre-CPU/IO samples still summarize { samples++ if ($2+0 > 0) { @@ -524,6 +656,18 @@ it_memory_summarize_samples() { if ($11+0 > peak_load) peak_load = $11+0 if (samples == 1) first_swap = $10+0 last_swap = $10+0 + # CPU / IO columns are absent in samples written before they were added. + if (NF >= 17) { + cpu_samples++ + karaf_cpu_sum += $12+0; if ($12+0 > peak_karaf_cpu) peak_karaf_cpu = $12+0 + search_cpu_sum += $15+0; if ($15+0 > peak_search_cpu) peak_search_cpu = $15+0 + io_sum += $13+0 + $14+0 + $16+0 + $17+0 + if ($13+0 + $14+0 > peak_karaf_io) peak_karaf_io = $13+0 + $14+0 + if ($16+0 + $17+0 > peak_search_io) peak_search_io = $16+0 + $17+0 + # "Idle" = neither process using meaningful CPU: the signature of a run that is + # waiting on latency rather than doing work. + if ($12+0 < 10 && $15+0 < 10) idle_samples++ + } } END { if (samples == 0) exit 1 @@ -541,6 +685,24 @@ it_memory_summarize_samples() { printf("memory.min.system.mem.available.mb=%d\n", min_mem_avail+0) >> summary printf("memory.peak.system.swap.used.mb=%d\n", peak_swap+0) >> summary printf("memory.peak.system.load.1m=%.2f\n", peak_load+0) >> summary + if (cpu_samples > 0) { + printf("cpu.samples.count=%d\n", cpu_samples) >> summary + printf("cpu.mean.karaf.pct=%.1f\n", karaf_cpu_sum / cpu_samples) >> summary + printf("cpu.peak.karaf.pct=%.1f\n", peak_karaf_cpu+0) >> summary + printf("cpu.mean.search.pct=%.1f\n", search_cpu_sum / cpu_samples) >> summary + printf("cpu.peak.search.pct=%.1f\n", peak_search_cpu+0) >> summary + printf("cpu.idle.samples.pct=%d\n", idle_samples * 100 / cpu_samples) >> summary + printf("io.peak.karaf.mb.s=%.2f\n", peak_karaf_io+0) >> summary + printf("io.peak.search.mb.s=%.2f\n", peak_search_io+0) >> summary + printf("io.mean.total.mb.s=%.2f\n", io_sum / cpu_samples) >> summary + # Mostly-idle CPU with negligible I/O means the run is latency-bound: time is + # going on waiting (polls, refresh intervals, timeouts), not on work. + if (idle_samples * 100 / cpu_samples >= 70 && io_sum / cpu_samples < 5) { + printf("cpu.warning.mostly.idle=true\n") >> summary + } else { + printf("cpu.warning.mostly.idle=false\n") >> summary + } + } printf("memory.karaf.headroom.pct=%d\n", karaf_headroom+0) >> summary printf("memory.search.headroom.pct=%d\n", es_headroom+0) >> summary if (swap_pressure) { diff --git a/itests/sample-it-memory.sh b/itests/sample-it-memory.sh index 19e0d5d5c0..e0469bd02d 100755 --- a/itests/sample-it-memory.sh +++ b/itests/sample-it-memory.sh @@ -39,7 +39,10 @@ source "$SCRIPT_DIR/lib/it-run.sh" source "$SCRIPT_DIR/lib/it-run-memory.sh" TARGET_DIR="$SCRIPT_DIR/target" -INTERVAL=30 +# 10s, down from 30s: at 30s a sample covers several ITs at once, so a stall cannot be +# attributed to the test that caused it. Each sample is a few cheap reads plus one +# `docker stats --no-stream`, and a 50-minute run produces ~300 rows (a few tens of KB). +INTERVAL=10 SEARCH_PORT="" PRINT_ONLY=false COMMAND="" @@ -58,7 +61,7 @@ Commands: Options: --target-dir DIR IT target directory (default: itests/target) - --interval SEC Sample interval in seconds for start (default: 30) + --interval SEC Sample interval in seconds for start (default: 10) --port PORT Search engine HTTP port override --print-only With operator-note: print to stdout instead of writing file -h, --help Show this help @@ -78,7 +81,7 @@ parse_args() { ;; --interval) shift - INTERVAL="${1:-30}" + INTERVAL="${1:-10}" ;; --port) shift