This is an automated email from the ASF dual-hosted git repository. imbajin pushed a commit to branch fix/scheduler in repository https://gitbox.apache.org/repos/asf/hugegraph.git
commit 69d11fe315405206de37463ef8b6aa613c019375 Author: imbajin <[email protected]> AuthorDate: Tue Jul 7 21:55:06 2026 +0800 fix(task): prevent task resurrection on delete - cancel locally running standard tasks before force deletion - tolerate missing distributed tasks during cancel retry - add regression coverage for delete and cancel races --- .../hugegraph/task/DistributedTaskScheduler.java | 8 ++- .../hugegraph/task/StandardTaskScheduler.java | 13 ++++- .../org/apache/hugegraph/core/TaskCoreTest.java | 48 ++++++++++++++++ .../unit/core/TaskSchedulerServerInfoTest.java | 66 ++++++++++++++++++++++ 4 files changed, 133 insertions(+), 2 deletions(-) diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java index 0edef4ecb..8b11db9b6 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java @@ -330,7 +330,13 @@ public class DistributedTaskScheduler extends TaskAndResultScheduler { task.overwriteStatus(TaskStatus.CANCELLING); } else { // Status race: re-read from DB and retry with fresh status - HugeTask<Object> reloaded = this.taskWithoutResult(task.id()); + HugeTask<Object> reloaded; + try { + reloaded = this.taskWithoutResult(task.id()); + } catch (NotFoundException e) { + LOG.info("Task '{}' already deleted, skip cancel", task.id()); + return; + } TaskStatus stored = reloaded.status(); if (stored != TaskStatus.CANCELLING && !TaskStatus.COMPLETED_STATUSES.contains(stored)) { diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java index 6a2e6ae5e..269094bc7 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java @@ -459,6 +459,7 @@ public class StandardTaskScheduler implements TaskScheduler { public <V> HugeTask<V> delete(Id id, boolean force) { this.checkOnMasterNode("delete"); + HugeTask<?> running = this.tasks.get(id); HugeTask<?> task = this.task(id, false); /* * The following is out of date when the task running on worker node: @@ -470,7 +471,17 @@ public class StandardTaskScheduler implements TaskScheduler { * in fact, it is also possible to appear on the backend tasks * when the database status is inconsistent. */ - if (task != null) { + if (running != null) { + E.checkArgument(force || running.completed(), + "Can't delete incomplete task '%s' in status %s" + + ", Please try to cancel the task first", + id, running.status()); + if (force && !running.completed()) { + running.overwriteStatus(TaskStatus.DELETING); + running.cancel(true); + } + this.remove(running, force); + } else if (task != null) { E.checkArgument(force || task.completed(), "Can't delete incomplete task '%s' in status %s" + ", Please try to cancel the task first", diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java index 5431be111..9e8269268 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/TaskCoreTest.java @@ -22,7 +22,9 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Random; +import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -262,6 +264,52 @@ public class TaskCoreTest extends BaseCoreTest { } } + @Test + public void testForceDeleteRunningTaskDoesNotResurrectAfterDone() + throws Exception { + HugeGraph graph = graph(); + TaskScheduler scheduler = graph.taskScheduler(); + if (!(scheduler instanceof StandardTaskScheduler)) { + return; + } + + Id id = IdGenerator.of(88897); + BlockingCallable.reset(); + HugeTask<?> task = new HugeTask<>(id, null, new BlockingCallable<>()); + task.type("test"); + task.name("force-delete-running-task"); + Future<?> future = null; + + try { + future = scheduler.schedule(task); + waitUntilTaskRunning(scheduler); + Assert.assertTrue(BlockingCallable.awaitStarted()); + + HugeTask<?> deleted = scheduler.delete(id, true); + Assert.assertNotNull(deleted); + Assert.assertThrows(NotFoundException.class, () -> { + scheduler.task(id); + }); + + BlockingCallable.release(); + try { + future.get(10L, TimeUnit.SECONDS); + } catch (CancellationException ignored) { + // Expected after deleting a locally running task. + } + + Assert.assertThrows(NotFoundException.class, () -> { + scheduler.task(id); + }); + } finally { + BlockingCallable.release(); + if (future != null && !future.isDone()) { + future.cancel(true); + } + deleteTaskAndWaitGone(scheduler, id); + } + } + @Test public void testDeleteRemoteLockedIncompleteTaskFailsFast() { HugeGraph graph = graph(); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/TaskSchedulerServerInfoTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/TaskSchedulerServerInfoTest.java index 32d646d55..758b8c1f2 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/TaskSchedulerServerInfoTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/TaskSchedulerServerInfoTest.java @@ -23,13 +23,19 @@ import java.util.concurrent.Executors; import org.apache.commons.configuration2.PropertiesConfiguration; import org.apache.hugegraph.HugeGraph; import org.apache.hugegraph.HugeGraphParams; +import org.apache.hugegraph.backend.id.Id; +import org.apache.hugegraph.backend.id.IdGenerator; import org.apache.hugegraph.concurrent.PausableScheduledThreadPool; import org.apache.hugegraph.config.CoreOptions; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.config.ServerOptions; import org.apache.hugegraph.core.GraphManager; import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.exception.NotFoundException; import org.apache.hugegraph.task.DistributedTaskScheduler; +import org.apache.hugegraph.task.HugeTask; +import org.apache.hugegraph.task.TaskCallable; +import org.apache.hugegraph.task.TaskStatus; import org.apache.hugegraph.testutil.Assert; import org.apache.hugegraph.testutil.Whitebox; import org.apache.hugegraph.util.ExecutorUtil; @@ -76,6 +82,66 @@ public class TaskSchedulerServerInfoTest { } } + @Test + public void testDistributedCancelTreatsMissingRetryTaskAsGone() { + HugeGraph graph = Mockito.mock(HugeGraph.class); + Mockito.when(graph.graphSpace()).thenReturn("DEFAULT"); + + HugeGraphParams params = Mockito.mock(HugeGraphParams.class); + Mockito.when(params.graph()).thenReturn(graph); + Mockito.when(params.name()).thenReturn("hugegraph"); + Mockito.when(params.spaceGraphName()).thenReturn("DEFAULT-hugegraph"); + Mockito.when(params.configuration()).thenReturn(newConfig()); + + PausableScheduledThreadPool schedulerExecutor = + ExecutorUtil.newPausableScheduledThreadPool( + 1, "distributed-cancel-test-%d"); + ExecutorService taskDbExecutor = Executors.newSingleThreadExecutor(); + ExecutorService schemaTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService olapTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService gremlinTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService ephemeralTaskExecutor = Executors.newSingleThreadExecutor(); + ExecutorService serverInfoDbExecutor = Executors.newSingleThreadExecutor(); + + try { + DistributedTaskScheduler scheduler = new DistributedTaskScheduler( + params, schedulerExecutor, taskDbExecutor, schemaTaskExecutor, + olapTaskExecutor, gremlinTaskExecutor, ephemeralTaskExecutor, + serverInfoDbExecutor) { + + @Override + protected boolean updateStatus(Id id, TaskStatus prestatus, + TaskStatus status) { + return false; + } + + @Override + protected <V> HugeTask<V> taskWithoutResult(Id id) { + throw new NotFoundException("Can't find task with id '%s'", id); + } + }; + Id id = IdGenerator.of(999997); + HugeTask<?> task = new HugeTask<>(id, null, new TaskCallable<Object>() { + @Override + public Object call() { + return null; + } + }); + task.type("test"); + task.name("missing-retry-task"); + + scheduler.cancel(task); + } finally { + schedulerExecutor.shutdownNow(); + taskDbExecutor.shutdownNow(); + schemaTaskExecutor.shutdownNow(); + olapTaskExecutor.shutdownNow(); + gremlinTaskExecutor.shutdownNow(); + ephemeralTaskExecutor.shutdownNow(); + serverInfoDbExecutor.shutdownNow(); + } + } + @Test public void testGraphManagerDoesNotGenerateServerIdWhenElectionDisabled() { HugeConfig config = newConfig();
