diff --git a/pom.xml b/pom.xml index 34e6454af..0a9f294ee 100644 --- a/pom.xml +++ b/pom.xml @@ -369,37 +369,8 @@ - - - smx.m2 - Apache ServiceMix M2 - https://svn.apache.org/repos/asf/servicemix/m2-repo/ - - - - apache-snapshots - Apache Snapshots Repository - https://repository.apache.org/content/groups/snapshots-group - - false - - - true - - - - - ops4j.sonatype.snapshots.deploy - OPS4J snapshot repository - https://oss.sonatype.org/content/repositories/ops4j-snapshots/ - - false - - - true - - - + bom 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 27178469b..8f3bd41a4 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 @@ -47,6 +47,13 @@ public class TaskExecutionManager { private final Map> executingTasksByType; private final Map activeLockRenewals = new ConcurrentHashMap<>(); private final AtomicBoolean running = new AtomicBoolean(false); + /** + * Set at the start of {@link #shutdown()} before canceling in-flight work. Failures that race + * with shutdown must not schedule retries: {@link ScheduledExecutorService#isShutdown()} only + * flips after {@code scheduler.shutdown()}, which runs after {@code future.cancel(true)} and + * can race with {@link #handleTaskError}. + */ + private final AtomicBoolean shuttingDown = new AtomicBoolean(false); private ScheduledFuture taskCheckerFuture; private SchedulerServiceImpl schedulerService; private TaskExecutorRegistry executorRegistry; @@ -776,8 +783,10 @@ private void handleTaskError(ScheduledTask task, String error, long startTime) { metricsManager.updateMetric(TaskMetricsManager.METRIC_TASKS_EXECUTION_TIME, executionTime); if (scheduleRetry) { - // Only schedule retry if scheduler is not shutting down - if (!scheduler.isShutdown() && !scheduler.isTerminated()) { + // Only schedule retry if this manager is not shutting down. Check shuttingDown before + // scheduler.isShutdown(): canceling in-flight futures can invoke handleTaskError before + // scheduler.shutdown() runs, which would otherwise queue a retry that fires after teardown. + if (!shuttingDown.get() && !scheduler.isShutdown() && !scheduler.isTerminated()) { try { Runnable retryTask = () -> { TaskExecutor executor = executorRegistry.getExecutor(task.getTaskType()); @@ -844,6 +853,8 @@ public void cancelTask(String taskId) { * Shuts down the execution manager */ public void shutdown() { + // Mark before canceling futures so in-flight failures cannot schedule retries. + shuttingDown.set(true); stopTaskChecker(); // Stop all lock heartbeats so held locks age out and peers can recover the work 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 582ecb938..bafcc5356 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 @@ -324,8 +324,22 @@ public void testHandleTaskErrorSkipsRetryScheduleDuringShutdown() throws Excepti executionManager.executeTask(task, executor); assertTrue(inFlight.await(5, TimeUnit.SECONDS)); + // Ignore the stubbing / any pre-shutdown registry lookups; we only care about retries after shutdown. + clearInvocations(executorRegistry); + // Release the in-flight task shortly after shutdown starts so awaitTermination can finish + // without waiting the full timeout (shutdown cancels futures then awaits the pool). + Thread releaser = new Thread(() -> { + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + allowFail.countDown(); + }, "shut-retry-releaser"); + releaser.setDaemon(true); + releaser.start(); executionManager.shutdown(); - allowFail.countDown(); + releaser.join(2000); Thread.sleep(200); assertEquals(ScheduledTask.TaskStatus.SCHEDULED, task.getStatus()); assertEquals(1, task.getFailureCount());