This is an automated email from the ASF dual-hosted git repository.
sergehuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 5f8f4dc30 UNOMI-939/940: Scheduler dispatch and cache refresh race
fixes (#792)
5f8f4dc30 is described below
commit 5f8f4dc300fe72c82775a3d2448acbf121952729
Author: Serge Huber <[email protected]>
AuthorDate: Wed Jul 1 06:09:20 2026 +0200
UNOMI-939/940: Scheduler dispatch and cache refresh race fixes (#792)
Merge scheduler cache improvements
---
.../cache/AbstractMultiTypeCachingService.java | 59 +++++++++++-
.../cache/AbstractMultiTypeCachingServiceTest.java | 90 ++++++++++++++++++
.../impl/scheduler/SchedulerServiceImpl.java | 8 +-
.../impl/scheduler/TaskExecutionManager.java | 84 +++++++++++------
.../impl/scheduler/TaskRecoveryManager.java | 9 ++
.../services/impl/scheduler/TaskStateManager.java | 5 +-
.../services/impl/events/EventServiceImplTest.java | 7 +-
.../impl/scheduler/SchedulerServiceImplTest.java | 104 +++++++++++++++++++++
.../tracing/impl/DefaultTracerServiceTest.java | 4 +-
9 files changed, 327 insertions(+), 43 deletions(-)
diff --git
a/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java
b/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java
index 499c7aa7d..4891fc347 100644
---
a/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java
+++
b/services-common/src/main/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingService.java
@@ -43,6 +43,9 @@ import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -68,6 +71,25 @@ public abstract class AbstractMultiTypeCachingService
extends AbstractContextAwa
protected TenantService tenantService;
protected AuditService auditService;
+ /**
+ * Per-item-type locks ensuring saveItem()/removeItem() (direct writes)
and refreshTypeCache()'s
+ * persistence-read-then-cache-write cycle can never interleave for the
same item type. Without this,
+ * refreshTypeCache() could read a snapshot from persistence just before a
concurrent direct write
+ * commits, then overwrite the cache with that stale snapshot after the
direct write's own cache.put()
+ * already applied the fresh value. Direct writes only need to exclude a
concurrent refresh of the
+ * same type, not each other, so they take the read lock (shared,
concurrent) while refreshTypeCache()
+ * takes the write lock (exclusive) for its whole read-then-cache cycle,
including the "remove items no
+ * longer in persistence" step, which reads from the same persistence
snapshot.
+ *
+ * Cardinality is bounded by the number of distinct item types a service
registers (getTypeConfigs()),
+ * typically one to a handful per service, so this map needs no eviction.
+ */
+ private final Map<String, ReadWriteLock> typeRefreshLocks = new
ConcurrentHashMap<>();
+
+ private ReadWriteLock typeRefreshLock(String itemType) {
+ return typeRefreshLocks.computeIfAbsent(itemType, k -> new
ReentrantReadWriteLock());
+ }
+
/**
* Map tracking which plugin/bundle contributed which items.
* Key is the bundle ID, value is the list of items contributed by that
bundle.
@@ -261,12 +283,25 @@ public abstract class AbstractMultiTypeCachingService
extends AbstractContextAwa
scheduledRefreshTasks.clear();
}
- @SuppressWarnings("unchecked")
protected <T extends Serializable> void
refreshTypeCache(CacheableTypeConfig<T> config) {
if (!config.isRequiresRefresh()) {
return;
}
+ // Excludes any concurrent saveItem()/removeItem() for this item type
for the duration of this
+ // refresh, so its persistence read can never be overwritten by - or
overwrite - a direct write.
+ // See typeRefreshLocks for the full rationale.
+ Lock refreshLock = typeRefreshLock(config.getItemType()).writeLock();
+ refreshLock.lock();
+ try {
+ refreshTypeCacheLocked(config);
+ } finally {
+ refreshLock.unlock();
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T extends Serializable> void
refreshTypeCacheLocked(CacheableTypeConfig<T> config) {
// Only create the global state maps if we need them
Map<String, Map<String, T>> oldGlobalState = null;
Map<String, Map<String, T>> newGlobalState = null;
@@ -858,8 +893,15 @@ public abstract class AbstractMultiTypeCachingService
extends AbstractContextAwa
}
}
- persistenceService.save(item);
- cacheService.put(itemType, itemId, currentTenant, item);
+ // Excludes a concurrent refreshTypeCache() for this item type: see
typeRefreshLocks.
+ Lock writeGuard = typeRefreshLock(itemType).readLock();
+ writeGuard.lock();
+ try {
+ persistenceService.save(item);
+ cacheService.put(itemType, itemId, currentTenant, item);
+ } finally {
+ writeGuard.unlock();
+ }
}
/**
@@ -872,7 +914,14 @@ public abstract class AbstractMultiTypeCachingService
extends AbstractContextAwa
*/
protected <T extends Item & Serializable> void removeItem(String id,
Class<T> itemClass, String itemType) {
String currentTenant =
contextManager.getCurrentContext().getTenantId();
- persistenceService.remove(id, itemClass);
- cacheService.remove(itemType, id, currentTenant, itemClass);
+ // Excludes a concurrent refreshTypeCache() for this item type: see
typeRefreshLocks.
+ Lock writeGuard = typeRefreshLock(itemType).readLock();
+ writeGuard.lock();
+ try {
+ persistenceService.remove(id, itemClass);
+ cacheService.remove(itemType, id, currentTenant, itemClass);
+ } finally {
+ writeGuard.unlock();
+ }
}
}
diff --git
a/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java
b/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java
index 73a1684a2..57923a771 100644
---
a/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java
+++
b/services-common/src/test/java/org/apache/unomi/services/common/cache/AbstractMultiTypeCachingServiceTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.unomi.services.common.cache;
+import org.apache.unomi.api.ExecutionContext;
import org.apache.unomi.api.Item;
import org.apache.unomi.api.services.ExecutionContextManager;
import org.apache.unomi.api.services.cache.CacheableTypeConfig;
@@ -32,6 +33,9 @@ import org.mockito.junit.MockitoJUnitRunner;
import java.io.Serializable;
import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import static org.junit.Assert.*;
@@ -110,6 +114,18 @@ public class AbstractMultiTypeCachingServiceTest {
}
}
+ // Minimal Item subclass: saveItem()/removeItem() require <T extends Item
& Serializable>,
+ // which the plain-Serializable TestSerializable above does not satisfy.
+ // Public (not just the ITEM_TYPE field): Item's reflective ITEM_TYPE
lookup needs the
+ // declaring class itself to be accessible, not just the field's own
modifiers.
+ public static class TestItem extends Item {
+ public static final String ITEM_TYPE = TEST_ITEM_TYPE;
+
+ public TestItem(String itemId) {
+ super(itemId);
+ }
+ }
+
private static class TestCachingServiceImpl extends
AbstractMultiTypeCachingService {
private final Set<CacheableTypeConfig<?>> typeConfigs = new
HashSet<>();
@@ -136,6 +152,11 @@ public class AbstractMultiTypeCachingServiceTest {
return typeConfigs;
}
+ // Exposes the protected saveItem() for the concurrency test below.
+ void callSaveItem(TestItem item) {
+ saveItem(item, Item::getItemId, TEST_ITEM_TYPE);
+ }
+
// Helper method to set a config as persistable for testing
void makeConfigPersistable() {
try {
@@ -194,6 +215,9 @@ public class AbstractMultiTypeCachingServiceTest {
runnable.run();
return null;
}).when(contextManager).executeAsSystem(any(Runnable.class));
+
+ // saveItem() reads the current tenant from
contextManager.getCurrentContext().getTenantId().
+ when(contextManager.getCurrentContext()).thenReturn(new
ExecutionContext(TEST_TENANT, null, null));
}
@Test
@@ -377,4 +401,70 @@ public class AbstractMultiTypeCachingServiceTest {
// Verify items were removed from system tenant
verify(cacheService).remove(eq(TEST_ITEM_TYPE), eq("item3"),
eq(SYSTEM_TENANT), eq(TestSerializable.class));
}
+
+ /**
+ * Validates the ReadWriteLock fix in AbstractMultiTypeCachingService:
saveItem() must block while
+ * a refreshTypeCache() for the same item type holds the write lock, so a
refresh's persistence read
+ * can never be followed by it overwriting a cache entry that a concurrent
direct write just applied.
+ * Without the fix, saveItem() and refreshTypeCache() could interleave
freely and the two race
+ * scenarios this test would otherwise need to get lucky on
(read-before-write, write-after-read)
+ * would only manifest occasionally - here we force the interleaving
deterministically instead.
+ */
+ @Test
+ public void testSaveItemExcludedByConcurrentRefresh() throws Exception {
+ when(cacheService.getTenantCache(eq(TEST_TENANT),
eq(TestSerializable.class))).thenReturn(new HashMap<>());
+ when(cacheService.getTenantCache(eq(SYSTEM_TENANT),
eq(TestSerializable.class))).thenReturn(new HashMap<>());
+
doReturn(Collections.singleton(TEST_TENANT)).when(testCachingService).getTenants();
+
+ CacheableTypeConfig<TestSerializable> config = null;
+ for (CacheableTypeConfig<?> typeConfig :
testCachingService.getTypeConfigs()) {
+ if (typeConfig.getType().equals(TestSerializable.class)) {
+ @SuppressWarnings("unchecked")
+ CacheableTypeConfig<TestSerializable> typedConfig =
(CacheableTypeConfig<TestSerializable>) typeConfig;
+ config = typedConfig;
+ break;
+ }
+ }
+ assertNotNull("Should find config for TestSerializable", config);
+ final CacheableTypeConfig<TestSerializable> finalConfig = config;
+
+ CountDownLatch refreshHoldingLock = new CountDownLatch(1);
+ CountDownLatch releaseRefresh = new CountDownLatch(1);
+ AtomicBoolean refreshStillRunningWhenSaveReturned = new
AtomicBoolean(true);
+
+ // Stand-in for the persistence read inside refreshTypeCache(): signal
that the refresh has
+ // entered its critical section (and therefore holds the write lock),
then block until the
+ // test releases it.
+ doAnswer(invocation -> {
+ refreshHoldingLock.countDown();
+ assertTrue("Test setup: refresh should be released within timeout",
+ releaseRefresh.await(5, TimeUnit.SECONDS));
+ return Collections.emptyList();
+ }).when(testCachingService).loadItemsForTenant(eq(TEST_TENANT),
eq(finalConfig));
+
+ Thread refreshThread = new Thread(() ->
testCachingService.refreshTypeCache(finalConfig));
+ refreshThread.start();
+
+ assertTrue("Refresh should reach loadItemsForTenant while holding the
write lock",
+ refreshHoldingLock.await(5, TimeUnit.SECONDS));
+
+ Thread saveThread = new Thread(() -> {
+ testCachingService.callSaveItem(new TestItem("item1"));
+ refreshStillRunningWhenSaveReturned.set(refreshThread.isAlive());
+ });
+ saveThread.start();
+
+ // saveItem()'s read lock should be blocked behind the refresh's write
lock: give it a moment,
+ // then confirm it has NOT returned yet.
+ saveThread.join(500);
+ assertTrue("saveItem() should still be blocked while refresh holds the
write lock", saveThread.isAlive());
+
+ releaseRefresh.countDown();
+ refreshThread.join(5000);
+ saveThread.join(5000);
+
+ assertFalse("saveThread should have finished", saveThread.isAlive());
+ assertFalse("saveItem() must not proceed until the concurrent refresh
released its write lock",
+ refreshStillRunningWhenSaveReturned.get());
+ }
}
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 74f3cbe71..646eb3561 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
@@ -1172,10 +1172,14 @@ public class SchedulerServiceImpl implements
SchedulerService {
}
ScheduledTask task = getTask(taskId);
if (task != null) {
- // Only cancel if in a cancellable state
+ // Only cancel if in a cancellable state. COMPLETED is included
because a periodic task
+ // can be observed in that transient state by a caller racing
handleTaskCompletion()
+ // (status briefly goes RUNNING -> COMPLETED -> SCHEDULED);
without it, a cancelTask()
+ // call landing in that window would silently no-op and leave the
task enabled.
if (task.getStatus() == ScheduledTask.TaskStatus.SCHEDULED ||
task.getStatus() == ScheduledTask.TaskStatus.WAITING ||
- task.getStatus() == ScheduledTask.TaskStatus.RUNNING) {
+ task.getStatus() == ScheduledTask.TaskStatus.RUNNING ||
+ task.getStatus() == ScheduledTask.TaskStatus.COMPLETED) {
task.setEnabled(false);
stateManager.updateTaskState(task,
ScheduledTask.TaskStatus.CANCELLED, null, nodeId);
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 f787ac563..206e2eb00 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
@@ -176,13 +176,28 @@ public class TaskExecutionManager {
// Ensure the executing set exists even under concurrent clears
during shutdown
Set<String> executingSet =
executingTasksByType.computeIfAbsent(taskType, k ->
ConcurrentHashMap.newKeySet());
+ // Atomically claim this task before dispatching. The
persisted/cached status check above
+ // is only updated once prepareForExecution() actually runs inside
the async taskWrapper below,
+ // so a second poll tick landing in that gap could otherwise see a
stale non-RUNNING status and
+ // dispatch the same task again. Set.add() is atomic, so only one
caller can win this race.
+ if (!executingSet.add(task.getItemId())) {
+ LOGGER.debug("Node {} : Task {} is already dispatched for
execution, skipping duplicate dispatch", nodeId, task.getItemId());
+ return;
+ }
+
TaskExecutor.TaskStatusCallback statusCallback =
createStatusCallback(task);
Runnable taskWrapper = createTaskWrapper(task, executor,
statusCallback);
// Execute task immediately using the scheduler
- ScheduledFuture<?> future = scheduler.schedule(taskWrapper, 0,
TimeUnit.MILLISECONDS);
- scheduledTasks.put(task.getItemId(), future);
- executingSet.add(task.getItemId());
+ try {
+ ScheduledFuture<?> future = scheduler.schedule(taskWrapper, 0,
TimeUnit.MILLISECONDS);
+ scheduledTasks.put(task.getItemId(), future);
+ } catch (Exception e) {
+ // Scheduling failed (e.g. rejected during shutdown): release
the claim so a later
+ // attempt isn't permanently blocked, since the wrapper that
would normally do so never ran.
+ executingSet.remove(task.getItemId());
+ throw e;
+ }
} catch (Exception e) {
LOGGER.error("Node "+nodeId+", Error executing task: " +
task.getItemId(), e);
handleTaskError(task, e.getMessage(), System.currentTimeMillis());
@@ -281,35 +296,34 @@ public class TaskExecutionManager {
return;
}
- // Check shutdown again before preparing for execution
- if (schedulerService != null && schedulerService.isShutdownNow()) {
- LOGGER.debug("Node {} : Skipping task {} execution as
scheduler is shutting down", nodeId, taskId);
- return;
- }
-
- // Prepare task for execution (both persistent and in-memory)
- if (!prepareForExecution(task)) {
- return;
- }
-
- // Final shutdown check before executing
- if (schedulerService != null && schedulerService.isShutdownNow()) {
- LOGGER.debug("Node {} : Skipping task {} execution as
scheduler is shutting down", nodeId, taskId);
- return;
- }
-
+ // From here on, executeTask() has already atomically claimed
taskId in executingTasksByType
+ // before scheduling this wrapper. Every exit path below -
including the early returns for
+ // shutdown and prepareForExecution() failing (e.g. lock already
held by the caller, as
+ // happens when
TaskRecoveryManager.attemptTaskResumption()/attemptTaskRestart() acquire the
+ // lock themselves before calling executeTask()) - must release
that claim in the outer
+ // finally below, or the task is permanently blocked from ever
being dispatched again.
+ boolean executingNodeIdSet = false;
try {
- // Get or create the executing tasks set
- Set<String> executingTasks =
executingTasksByType.computeIfAbsent(taskType,
- k -> ConcurrentHashMap.newKeySet());
+ // Check shutdown again before preparing for execution
+ if (schedulerService != null &&
schedulerService.isShutdownNow()) {
+ LOGGER.debug("Node {} : Skipping task {} execution as
scheduler is shutting down", nodeId, taskId);
+ return;
+ }
+
+ // Prepare task for execution (both persistent and in-memory)
+ if (!prepareForExecution(task)) {
+ return;
+ }
- // Only add to executing set if not already there
- if (taskId != null) {
- executingTasks.add(taskId);
+ // Final shutdown check before executing
+ if (schedulerService != null &&
schedulerService.isShutdownNow()) {
+ LOGGER.debug("Node {} : Skipping task {} execution as
scheduler is shutting down", nodeId, taskId);
+ return;
}
// Set the executing node ID
task.setExecutingNodeId(nodeId);
+ executingNodeIdSet = true;
schedulerService.saveTask(task);
long startTime = System.currentTimeMillis();
@@ -329,11 +343,15 @@ public class TaskExecutionManager {
LOGGER.error("Unexpected error while executing task: " +
taskId, e);
statusCallback.fail("Unexpected error: " + e.getMessage());
} finally {
- // Clear executing node ID
- task.setExecutingNodeId(null);
- schedulerService.saveTask(task);
+ // Only clear/save executingNodeId if we actually set it
above; otherwise we never
+ // touched the task and a redundant save here could race a
concurrent legitimate holder.
+ if (executingNodeIdSet) {
+ task.setExecutingNodeId(null);
+ schedulerService.saveTask(task);
+ }
- // Remove task from executing set
+ // Always release the dispatch claim taken by executeTask(),
regardless of which path
+ // above was taken.
try {
Set<String> executingTasks =
executingTasksByType.get(taskType);
if (executingTasks != null && taskId != null) {
@@ -424,7 +442,11 @@ public class TaskExecutionManager {
executeTask(task, executor);
}
};
- long retryDelay =
task.getNextScheduledExecution().getTime() - System.currentTimeMillis();
+ // Use the configured retry delay directly rather than
re-deriving it from
+ // nextScheduledExecution: that target was computed
before the state/history/metrics
+ // bookkeeping above ran, so subtracting "now" here
would silently erode the delay
+ // by however long that bookkeeping took (worse under
slower/contended runners).
+ long retryDelay = task.getRetryDelay();
scheduler.schedule(retryTask, retryDelay,
TimeUnit.MILLISECONDS);
LOGGER.debug("Scheduled retry #{} for task {} in {}
ms",
task.getFailureCount(), task.getItemId(),
retryDelay);
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 03691ba62..56c771317 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
@@ -134,6 +134,15 @@ public class TaskRecoveryManager {
return;
}
+ // Re-check shutdown right before writing: preDestroy() marks RUNNING
tasks owned by this
+ // node as crashed with a more specific "Interrupted by scheduler
shutdown" cause, and that
+ // write must win over the generic "Node failure detected" cause below
if both race.
+ if (shutdownNow) {
+ LOGGER.debug("Node {} Skipping recovery of task {} : {} as
scheduler is shutting down",
+ nodeId, task.getTaskType(), task.getItemId());
+ return;
+ }
+
// First mark as crashed and release lock
String previousOwner = task.getLockOwner();
if (task.getStatus() != ScheduledTask.TaskStatus.CRASHED) {
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 cce41b29b..1e909bd90 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
@@ -39,7 +39,10 @@ public class TaskStateManager {
EXECUTE(TaskStatus.RUNNING, EnumSet.of(TaskStatus.SCHEDULED,
TaskStatus.CRASHED, TaskStatus.WAITING)),
COMPLETE(TaskStatus.COMPLETED, EnumSet.of(TaskStatus.RUNNING)),
FAIL(TaskStatus.FAILED, EnumSet.of(TaskStatus.RUNNING)),
- CANCEL(TaskStatus.CANCELLED, EnumSet.of(TaskStatus.RUNNING,
TaskStatus.SCHEDULED, TaskStatus.WAITING)),
+ // COMPLETED is included because a periodic task can be observed in
that transient state
+ // (briefly, between finishing one run and being rescheduled for the
next) by a concurrent
+ // cancelTask() call; cancellation should still apply rather than
silently no-op.
+ CANCEL(TaskStatus.CANCELLED, EnumSet.of(TaskStatus.RUNNING,
TaskStatus.SCHEDULED, TaskStatus.WAITING, TaskStatus.COMPLETED)),
CRASH(TaskStatus.CRASHED, EnumSet.of(TaskStatus.RUNNING,
TaskStatus.SCHEDULED)),
WAIT(TaskStatus.WAITING, EnumSet.of(TaskStatus.SCHEDULED,
TaskStatus.RUNNING));
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/events/EventServiceImplTest.java
b/services/src/test/java/org/apache/unomi/services/impl/events/EventServiceImplTest.java
index 109d5ef97..a5092fd22 100644
---
a/services/src/test/java/org/apache/unomi/services/impl/events/EventServiceImplTest.java
+++
b/services/src/test/java/org/apache/unomi/services/impl/events/EventServiceImplTest.java
@@ -417,10 +417,13 @@ public class EventServiceImplTest {
condition.setParameter("propertyValue", "test");
query.setCondition(condition);
- // Test - retry until event is available (handles refresh delay)
+ // Test - retry until event is available (handles refresh delay).
+ // Full-text indexing can take longer than the default 2000ms budget
on loaded CI runners,
+ // so use a more generous timeout for this specific query.
PartialList<Event> result = TestHelper.retryQueryUntilAvailable(
() -> eventService.search(query),
- 1
+ 1,
+ 10000L
);
// Verify
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 593867e2a..74bb68f12 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
@@ -622,6 +622,56 @@ public class SchedulerServiceImplTest {
assertTrue(recovered.get(), "Task should be recovered and executed");
}
+ @Test
+ @Tag("ClusterTests")
+ public void testNoDuplicateDispatchUnderSlowExecution() throws Exception {
+ // Validates TaskExecutionManager.executeTask(): the dispatch claim it
takes in
+ // executingTasksByType before scheduling a task wrapper must hold for
the entire run, or a
+ // poll tick landing while the first run is still in flight (status
not yet flipped to RUNNING,
+ // or - as observed before this fix - never released because
prepareForExecution() returned
+ // false on an early-return path) could dispatch the same task a
second time. The execution
+ // below deliberately runs past several poll ticks
(TASK_CHECK_INTERVAL = 1000ms) to exercise
+ // that window.
+ AtomicInteger concurrentExecutions = new AtomicInteger(0);
+ AtomicInteger maxConcurrentExecutions = new AtomicInteger(0);
+ AtomicInteger totalExecutions = new AtomicInteger(0);
+ CountDownLatch completionLatch = new CountDownLatch(1);
+
+ TaskExecutor executor = new TaskExecutor() {
+ @Override
+ public String getTaskType() {
+ return "slow-dispatch-test";
+ }
+
+ @Override
+ public void execute(ScheduledTask task, TaskStatusCallback
callback) {
+ int concurrent = concurrentExecutions.incrementAndGet();
+ maxConcurrentExecutions.updateAndGet(max -> Math.max(max,
concurrent));
+ totalExecutions.incrementAndGet();
+ try {
+ Thread.sleep(2500);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ concurrentExecutions.decrementAndGet();
+ }
+ callback.complete();
+ completionLatch.countDown();
+ }
+ };
+
+ schedulerService.registerTaskExecutor(executor);
+ schedulerService.newTask("slow-dispatch-test")
+ .disallowParallelExecution()
+ .schedule();
+
+ assertTrue(completionLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), "Task
should complete");
+
+ assertEquals(1, maxConcurrentExecutions.get(), "Task must never run
concurrently with itself");
+ assertEquals(1, totalExecutions.get(),
+ "Task must execute exactly once despite a slow run spanning
multiple poll ticks");
+ }
+
// Metrics and history tests
@Test
@Tag("MetricsTests")
@@ -1223,6 +1273,60 @@ public class SchedulerServiceImplTest {
assertFalse(completed.get(), "Task should not complete after
cancellation");
}
+ /**
+ * Validates the fix broadening the CANCEL transition to accept COMPLETED.
A periodic task only
+ * observes COMPLETED transiently (RUNNING -> COMPLETED ->
SCHEDULED), but a one-shot task stays
+ * COMPLETED indefinitely once it finishes, giving a fully deterministic
way to exercise
+ * cancelTask() against a task that is genuinely COMPLETED rather than
racing the transient window.
+ * Before the fix, cancelTaskInternal() only handled
SCHEDULED/WAITING/RUNNING and silently left a
+ * COMPLETED task enabled.
+ */
+ @Test
+ @Tag("StateTests")
+ public void testCancelTaskWhileCompleted() throws Exception {
+ CountDownLatch executionLatch = new CountDownLatch(1);
+
+ TaskExecutor executor = new TaskExecutor() {
+ @Override
+ public String getTaskType() {
+ return "cancel-completed-test";
+ }
+
+ @Override
+ public void execute(ScheduledTask task, TaskStatusCallback
callback) {
+ callback.complete();
+ executionLatch.countDown();
+ }
+ };
+
+ schedulerService.registerTaskExecutor(executor);
+
+ ScheduledTask task = schedulerService.newTask("cancel-completed-test")
+ .asOneShot()
+ .schedule();
+
+ assertTrue(executionLatch.await(TEST_TIMEOUT, TEST_TIME_UNIT), "Task
should execute");
+
+ TestHelper.retryUntil(
+ () -> schedulerService.getTask(task.getItemId()),
+ t -> t != null && t.getStatus() ==
ScheduledTask.TaskStatus.COMPLETED
+ );
+
+ schedulerService.cancelTask(task.getItemId());
+
+ TestHelper.retryUntil(
+ () -> schedulerService.getTask(task.getItemId()),
+ t -> t != null && t.getStatus() ==
ScheduledTask.TaskStatus.CANCELLED
+ );
+
+ ScheduledTask cancelledTask =
schedulerService.getTask(task.getItemId());
+ assertEquals(
+ ScheduledTask.TaskStatus.CANCELLED,
+ cancelledTask.getStatus(),
+ "cancelTask() on a COMPLETED task must actually cancel it, not
silently no-op");
+ assertFalse(cancelledTask.isEnabled(), "Cancelled task should be
disabled");
+ }
+
/**
* Tests task querying and filtering capabilities.
*/
diff --git
a/tracing/tracing-impl/src/test/java/org/apache/unomi/tracing/impl/DefaultTracerServiceTest.java
b/tracing/tracing-impl/src/test/java/org/apache/unomi/tracing/impl/DefaultTracerServiceTest.java
index 87577ae7a..15771f9b7 100644
---
a/tracing/tracing-impl/src/test/java/org/apache/unomi/tracing/impl/DefaultTracerServiceTest.java
+++
b/tracing/tracing-impl/src/test/java/org/apache/unomi/tracing/impl/DefaultTracerServiceTest.java
@@ -86,9 +86,9 @@ public class DefaultTracerServiceTest {
RequestTracer tracer = tracerService.getCurrentTracer();
// Start a root operation
- tracer.startOperation("test", "Root operation", null);
long startTime = System.currentTimeMillis();
-
+ tracer.startOperation("test", "Root operation", null);
+
// Add some traces
tracer.trace("Test message", null);
tracer.trace("Test with context", "context-data");
