From 198ff63c47651fe4a749da94eaab01efa28ff779 Mon Sep 17 00:00:00 2001 From: imbajin Date: Tue, 13 Oct 2026 01:07:04 +0800 Subject: [PATCH 1/5] fix(core): drain tasks before graph deletion - fence admission and track actual execution owners - wait for callbacks and transaction cleanup before deletion - retain graphs and schedulers when draining fails - keep core regressions and defer reload coordination --- docs/server-runtime.md | 17 + .../apache/hugegraph/core/GraphManager.java | 87 ++- .../apache/hugegraph/StandardHugeGraph.java | 20 +- .../task/DistributedTaskScheduler.java | 318 ++++++++++- .../hugegraph/task/StandardTaskScheduler.java | 170 +++++- .../apache/hugegraph/task/TaskManager.java | 27 +- .../DistributedTaskAdmissionTest.java | 521 ++++++++++++++++++ .../hugegraph/GraphDropPendingTest.java | 330 +++++++++++ .../org/apache/hugegraph/HugeFactoryTest.java | 21 +- .../hugegraph/StandardTaskAdmissionTest.java | 331 +++++++++++ .../apache/hugegraph/unit/UnitTestSuite.java | 6 + .../unit/api/space/GraphSpaceAPITest.java | 26 +- 12 files changed, 1730 insertions(+), 144 deletions(-) create mode 100644 hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java create mode 100644 hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java create mode 100644 hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java diff --git a/docs/server-runtime.md b/docs/server-runtime.md index d35b6b34ab..ee17e4287f 100644 --- a/docs/server-runtime.md +++ b/docs/server-runtime.md @@ -463,3 +463,20 @@ Structure or Process suites must execute tests; an all-skipped report is not compatibility evidence. Run the read-only application examples against representative existing data and validate the serializers, client versions and authentication combinations actually used by the application. + + +## Deleting a graph with local tasks + +Stop application writes before deletion and keep them stopped during retries. The local +scheduler closes task admission and waits for actual executions, cancellation callbacks +and their transaction cleanup before data or metadata is removed. If draining times out, +the graph stays registered with its data and configuration; retry after its owners exit. +Closing the scheduler alone leaves persistent work that has not started in storage. + +This is a local shutdown guarantee. For a graph shared by several Servers, stop writes +and tasks on every Server before deletion; notification delivery is not a cluster-wide +barrier. Do not recreate the same graph name until deletion finishes. Use a restart for +configuration changes rather than hot reload, whose recovery is handled separately. +Deletion is not transactional: a failure after destructive cleanup needs operator +inspection, and retry does not restore removed data. An owner-cleanup failure must not +be treated as a successful drain; inspect it instead of force-removing the scheduler. diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java index 1a8c1b5d1e..fd4996890a 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java @@ -36,6 +36,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -114,6 +115,7 @@ import org.apache.hugegraph.space.register.dto.ServiceDTO; import org.apache.hugegraph.space.register.registerImpl.PdRegister; import org.apache.hugegraph.task.TaskManager; +import org.apache.hugegraph.task.TaskScheduler; import org.apache.hugegraph.testutil.Whitebox; import org.apache.hugegraph.traversal.optimize.HugeScriptTraversal; import org.apache.hugegraph.type.define.CollectionType; @@ -1107,23 +1109,21 @@ public void getSpaceStorage(String graphSpace) { } public void clearGraphSpace(String name) { - // Clear all roles - this.metaManager.clearGraphAuth(name); - - // Clear all schemaTemplate - this.metaManager.clearSchemaTemplate(name); - // Clear all graphs for (String key : this.graphs.keySet()) { - if (key.startsWith(name)) { + if (key.startsWith(name + DELIMITER)) { String[] parts = key.split(DELIMITER); this.dropGraph(parts[0], parts[1], true); } } + // Shared metadata remains available if a graph refuses deletion. + this.metaManager.clearGraphAuth(name); + this.metaManager.clearSchemaTemplate(name); + // Clear all services for (String key : this.services.keySet()) { - if (key.startsWith(name)) { + if (key.startsWith(name + DELIMITER)) { String[] parts = key.split(DELIMITER); this.dropService(parts[0], parts[1]); } @@ -1457,12 +1457,13 @@ private HugeGraph createGraphLocal(HugeConfig config, String name) { LOG.error("Failed to create graph '{}' due to: {}", name, e.getMessage(), e); if (graph != null) { - this.graphs.remove(graph.spaceGraphName(), graph); try { this.dropGraphLocal(graph); - } finally { - // The create event may have partially registered the graph + this.graphs.remove(graph.spaceGraphName(), graph); this.notifyEventLenient(Events.GRAPH_DROP, graph); + } catch (Throwable cleanup) { + e.addSuppressed(cleanup); + this.graphs.putIfAbsent(graph.spaceGraphName(), graph); } } throw e; @@ -1472,16 +1473,8 @@ private HugeGraph createGraphLocal(HugeConfig config, String name) { } private void dropGraphLocal(HugeGraph graph) { - try { - // Clear data and config files - graph.drop(); - } finally { - /* - * Will fill graph instance into HugeFactory.graphs after - * GraphFactory.open() succeed, remove it when the graph drops - */ - HugeFactory.remove(graph); - } + graph.drop(); + HugeFactory.remove(graph); } public HugeGraph createGraph(String graphSpace, String name, String creator, @@ -1614,16 +1607,14 @@ public HugeGraph createGraph(String graphSpace, String name, String creator, try { this.notifyEvent(Events.GRAPH_CREATE, graph); } catch (Throwable e) { - this.notifyEventLenient(Events.GRAPH_DROP, graph); - this.graphs.remove(graphName, graph); try { graph.close(); + this.notifyEventLenient(Events.GRAPH_DROP, graph); + this.graphs.remove(graphName, graph); + HugeFactory.remove(graph); } catch (Exception e1) { - if (graph instanceof StandardHugeGraph) { - ((StandardHugeGraph) graph).clearSchedulerAndLock(); - } + e.addSuppressed(e1); } - HugeFactory.remove(graph); throw e; } @@ -2086,12 +2077,11 @@ private HugeGraph createGraph(String graphSpace, HugeConfig config, } catch (BackendException e) { try { graph.close(); + HugeFactory.remove(graph); } catch (Exception e1) { - if (graph instanceof StandardHugeGraph) { - ((StandardHugeGraph) graph).clearSchedulerAndLock(); - } + e.addSuppressed(e1); + this.graphs.putIfAbsent(graph.spaceGraphName(), graph); } - HugeFactory.remove(graph); throw e; } } @@ -2269,6 +2259,18 @@ public void dropGraph(String graphSpace, String name, boolean clear) { String graphName = spaceGraphName(graphSpace, name); if (clear) { + // The auth proxy checks ADMIN before the local deletion fence. + TaskScheduler scheduler = g.taskScheduler(); + if (!scheduler.close()) { + try { + scheduler.waitUntilAllTasksCompleted(g.configuration().get(CoreOptions.TASK_WAIT_TIMEOUT)); + } catch (TimeoutException e) { + throw new HugeException("Tasks are still active in graph '%s', please retry later", + e, graphName); + } + E.checkState(scheduler.close(), "Tasks are still active in graph '%s', please retry later", + graphName); + } this.removingGraphs.add(graphName); try { this.metaManager.removeGraphConfig(graphSpace, name); @@ -2278,24 +2280,13 @@ public void dropGraph(String graphSpace, String name, boolean clear) { "Failed to remove graph config of '%s'", name, e); } - /** - * close task scheduler before clear data, - * because taskinfo stored in backend in - * {@link org.apache.hugegraph.task.DistributedTaskScheduler} - */ - try { - g.taskScheduler().close(); - } catch (Throwable t) { - LOG.warn(String.format( - "Error when close TaskScheduler of %s", - graphName), - t); - } - g.clearBackend(); try { g.close(); } catch (Exception e) { + if (!g.closed()) { + throw new HugeException("Failed to close graph '%s', please retry later", e, graphName); + } LOG.warn("Failed to close graph", e); } } @@ -2309,13 +2300,17 @@ public void dropGraph(String graphSpace, String name, boolean clear) { } // Let gremlin server and rest server context remove graph LOG.info("Notify remove graph {} by GRAPH_DROP event", name); - Graph graph = this.graphs.remove(graphName); + Graph graph = this.graphs.get(graphName); if (graph != null) { try { graph.close(); } catch (Exception e) { + if (!((HugeGraph) graph).closed()) { + throw new HugeException("Failed to close graph '%s', please retry later", e, graphName); + } LOG.warn("Failed to close graph", e); } + this.graphs.remove(graphName, graph); try { // Delete alias graph in HugeFactory HugeFactory.remove((HugeGraph) graph); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java index 4649dd1fc9..18416c07eb 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java @@ -537,6 +537,7 @@ public void initBackend() { @Override public void clearBackend() { + this.checkGraphNotClosed(); this.waitUntilAllTasksCompleted(); this.loadSchemaStore().open(this.configuration); @@ -1212,12 +1213,16 @@ public void create(String configPath, GlobalMasterInfo nodeInfo) { @Override public void drop() { + this.checkGraphNotClosed(); + TaskScheduler scheduler = this.taskManager.getScheduler(this.params); + if (scheduler != null && !scheduler.close()) { + // Fence admission before waiting for actual task owners to exit. + this.waitUntilAllTasksCompleted(); + E.checkState(scheduler.close(), "Tasks are still active in graph '%s', please retry later", + this.spaceGraphName()); + } this.clearBackend(); - HugeConfig config = this.configuration(); - this.storeProvider.onDeleteConfig(config); - ConfigUtil.deleteFile(config.file()); - try { /* * It's hard to ensure all threads close the tx. @@ -1228,8 +1233,15 @@ public void drop() { */ this.close(); } catch (Throwable e) { + if (!this.closed()) { + throw new HugeException("Failed to close graph '%s', please retry later", + e, this.spaceGraphName()); + } LOG.warn("Failed to close graph {} {}", this, e); } + HugeConfig config = this.configuration(); + this.storeProvider.onDeleteConfig(config); + ConfigUtil.deleteFile(config.file()); } @Override 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 d89a32040f..f29224bc2a 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 @@ -25,6 +25,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; @@ -32,6 +33,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.hugegraph.exception.HugeException; import org.apache.hugegraph.HugeGraph; @@ -67,8 +69,11 @@ public class DistributedTaskScheduler extends TaskAndResultScheduler { */ private final AtomicBoolean closed = new AtomicBoolean(true); private volatile boolean closeCompleted; + private final AtomicInteger cancellations = new AtomicInteger(); private final ConcurrentHashMap> runningTasks = new ConcurrentHashMap<>(); + // Local execution owners, including accepted work that has not acquired its task lock. + private final ConcurrentHashMap executions = new ConcurrentHashMap<>(); private final Set deletingTasks = ConcurrentHashMap.newKeySet(); public DistributedTaskScheduler(HugeGraphParams graph, @@ -120,7 +125,7 @@ private static boolean sleep(long ms) { } } - public void cronSchedule() { + public synchronized void cronSchedule() { // Perform periodic scheduling tasks // Check closed flag first to exit early @@ -194,7 +199,8 @@ public void cronSchedule() { initTaskParams(cancelling); LOG.info("Try to cancel task({})@({}/{})", cancelling.id(), this.graphSpace, this.graphName); - if (!cancelling.cancel(true)) { + if (!this.cancelOwned(this.executions.get(cancellingId), + () -> cancelling.cancel(true), true)) { // Task already completed normally; force CANCELLED so // it doesn't stay stuck in CANCELLING forever. updateStatusWithLock(cancellingId, TaskStatus.CANCELLING, @@ -217,8 +223,10 @@ public void cronSchedule() { Id deletingId = deletings.next().id(); HugeTask deleting = runningTasks.get(deletingId); if (deleting != null) { - this.markTaskDeleting(deleting); - deleting.cancel(true); + this.cancelOwned(this.executions.get(deletingId), () -> { + this.markTaskDeleting(deleting); + return deleting.cancel(true); + }, true); } else { // Local has no task execution, but the current task has no nodes executing anymore. if (!isLockedTask(deletingId.toString())) { @@ -242,7 +250,7 @@ public HugeGraph graph() { @Override public int pendingTasks() { - return this.runningTasks.size(); + return this.executions.size() + this.cancellations.get(); } @Override @@ -251,14 +259,21 @@ public void restoreTasks() { } @Override - public Future schedule(HugeTask task) { + public synchronized Future schedule(HugeTask task) { E.checkArgumentNotNull(task, "Task can't be null"); + E.checkState(!this.closed.get(), "TaskScheduler has closed"); + PendingTask pending = this.executions.get(task.id()); + if (pending != null && pending.task == task && task.status() == TaskStatus.NEW && + pending.started.get() && !pending.ownerFinished) { + // HugeTask checks dependencies before running and may reschedule itself. + return pending; + } initTaskParams(task); if (task.ephemeralTask()) { // Handle ephemeral tasks, no scheduling needed, execute directly - return this.ephemeralTaskExecutor.submit(task); + return this.submitTask(task, this.ephemeralTaskExecutor, task); } // Validate task state before saving to ensure correct exception type @@ -305,6 +320,59 @@ protected void initTaskParams(HugeTask task) { @Override public void cancel(HugeTask task) { E.checkArgumentNotNull(task, "Task can't be null"); + PendingTask pending; + synchronized (this) { + E.checkState(!this.closed.get(), "TaskScheduler has closed"); + pending = this.executions.get(task.id()); + if (pending != null) { + synchronized (pending) { + pending.callbacks++; + } + } + this.cancellations.incrementAndGet(); + } + String context = TaskManager.getContext(); + AtomicBoolean cleaned = new AtomicBoolean(true); + TaskManager.ContextCallable owned = new TaskManager.ContextCallable<>(() -> { + try { + this.cancelTask(task, true); + return null; + } finally { + try { + this.graph.closeTx(); + } catch (RuntimeException | Error error) { + cleaned.set(false); + throw error; + } + } + }); + try { + // Keep the original synchronous callback semantics, including nested cancel requests. + owned.call(); + } catch (RuntimeException | Error error) { + throw error; + } catch (Exception e) { + throw new HugeException("Failed to cancel task '%s'", e, task.id()); + } finally { + TaskManager.setContext(context); + if (cleaned.get()) { + this.releaseCancellation(pending); + } + } + } + + private void releaseCancellation(PendingTask pending) { + if (pending != null) { + synchronized (pending) { + pending.callbacks--; + pending.release(); + } + } + this.cancellations.decrementAndGet(); + } + + private void cancelTask(HugeTask task, boolean direct) { + E.checkArgumentNotNull(task, "Task can't be null"); if (task.completed() || task.cancelling()) { return; @@ -314,12 +382,19 @@ public void cancel(HugeTask task) { // Check if task is running locally, cancel it directly if so HugeTask runningTask = this.runningTasks.get(task.id()); - if (runningTask != null) { - boolean cancelled = runningTask.cancel(true); - if (cancelled) { - task.overwriteStatus(TaskStatus.CANCELLED); - this.save(runningTask); - } + PendingTask pending = this.executions.get(task.id()); + if (runningTask != null || task.ephemeralTask()) { + HugeTask local = runningTask != null ? runningTask : task; + Callable cancellation = () -> { + boolean accepted = local.cancel(true); + if (accepted && !local.ephemeralTask()) { + task.overwriteStatus(TaskStatus.CANCELLED); + this.save(local); + } + return accepted; + }; + boolean cancelled = pending == null ? local.cancel(true) : + this.cancelOwned(pending, cancellation, direct); LOG.info("Cancel local running task '{}' result: {}", task.id(), cancelled); return; } @@ -475,8 +550,10 @@ public HugeTask delete(Id id, boolean force) { HugeTask running = this.runningTasks.get(id); if (running != null) { - this.markTaskDeleting(running); - running.cancel(true); + this.cancelOwned(this.executions.get(id), () -> { + this.markTaskDeleting(running); + return running.cancel(true); + }, false); @SuppressWarnings("unchecked") HugeTask result = (HugeTask) running; return result; @@ -532,8 +609,10 @@ public boolean close() { return true; } - // set closed - this.closed.set(true); + // Fence schedule() and cron admission before starting the drain. + synchronized (this) { + this.closed.set(true); + } // cancel cron thread if (!cronFuture.isDone() && !cronFuture.isCancelled()) { @@ -557,17 +636,22 @@ public boolean close() { LOG.warn("Exception while waiting for cron task to complete", e); return false; } catch (RejectedExecutionException e) { - if (!this.schedulerExecutor.isTerminated()) { - LOG.warn("Scheduler executor has not drained when closing scheduler", e); - return false; - } + LOG.warn("Scheduler executor was stopped before its graphs drained", e); + return false; } // cancel all running tasks - for (HugeTask task : this.runningTasks.values()) { + for (PendingTask pending : this.executions.values()) { + // Durable work that never ran stays NEW for a later scheduler. + if (!pending.task.ephemeralTask() && + !this.runningTasks.containsKey(pending.task.id())) { + pending.cancelBeforeStart(); + // A wrapper already entering its lock/read phase cleans up without running the job. + continue; + } LOG.info("cancel task({}) @({}/{}) when closing scheduler", - task.id(), graphSpace, graphName); - this.cancel(task); + pending.task.id(), graphSpace, graphName); + this.cancelTask(pending.task, false); } try { @@ -767,7 +851,10 @@ protected boolean updateStatusWithLock(Id id, TaskStatus prestatus, * @param task * @return true if the task have start */ - private boolean tryStartHugeTask(HugeTask task) { + private synchronized boolean tryStartHugeTask(HugeTask task) { + if (this.closed.get() || this.executions.containsKey(task.id())) { + return false; + } // Print Scheduler status logCurrentState(); @@ -795,7 +882,7 @@ private boolean tryStartHugeTask(HugeTask task) { ThreadPoolExecutor executor = (ThreadPoolExecutor) chosenExecutor; if (executor.getActiveCount() < executor.getMaximumPoolSize()) { TaskRunner runner = new TaskRunner<>(task); - chosenExecutor.submit(runner); + this.submitTask(task, chosenExecutor, runner); LOG.info("Submit task({})@({}/{})", task.id(), this.graphSpace, this.graphName); @@ -805,6 +892,176 @@ private boolean tryStartHugeTask(HugeTask task) { return false; } + private Future submitTask(HugeTask task, ExecutorService executor, Runnable work) { + TaskManager.ContextCallable owned = new TaskManager.ContextCallable<>(() -> { + try { + work.run(); + return null; + } finally { + this.closeOwner(this.executions.get(task.id())); + } + }); + PendingTask pending = new PendingTask(task, executor, owned); + E.checkState(this.executions.putIfAbsent(task.id(), pending) == null, + "Task '%s' is already pending", task.id()); + try { + executor.execute(pending); + return pending; + } catch (RuntimeException | Error error) { + pending.cancelBeforeStart(); + throw error; + } + } + + private boolean cancelOwned(PendingTask pending, Callable action, boolean direct) { + if (pending == null) { + return false; + } + synchronized (pending) { + if (this.executions.get(pending.task.id()) != pending) { + return false; + } + pending.callbacks++; + } + TaskManager.ContextCallable owned = new TaskManager.ContextCallable<>(() -> { + try { + return action.call(); + } finally { + this.closeOwner(pending); + } + }); + Callable cancellation = () -> { + try { + return owned.call(); + } finally { + synchronized (pending) { + pending.callbacks--; + if (pending.task.isDone()) { + pending.cancelBeforeStart(); + } + pending.release(); + } + } + }; + if (direct) { + try { + return cancellation.call(); + } catch (Exception e) { + throw new HugeException("Failed to cancel task '%s'", e, pending.task.id()); + } + } + Future future; + try { + future = this.schedulerExecutor.submit(cancellation); + } catch (RuntimeException | Error error) { + synchronized (pending) { + pending.callbacks--; + pending.release(); + } + throw error; + } + try { + return future.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new HugeException("Interrupted when cancelling task '%s'", e, pending.task.id()); + } catch (ExecutionException e) { + throw new HugeException("Failed to cancel task '%s'", e, pending.task.id()); + } + } + + private void closeOwner(PendingTask pending) { + try { + this.graph.closeTx(); + } catch (RuntimeException | Error error) { + synchronized (pending) { + // A failed owner cleanup must never become a successful graph drain. + pending.cleanupFailed = true; + } + throw error; + } + } + + private class PendingTask extends FutureTask { + + private final HugeTask task; + private final ExecutorService executor; + private final AtomicBoolean started = new AtomicBoolean(); + private volatile boolean ownerFinished; + private int callbacks; + private boolean cleanupFailed; + + PendingTask(HugeTask task, ExecutorService executor, Callable work) { + super(work); + this.task = task; + this.executor = executor; + } + + @Override + public void run() { + if (!this.started.compareAndSet(false, true)) { + return; + } + try { + if (this.task.ephemeralTask()) { + if (super.runAndReset() && this.task.completed()) { + this.set(null); + } + } else { + super.run(); + } + } finally { + synchronized (this) { + this.ownerFinished = true; + if (this.task.ephemeralTask() && !this.task.completed() && + !this.isDone() && !closed.get() && this.callbacks == 0) { + this.started.set(false); + this.ownerFinished = false; + try { + this.executor.execute(this); + } catch (RuntimeException | Error error) { + this.ownerFinished = true; + this.setException(error); + } + } + // A dependency retry fenced by close still needs its cancellation callback. + if (!this.task.ephemeralTask() || this.task.completed() || !closed.get()) { + this.release(); + } + } + } + } + + @Override + protected void done() { + if (this.isCancelled()) { + this.cancelBeforeStart(); + } + } + + private synchronized boolean cancelBeforeStart() { + if (!this.started.compareAndSet(false, true)) { + return false; + } + this.cancel(false); + if (this.executor instanceof ThreadPoolExecutor) { + ((ThreadPoolExecutor) this.executor).remove(this); + } + this.ownerFinished = true; + this.release(); + return true; + } + + private void release() { + if (this.ownerFinished && this.callbacks == 0 && !this.cleanupFailed) { + executions.remove(this.task.id(), this); + if (this.task.completed() && !this.isDone()) { + this.set(null); + } + } + } + } + protected void logCurrentState() { int gremlinActive = ((ThreadPoolExecutor) gremlinTaskExecutor).getActiveCount(); @@ -894,7 +1151,12 @@ public void run() { return; } - runningTasks.put(task.id(), task); + synchronized (DistributedTaskScheduler.this) { + if (closed.get()) { + return; + } + runningTasks.put(task.id(), task); + } // Task execution will not throw exceptions, HugeTask will catch exceptions // during execution and store them in the DB. @@ -902,7 +1164,7 @@ public void run() { } catch (Throwable t) { LOG.warn("exception when execute task", t); } finally { - runningTasks.remove(task.id()); + runningTasks.remove(task.id(), task); deletingTasks.remove(task.id()); unlockTask(task.id().asString(), lockResult); 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 15249067bc..8d91069553 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 @@ -26,7 +26,10 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.hugegraph.exception.HugeException; import org.apache.hugegraph.HugeGraph; @@ -69,8 +72,11 @@ public class StandardTaskScheduler implements TaskScheduler { private final ExecutorService taskDbExecutor; private final Map> tasks; + private final Map> executions; private volatile TaskTransaction taskTx; + private boolean admissionClosed; + private int cancellations; public StandardTaskScheduler(HugeGraphParams graph, ExecutorService taskExecutor, @@ -85,6 +91,7 @@ public StandardTaskScheduler(HugeGraphParams graph, this.serverManager = new ServerInfoManager(graph); this.tasks = new ConcurrentHashMap<>(); + this.executions = new ConcurrentHashMap<>(); this.taskTx = null; } @@ -115,8 +122,8 @@ public String spaceGraphName() { } @Override - public int pendingTasks() { - return this.tasks.size(); + public synchronized int pendingTasks() { + return this.tasks.size() + this.cancellations; } private TaskTransaction tx() { @@ -141,7 +148,8 @@ private TaskTransaction tx() { } @Override - public void restoreTasks() { + public synchronized void restoreTasks() { + this.checkAdmission(); List> taskList = new ArrayList<>(); // Single-node mode: restore pending tasks without server filtering. // Don't restore legacy SCHEDULING/SCHEDULED tasks globally; they were @@ -184,7 +192,8 @@ private Future restore(HugeTask task) { } @Override - public Future schedule(HugeTask task) { + public synchronized Future schedule(HugeTask task) { + this.checkAdmission(); E.checkArgumentNotNull(task, "Task can't be null"); /* @@ -230,7 +239,7 @@ private Future submitTask(HugeTask task, boolean saveAfterQueued) { throw e; } } - return this.taskExecutor.submit(task); + return this.submitExecution(task); } private Future resubmitTask(HugeTask task) { @@ -240,7 +249,88 @@ private Future resubmitTask(HugeTask task) { E.checkArgument(this.tasks.containsKey(task.id()), "Can't resubmit task '%s' not been submitted before", task.id()); - return this.taskExecutor.submit(task); + return this.submitExecution(task); + } + + private Future submitExecution(HugeTask task) { + PendingTask pending = new PendingTask(task); + this.executions.computeIfAbsent(task.id(), id -> new ArrayList<>()).add(pending); + try { + this.taskExecutor.execute(pending); + } catch (RuntimeException | Error error) { + this.finishExecution(pending); + throw error; + } + return pending; + } + + private synchronized void finishExecution(PendingTask pending) { + List outstanding = this.executions.get(pending.task.id()); + if (outstanding != null && outstanding.remove(pending) && outstanding.isEmpty()) { + this.executions.remove(pending.task.id()); + this.tasks.remove(pending.task.id(), pending.task); + } + } + + private synchronized void cancelQueuedExecutions(Id id) { + List outstanding = this.executions.get(id); + if (outstanding != null) { + for (PendingTask pending : new ArrayList<>(outstanding)) { + pending.cancelBeforeStart(); + } + } + } + + private final class PendingTask extends FutureTask { + + private final HugeTask task; + private final AtomicBoolean started = new AtomicBoolean(); + + private PendingTask(HugeTask task) { + super(new ContextCallable<>(() -> { + task.run(); + return null; + })); + this.task = task; + } + + @Override + public void run() { + if (!this.started.compareAndSet(false, true)) { + return; + } + try { + super.run(); + } finally { + // A cleanup failure must retain this owner and prevent close ACK. + StandardTaskScheduler.this.graph.closeTx(); + StandardTaskScheduler.this.finishExecution(this); + } + } + + @Override + protected void done() { + if (this.isCancelled()) { + // Cancelling the dispatch Future must not run HugeTask callbacks + // on its caller or cancel a later dependency retry. + this.cancelBeforeStart(); + } + } + + private void cancelBeforeStart() { + if (this.started.compareAndSet(false, true)) { + super.cancel(false); + if (StandardTaskScheduler.this.taskExecutor instanceof ThreadPoolExecutor) { + ((ThreadPoolExecutor) StandardTaskScheduler.this.taskExecutor).remove(this); + } + StandardTaskScheduler.this.finishExecution(this); + } + } + } + + private void checkAdmission() { + E.checkState(!this.admissionClosed, "Task scheduler for graph '%s' is closing", + this.spaceGraphName()); } public void initTaskCallable(HugeTask task) { @@ -256,32 +346,45 @@ public void initTaskCallable(HugeTask task) { } @Override - public synchronized void cancel(HugeTask task) { + public void cancel(HugeTask task) { E.checkArgumentNotNull(task, "Task can't be null"); - - if (task.completed() || task.cancelling()) { - return; - } - - LOG.info("Cancel task '{}' in status {}", task.id(), task.status()); - - HugeTask memTask = this.tasks.get(task.id()); - if (memTask != null) { - boolean cancelled = memTask.cancel(true); - if (cancelled) { - task.overwriteStatus(TaskStatus.CANCELLED); + HugeTask memTask; + synchronized (this) { + if (task.completed() || task.cancelling()) { + return; } - LOG.info("Task '{}' cancel result: {}", task.id(), cancelled); - return; + memTask = this.tasks.get(task.id()); + if (memTask == null) { + this.checkAdmission(); + } + this.cancellations++; } - - if (task.status(TaskStatus.CANCELLED)) { - this.save(task); - return; + // HugeTask.cancel() runs done/cancelled callbacks on this caller. + // Do not hold the scheduler monitor while they wait for storage/workers. + try { + if (memTask != null) { + boolean cancelled = memTask.cancel(true); + if (cancelled) { + task.overwriteStatus(TaskStatus.CANCELLED); + } + return; + } + if (task.status(TaskStatus.CANCELLED)) { + this.save(task); + return; + } + throw new HugeException("Can't cancel task '%s' in status %s", + task.id(), task.status()); + } finally { + if (memTask != null && memTask.isCancelled()) { + this.cancelQueuedExecutions(task.id()); + } + // Keep the callback registered until its transaction really closes. + this.graph.closeTx(); + synchronized (this) { + this.cancellations--; + } } - - throw new HugeException("Can't cancel task '%s' in status %s", - task.id(), task.status()); } @Override @@ -290,7 +393,10 @@ public ServerInfoManager serverManager() { } @Override public void taskDone(HugeTask task) { - this.remove(task); + // Logical completion/cancellation can precede the worker's final cleanup. + if (!this.executions.containsKey(task.id())) { + this.remove(task); + } // Single-node mode: no need to manage load LOG.debug("Task '{}' done", task.id()); } @@ -301,6 +407,9 @@ protected void remove(HugeTask task) { protected void remove(HugeTask task, boolean force) { E.checkNotNull(task, "remove task"); + if (this.executions.containsKey(task.id())) { + return; + } HugeTask delTask = this.tasks.remove(task.id()); if (delTask != null && delTask != task) { LOG.warn("Task '{}' may be inconsistent status {}(expect {})", @@ -331,7 +440,8 @@ public void init() { } @Override - public boolean close() { + public synchronized boolean close() { + this.admissionClosed = true; // Running tasks still need the task DB transaction to persist done(). // Retain the scheduler and its owners until a later close attempt. if (this.pendingTasks() != 0) { diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java index 0342eb63bc..6ad1ebf9e1 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java @@ -127,28 +127,17 @@ public void closeScheduler(HugeGraphParams graph) { Throwable failure = null; boolean drained = scheduler == null; if (scheduler != null) { - /* - * Keep close+remove exclusive with scheduler iteration: in their gap - * a scheduler DB worker could otherwise reopen the graph transaction. - */ - synchronized (scheduler) { - boolean stopped = false; - try { - stopped = scheduler.close(); - } catch (RuntimeException | Error error) { - failure = error; - // Standard closes its server manager in finally; Distributed - // sets its dispatch-stop flag before cancellation/drain begins. - stopped = true; - } - // A timeout leaves running jobs registered for the next drain attempt. - if (stopped && scheduler.pendingTasks() == 0) { + try { + // Distributed close waits for cron without holding its monitor. + if (scheduler.close() && scheduler.pendingTasks() == 0) { this.schedulers.remove(graph, scheduler); drained = true; } + } catch (RuntimeException | Error error) { + failure = error; } } - if (!drained) { + if (!drained && (failure == null || scheduler.pendingTasks() != 0)) { // A running cron/job must finish before owner callbacks are queued // on its executor. In particular, do not rejoin a timed-out cron. if (failure instanceof Error) { @@ -159,6 +148,7 @@ public void closeScheduler(HugeGraphParams graph) { } return; } + // An owner-close failure retains the scheduler, but idle workers can still clean up. for (Runnable close : new Runnable[]{() -> { if (!this.taskExecutor.isTerminated()) { this.closeTaskTx(graph); @@ -261,7 +251,8 @@ public ServerInfoManager getServerInfoManager(HugeGraphParams graph) { } public void shutdown(long timeout) { - assert this.schedulers.isEmpty() : this.schedulers.size(); + E.checkState(this.schedulers.isEmpty(), + "Task schedulers are still registered: %s", this.schedulers.size()); Throwable ex = null; boolean terminated = this.distributedSchedulerExecutor.isTerminated(); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java new file mode 100644 index 0000000000..c587994473 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java @@ -0,0 +1,521 @@ +/* + * 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.hugegraph; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.dist.RegisterUtil; +import org.apache.hugegraph.id.Id; +import org.apache.hugegraph.id.IdGenerator; +import org.apache.hugegraph.meta.MetaManager; +import org.apache.hugegraph.meta.lock.LockResult; +import org.apache.hugegraph.meta.managers.TaskMetaManager; +import org.apache.hugegraph.job.EphemeralJob; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; +import org.apache.hugegraph.task.DistributedTaskScheduler; +import org.apache.hugegraph.task.HugeTask; +import org.apache.hugegraph.task.TaskAndResultSchedulerTest.EmptyCallable; +import org.apache.hugegraph.task.TaskCallable; +import org.apache.hugegraph.task.TaskManager; +import org.apache.hugegraph.task.TaskScheduler; +import org.apache.hugegraph.task.TaskStatus; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.unit.FakeObjects; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +public class DistributedTaskAdmissionTest { + + private TaskMetaManager originalLocks; + + @Before + public void prepareLocks() { + this.originalLocks = Whitebox.getInternalState(MetaManager.instance(), "taskMetaManager"); + TaskMetaManager locks = Mockito.mock(TaskMetaManager.class); + Mockito.when(locks.tryLockTask(Mockito.anyString(), Mockito.anyString(), Mockito.anyString())) + .thenAnswer(invocation -> { + LockResult acquired = new LockResult(); + acquired.lockSuccess(true); + return acquired; + }); + Whitebox.setInternalState(MetaManager.instance(), "taskMetaManager", locks); + } + + @After + public void restoreLocks() { + Whitebox.setInternalState(MetaManager.instance(), "taskMetaManager", this.originalLocks); + } + + @Test + public void testCloseDuringPersistentRegistrationPreservesNewTask() throws Exception { + try (Fixture fixture = new Fixture("register_new")) { + CountDownLatch reading = new CountDownLatch(1); + CountDownLatch finishRead = new CountDownLatch(1); + fixture.scheduler.beforeTaskRead = () -> { + reading.countDown(); + awaitUninterruptibly(finishRead); + }; + HugeTask task = task(false, new AtomicBoolean(), 9999917L); + ExecutorService requests = Executors.newSingleThreadExecutor(); + try { + fixture.scheduler.schedule(task); + Assert.assertTrue(reading.await(10L, TimeUnit.SECONDS)); + Future close = requests.submit(fixture.scheduler::close); + AtomicBoolean closed = Whitebox.getInternalState(fixture.scheduler, "closed"); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10L); + while (!closed.get() && System.nanoTime() < deadline) { + Thread.yield(); + } + Assert.assertTrue(closed.get()); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + Assert.assertFalse(close.isDone()); + finishRead.countDown(); + Assert.assertTrue(close.get(10L, TimeUnit.SECONDS)); + fixture.scheduler.beforeTaskRead = () -> { }; + Assert.assertEquals(TaskStatus.NEW, fixture.scheduler.task(task.id(), false).status()); + Assert.assertEquals(TaskStatus.NEW, task.status()); + fixture.assertWorkerClean(); + } finally { + finishRead.countDown(); + stop(requests); + } + } + } + + @Test + public void testAcceptedRemoteCancellationPreventsEarlyDrain() throws Exception { + try (Fixture fixture = new Fixture("cancel_gate")) { + CountDownLatch busy = new CountDownLatch(1); + CountDownLatch finishWorker = new CountDownLatch(1); + Future occupied = fixture.worker.submit(() -> { + busy.countDown(); + awaitUninterruptibly(finishWorker); + }); + ExecutorService requests = Executors.newFixedThreadPool(2); + GatedTask task = new GatedTask(9999918L, new EmptyCallable()); + try { + Assert.assertTrue(busy.await(10L, TimeUnit.SECONDS)); + fixture.scheduler.schedule(task); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + task.armed.set(true); + Future cancel = requests.submit(() -> fixture.scheduler.cancel(task)); + Assert.assertTrue(task.reached.await(10L, TimeUnit.SECONDS)); + Future close = requests.submit(fixture.scheduler::close); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + Assert.assertFalse(close.isDone()); + task.release.countDown(); + cancel.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(close.get(10L, TimeUnit.SECONDS)); + Assert.assertEquals(TaskStatus.CANCELLING, fixture.scheduler.task(task.id(), false).status()); + } finally { + task.release.countDown(); + finishWorker.countDown(); + occupied.get(10L, TimeUnit.SECONDS); + stop(requests); + } + } + } + + @Test + public void testQueuedPersistentClosePreservesNewTask() throws Exception { + try (Fixture fixture = new Fixture("queued_new")) { + fixture.worker.armed.set(true); + HugeTask task = task(false, new AtomicBoolean(), 9999910L); + fixture.scheduler.schedule(task); + Assert.assertTrue(fixture.worker.reached.await(10L, TimeUnit.SECONDS)); + Assert.assertEquals(1, fixture.scheduler.pendingTasks()); + Assert.assertTrue(fixture.scheduler.close()); + Assert.assertFalse(task.isCancelled()); + Assert.assertEquals(TaskStatus.NEW, task.status()); + Assert.assertEquals(TaskStatus.NEW, fixture.scheduler.task(task.id(), false).status()); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + for (boolean ephemeral : new boolean[]{true, false}) { + Assert.assertThrows(IllegalStateException.class, + () -> fixture.scheduler.schedule(task(ephemeral, new AtomicBoolean(), 9999911L))); + } + } + } + + @Test + public void testBusyWorkerLeavesPersistentBacklogInDatabase() throws Exception { + try (Fixture fixture = new Fixture("busy_new")) { + CountDownLatch busy = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Future occupied = fixture.worker.submit(() -> { + busy.countDown(); + Assert.assertTrue(release.await(10L, TimeUnit.SECONDS)); + return null; + }); + try { + Assert.assertTrue(busy.await(10L, TimeUnit.SECONDS)); + HugeTask task = task(false, new AtomicBoolean(), 9999912L); + fixture.scheduler.schedule(task); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + Assert.assertTrue(fixture.worker.getQueue().isEmpty()); + Assert.assertEquals(TaskStatus.NEW, fixture.scheduler.task(task.id(), false).status()); + Assert.assertTrue(fixture.scheduler.close()); + } finally { + release.countDown(); + occupied.get(10L, TimeUnit.SECONDS); + } + } + } + + @Test + public void testQueuedOuterCancellationPreservesCallerTransaction() throws Exception { + try (Fixture fixture = new Fixture("queued_future")) { + fixture.worker.armed.set(true); + AtomicBoolean ran = new AtomicBoolean(); + HugeTask task = task(true, ran, 9999913L); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertTrue(fixture.worker.reached.await(10L, TimeUnit.SECONDS)); + fixture.graph.schema().vertexLabel("caller").useCustomizeStringId().create(); + fixture.graph.addVertex(org.apache.tinkerpop.gremlin.structure.T.id, "pending", + org.apache.tinkerpop.gremlin.structure.T.label, "caller"); + Assert.assertTrue(dispatch.cancel(false)); + Assert.assertTrue(fixture.graph.tx().isOpen()); + Assert.assertTrue(fixture.graph.vertices("pending").hasNext()); + fixture.graph.tx().rollback(); + Assert.assertFalse(task.isCancelled()); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + Assert.assertTrue(fixture.scheduler.close()); + fixture.worker.release.countDown(); + fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS); + Assert.assertFalse(ran.get()); + } + } + + @Test + public void testActiveOuterCancellationRetainsActualWorkerOwner() throws Exception { + try (Fixture fixture = new Fixture("active_future")) { + CountDownLatch started = new CountDownLatch(1); + CountDownLatch finish = new CountDownLatch(1); + HugeTask task = new HugeTask<>(IdGenerator.of(-9999914L), null, new EphemeralJob() { + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + this.graph().vertices().hasNext(); + started.countDown(); + awaitUninterruptibly(finish); + return null; + } + }); + task.type("test"); + task.name("active-owner"); + Future dispatch = fixture.scheduler.schedule(task); + ExecutorService requests = Executors.newSingleThreadExecutor(); + try { + Assert.assertTrue(started.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(dispatch.cancel(true)); + Future close = requests.submit(fixture.scheduler::close); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + Assert.assertFalse(close.isDone()); + finish.countDown(); + Assert.assertTrue(close.get(10L, TimeUnit.SECONDS)); + fixture.assertWorkerClean(); + } finally { + finish.countDown(); + stop(requests); + } + } + } + + @Test + public void testAcceptedLocalCancellationRetainsCallbackAndSupportsNestedCancel() throws Exception { + try (Fixture fixture = new Fixture("nested_callback")) { + fixture.worker.armed.set(true); + HugeTask other = task(true, new AtomicBoolean(), 9999919L); + GatedTask task = new GatedTask(-9999915L, new EphemeralJob() { + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + Assert.fail("Cancelled queued work must not execute"); + return null; + } + + @Override + protected void cancelled() { + fixture.scheduler.cancel(other); + } + }); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertTrue(fixture.worker.reached.await(10L, TimeUnit.SECONDS)); + fixture.scheduler.schedule(other); + task.armed.set(true); + ExecutorService requests = Executors.newSingleThreadExecutor(); + try { + Future cancel = requests.submit(() -> { + TaskManager.setContext("nested-cancel"); + try { + fixture.scheduler.cancel(task); + Assert.assertEquals("nested-cancel", TaskManager.getContext()); + } finally { + TaskManager.resetContext(); + } + }); + Assert.assertTrue(task.reached.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(dispatch.cancel(false)); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + task.release.countDown(); + cancel.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(task.isCancelled()); + Assert.assertTrue(other.isCancelled()); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + } finally { + task.release.countDown(); + stop(requests); + } + } + } + + @Test + public void testCallbackCleanupCompletesBeforeGraphDrain() throws Exception { + try (Fixture fixture = new Fixture("callback_owner")) { + fixture.worker.armed.set(true); + CountDownLatch callback = new CountDownLatch(1); + CountDownLatch finish = new CountDownLatch(1); + HugeTask task = new HugeTask<>(IdGenerator.of(-9999916L), null, new EphemeralJob() { + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + Assert.fail("Cancelled queued work must not execute"); + return null; + } + + @Override + protected void done() { + this.graph().vertices().hasNext(); + callback.countDown(); + awaitUninterruptibly(finish); + super.done(); + } + }); + task.type("test"); + task.name("callback-owner"); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertTrue(fixture.worker.reached.await(10L, TimeUnit.SECONDS)); + ExecutorService requests = Executors.newSingleThreadExecutor(); + ExecutorService callbacks = Executors.newSingleThreadExecutor(); + try { + Future cancel = callbacks.submit(() -> fixture.scheduler.cancel(task)); + Assert.assertTrue(callback.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(dispatch.cancel(true)); + Future close = requests.submit(fixture.scheduler::close); + Assert.assertTrue(fixture.scheduler.pendingTasks() > 0); + Assert.assertFalse(close.isDone()); + finish.countDown(); + cancel.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(close.get(10L, TimeUnit.SECONDS)); + ThreadLocal owners = Whitebox.getInternalState(fixture.graph.tx(), "transactions"); + Assert.assertNull(callbacks.submit(owners::get).get(10L, TimeUnit.SECONDS)); + Assert.assertNull(callbacks.submit(TaskManager::getContext).get(10L, TimeUnit.SECONDS)); + } finally { + finish.countDown(); + stop(requests); + stop(callbacks); + } + } + } + + private static void awaitUninterruptibly(CountDownLatch release) { + while (release.getCount() != 0L) { + try { + Assert.assertTrue(release.await(10L, TimeUnit.SECONDS)); + } catch (InterruptedException ignored) { + // Keep the real owner active after logical cancellation. + } + } + } + + private static void stop(ExecutorService executor) throws InterruptedException { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(10L, TimeUnit.SECONDS)); + } + + private static HugeTask task(boolean ephemeral, AtomicBoolean ran, long id) { + HugeTask task; + if (ephemeral) { + task = new HugeTask<>(IdGenerator.of(-id), null, new EphemeralJob() { + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + ran.set(true); + return null; + } + }); + } else { + task = new HugeTask<>(IdGenerator.of(id), null, new EmptyCallable()); + } + task.type("test"); + task.name("admission-test"); + return task; + } + + private static class Fixture implements AutoCloseable { + + private final HugeGraph graph; + private final ScheduledThreadPoolExecutor cron = new ScheduledThreadPoolExecutor(1); + private final ExecutorService database = Executors.newSingleThreadExecutor(); + private final GatedExecutor worker = new GatedExecutor(); + private final TestScheduler scheduler; + private final HugeGraphParams params; + private final TaskScheduler originalScheduler; + private final Map registry; + + Fixture(String suffix) { + RegisterUtil.registerBackends(); + HugeConfig config = FakeObjects.newConfig(); + config.setProperty("backend", "memory"); + config.setProperty("serializer", "text"); + config.setProperty("store", "distributed_admission_" + suffix); + config.setProperty("task.schedule_period", 600L); + config.setProperty("task.retry", 0); + this.graph = HugeFactory.open(config); + this.graph.initBackend(); + this.graph.serverStarted(GlobalMasterInfo.master("distributed-admission-test")); + this.params = Whitebox.getInternalState(this.graph, "params"); + this.registry = Whitebox.getInternalState(TaskManager.instance(), "schedulers"); + this.originalScheduler = this.registry.get(this.params); + Assert.assertTrue(this.originalScheduler.close()); + this.scheduler = new TestScheduler(this.params, this.cron, this.database, this.worker); + this.scheduler.init(); + Assert.assertTrue(this.registry.replace(this.params, this.originalScheduler, this.scheduler)); + } + + void assertWorkerClean() throws Exception { + this.worker.submit(() -> null).get(10L, TimeUnit.SECONDS); + Assert.assertEquals(0, this.scheduler.pendingTasks()); + ThreadLocal owners = Whitebox.getInternalState(this.graph.tx(), "transactions"); + Assert.assertNull(this.worker.submit(owners::get).get(10L, TimeUnit.SECONDS)); + Assert.assertNull(this.worker.submit(TaskManager::getContext).get(10L, TimeUnit.SECONDS)); + } + + @Override + public void close() throws Exception { + this.worker.release.countDown(); + try { + Assert.assertTrue(this.scheduler.close()); + this.scheduler.closeDatabaseOwner(); + } finally { + Assert.assertTrue(this.registry.replace(this.params, this.scheduler, this.originalScheduler)); + for (ExecutorService executor : new ExecutorService[]{this.worker, this.database, this.cron}) { + stop(executor); + } + try { + this.graph.close(); + } finally { + HugeFactory.remove(this.graph); + } + } + } + } + + private static class GatedTask extends HugeTask { + + private final AtomicBoolean armed = new AtomicBoolean(); + private final CountDownLatch reached = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + GatedTask(long id, TaskCallable callable) { + super(IdGenerator.of(id), null, callable); + this.type("test"); + this.name("accepted-cancel"); + } + + @Override + public boolean completed() { + if (this.armed.compareAndSet(true, false)) { + this.reached.countDown(); + awaitUninterruptibly(this.release); + } + return super.completed(); + } + } + + private static class GatedExecutor extends ThreadPoolExecutor { + + private final AtomicBoolean armed = new AtomicBoolean(); + private final CountDownLatch reached = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + GatedExecutor() { + super(1, 1, 0L, TimeUnit.SECONDS, new LinkedBlockingQueue<>()); + } + + @Override + protected void beforeExecute(Thread thread, Runnable work) { + if (this.armed.compareAndSet(true, false)) { + this.reached.countDown(); + awaitUninterruptibly(this.release); + } + } + } + + private static class TestScheduler extends DistributedTaskScheduler { + + private volatile Runnable beforeTaskRead = () -> { }; + + @Override + public HugeTask task(Id id, boolean withResult) { + this.beforeTaskRead.run(); + return super.task(id, withResult); + } + + TestScheduler(HugeGraphParams graph, ScheduledThreadPoolExecutor cron, + ExecutorService database, ExecutorService worker) { + super(graph, cron, database, worker, worker, worker, worker); + } + + void closeDatabaseOwner() { + this.call(() -> { + this.tx().close(); + this.graph.closeTx(); + return null; + }); + } + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java new file mode 100644 index 0000000000..9dac803821 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java @@ -0,0 +1,330 @@ +/* + * 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.hugegraph; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.commons.io.FileUtils; +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.dist.RegisterUtil; +import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.exception.HugeException; +import org.apache.hugegraph.id.IdGenerator; +import org.apache.hugegraph.job.SysJob; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; +import org.apache.hugegraph.meta.MetaManager; +import org.apache.hugegraph.space.GraphSpace; +import org.apache.hugegraph.task.HugeTask; +import org.apache.hugegraph.task.TaskScheduler; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.unit.FakeObjects; +import org.apache.hugegraph.util.Events; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.apache.tinkerpop.gremlin.structure.T; +import org.junit.Test; +import org.mockito.Mockito; + +public class GraphDropPendingTest { + + @Test + public void testLocalDropKeepsDataAndRegistrationUntilCancelledWorkerExits() throws Exception { + try (Fixture fixture = new Fixture("local_pending", false)) { + BlockingJob job = fixture.start(); + fixture.scheduler.cancel(job.task()); + Assert.assertTrue(job.task().isCancelled()); + HugeException refusal = Assert.assertThrows(HugeException.class, fixture::drop); + Assert.assertContains(fixture.pd ? "Tasks are still active" : "Failed to wait", refusal.getMessage()); + fixture.assertRetained(); + job.release.countDown(); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + fixture.drop(); + Assert.assertTrue(fixture.graph.closed()); + Assert.assertNull(fixture.manager.graph(fixture.graph.spaceGraphName())); + Assert.assertFalse(Files.exists(fixture.file)); + Assert.assertEquals(1, fixture.dropEvents); + } + } + + @Test + public void testPdDropPreservesMetadataUntilLocalTasksDrain() throws Exception { + try (Fixture fixture = new Fixture("pd_pending", true)) { + BlockingJob job = fixture.start(); + HugeException refusal = Assert.assertThrows(HugeException.class, fixture::drop); + Assert.assertContains(fixture.pd ? "Tasks are still active" : "Failed to wait", refusal.getMessage()); + fixture.assertRetained(); + Assert.assertTrue(fixture.metadataPresent.get()); + job.release.countDown(); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + fixture.drop(); + Assert.assertFalse(fixture.metadataPresent.get()); + Assert.assertTrue(fixture.graph.closed()); + Assert.assertNull(fixture.manager.graph(fixture.graph.spaceGraphName())); + } + } + + @Test + public void testConfiguredWaitAllowsShortTaskToFinishBeforePdDrop() throws Exception { + try (Fixture fixture = new Fixture("pd_wait", true)) { + fixture.graph.configuration().setProperty(CoreOptions.TASK_WAIT_TIMEOUT.name(), 2L); + BlockingJob job = fixture.start(); + CountDownLatch fenced = new CountDownLatch(1); + TaskScheduler scheduler = Mockito.mock(TaskScheduler.class, + org.mockito.AdditionalAnswers.delegatesTo(fixture.scheduler)); + Mockito.doAnswer(invocation -> { + boolean drained = fixture.scheduler.close(); + if (!drained) { + fenced.countDown(); + } + return drained; + }).when(scheduler).close(); + HugeGraph proxy = Mockito.mock(HugeGraph.class, org.mockito.AdditionalAnswers.delegatesTo(fixture.graph)); + Mockito.doReturn(scheduler).when(proxy).taskScheduler(); + fixture.graphs.put(fixture.graph.spaceGraphName(), proxy); + ExecutorService caller = Executors.newSingleThreadExecutor(); + try { + Future deletion = caller.submit(fixture::drop); + Assert.assertTrue(fenced.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(fixture.metadataPresent.get()); + job.release.countDown(); + deletion.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(fixture.graph.closed()); + Assert.assertFalse(fixture.metadataPresent.get()); + } finally { + job.release.countDown(); + caller.shutdown(); + Assert.assertTrue(caller.awaitTermination(10L, TimeUnit.SECONDS)); + fixture.graphs.replace(fixture.graph.spaceGraphName(), proxy, fixture.graph); + } + } + } + + @Test + public void testSchedulerCloseFailureDoesNotRemovePdMetadata() throws Exception { + try (Fixture fixture = new Fixture("pd_close_failure", true)) { + Graph proxy = Mockito.mock(HugeGraph.class, Mockito.withSettings().defaultAnswer( + org.mockito.AdditionalAnswers.delegatesTo(fixture.graph))); + TaskScheduler failed = Mockito.spy(fixture.scheduler); + Mockito.doThrow(new IllegalStateException("scheduler close failed")).when(failed).close(); + Mockito.doReturn(failed).when((HugeGraph) proxy).taskScheduler(); + fixture.graphs.put(fixture.graph.spaceGraphName(), proxy); + try { + RuntimeException error = Assert.assertThrows(RuntimeException.class, fixture::drop); + Assert.assertContains("scheduler close failed", error.getMessage()); + Assert.assertTrue(fixture.metadataPresent.get()); + Assert.assertFalse(fixture.graph.closed()); + Assert.assertTrue(Files.exists(fixture.file)); + } finally { + fixture.graphs.put(fixture.graph.spaceGraphName(), fixture.graph); + } + } + } + + @Test + public void testMissingGraphReportsArgumentErrorFromInitializedManager() throws Exception { + try (Fixture fixture = new Fixture("missing_graph", true)) { + IllegalArgumentException error = Assert.assertThrows(IllegalArgumentException.class, + () -> fixture.manager.dropGraph("DEFAULT", "missing", true)); + Assert.assertContains("doesn't exist", error.getMessage()); + fixture.assertRetained(); + Assert.assertTrue(fixture.metadataPresent.get()); + } + } + + @Test + public void testGraphSpaceSharedMetadataSurvivesRefusedDrop() throws Exception { + try (Fixture fixture = new Fixture("space_pending", true)) { + BlockingJob job = fixture.start(); + HugeException refusal = Assert.assertThrows(HugeException.class, + () -> fixture.manager.clearGraphSpace("DEFAULT")); + Assert.assertContains("Tasks are still active", refusal.getMessage()); + Assert.assertFalse(fixture.sharedMetadataCleared.get()); + Assert.assertSame(fixture.graph, fixture.manager.graph(fixture.graph.spaceGraphName())); + job.release.countDown(); + } + } + + private static final class Fixture implements AutoCloseable { + + private final Path directory; + private final Path file; + private final GraphManager manager; + private final Map graphs; + private final HugeGraph graph; + private final HugeGraph neighbor; + private final HugeGraphParams params; + private final TaskScheduler scheduler; + private final boolean pd; + private final AtomicBoolean metadataPresent = new AtomicBoolean(true); + private final AtomicBoolean sharedMetadataCleared = new AtomicBoolean(); + private final List jobs = new ArrayList<>(); + private int dropEvents; + + private Fixture(String name, boolean pd) throws Exception { + RegisterUtil.registerBackends(); + this.pd = pd; + this.directory = Files.createTempDirectory("graph-drop-core-"); + HugeConfig config = FakeObjects.newConfig(); + config.setProperty(ServerOptions.USE_PD.name(), false); + config.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false); + config.setProperty(ServerOptions.ENABLE_DYNAMIC_CREATE_DROP.name(), true); + config.setProperty(ServerOptions.GRAPHS.name(), this.directory.toString()); + EventHub events = new EventHub("drop-core-" + name); + this.manager = new GraphManager(config, events); + this.graphs = Whitebox.getInternalState(this.manager, "graphs"); + this.graph = this.open(name); + this.neighbor = this.open(name + "_neighbor"); + this.params = Whitebox.getInternalState(this.graph, "params"); + this.scheduler = this.graph.taskScheduler(); + this.file = this.directory.resolve(name + ".properties"); + events.listen(Events.GRAPH_DROP, event -> { + if (event.args()[0] == this.graph) { + this.dropEvents++; + } + return null; + }); + if (pd) { + // Replace the external PD service, never bypass manager initialization. + Whitebox.setInternalState(this.manager, "PDExist", true); + MetaManager meta = Mockito.mock(MetaManager.class); + Whitebox.setInternalState(this.manager, "metaManager", meta); + Mockito.when(meta.graphConfigs("DEFAULT")).thenReturn(java.util.Collections.emptyMap()); + Mockito.doAnswer(invocation -> { + this.metadataPresent.set(false); + return null; + }).when(meta).removeGraphConfig(Mockito.eq("DEFAULT"), Mockito.eq(name)); + Mockito.doAnswer(invocation -> { + this.sharedMetadataCleared.set(true); + return null; + }).when(meta).clearGraphAuth("DEFAULT"); + Map spaces = Whitebox.getInternalState(this.manager, "graphSpaces"); + spaces.put("DEFAULT", Mockito.mock(GraphSpace.class)); + } + } + + private HugeGraph open(String name) throws Exception { + HugeConfig config = FakeObjects.newConfig(); + config.setProperty(Graph.GRAPH, HugeFactory.class.getName()); + config.setProperty("backend", "memory"); + config.setProperty("serializer", "text"); + config.setProperty("store", name); + config.setProperty(CoreOptions.TASK_WAIT_TIMEOUT.name(), 0L); + Path path = this.directory.resolve(name + ".properties"); + config.save(path.toFile()); + config.file(path.toString()); + HugeGraph result = HugeFactory.open(config); + result.initBackend(); + result.serverStarted(GlobalMasterInfo.master("drop-core-test")); + result.schema().vertexLabel("person").useCustomizeStringId().create(); + result.addVertex(T.id, "kept", T.label, "person"); + result.tx().commit(); + ((HugeGraphParams) Whitebox.getInternalState(result, "params")).closeTx(); + this.graphs.put(result.spaceGraphName(), result); + return result; + } + + private BlockingJob start() throws Exception { + BlockingJob job = new BlockingJob(); + this.jobs.add(job); + HugeTask task = new HugeTask<>(IdGenerator.of(9997000L), null, job); + task.type("test"); + task.name("drop-owner"); + this.scheduler.schedule(task); + Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS)); + return job; + } + + private void drop() { + if (this.pd) { + this.manager.dropGraph("DEFAULT", this.graph.name(), true); + } else { + this.manager.dropGraphLocal(this.graph.name()); + } + } + + private void assertRetained() { + Assert.assertSame(this.graph, this.manager.graph(this.graph.spaceGraphName())); + Assert.assertFalse(this.graph.closed()); + Assert.assertTrue(Files.exists(this.file)); + Assert.assertEquals(IdGenerator.of("kept"), this.graph.vertices("kept").next().id()); + this.params.closeTx(); + Map registered = Whitebox.getInternalState(HugeFactory.class, "GRAPHS"); + Assert.assertSame(this.graph, registered.get(this.graph.spaceGraphName())); + Assert.assertEquals(0, this.dropEvents); + } + + @Override + public void close() throws Exception { + for (BlockingJob job : this.jobs) { + job.release.countDown(); + } + this.scheduler.waitUntilAllTasksCompleted(10L); + this.manager.close(); + for (HugeGraph current : new HugeGraph[]{this.graph, this.neighbor}) { + if (!current.closed()) { + current.close(); + } + HugeFactory.remove(current); + } + this.graphs.clear(); + FileUtils.deleteDirectory(this.directory.toFile()); + } + } + + public static class BlockingJob extends SysJob { + + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + this.graph().vertices("kept").next(); + this.started.countDown(); + boolean interrupted = false; + while (this.release.getCount() != 0L) { + try { + this.release.await(); + } catch (InterruptedException ignored) { + interrupted = true; + } + } + if (interrupted) { + Thread.currentThread().interrupt(); + } + return null; + } + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java index ba34359d46..df9a69b976 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java @@ -76,12 +76,12 @@ public void testCleanupContinuesAfterGraphFailure() { } @Test - public void testGraphCloseContinuesAfterReleasedWorkerOwnerFails() throws Exception { + public void testGraphCloseRetainsSchedulerAfterReleasedWorkerOwnerFails() throws Exception { this.checkGraphCloseAfterOwnerFailure(false); } @Test - public void testGraphCloseContinuesAfterWorkerOwnerFailsBeforeRelease() throws Exception { + public void testGraphCloseRetainsSchedulerAfterWorkerOwnerFailsBeforeRelease() throws Exception { this.checkGraphCloseAfterOwnerFailure(true); } @@ -129,14 +129,27 @@ private void checkGraphCloseAfterOwnerFailure(boolean beforeRelease) throws Exce Throwable failure = Assert.assertThrows(HugeException.class, graph::close); Assert.assertSame(first, HugeException.rootCause(failure)); Assert.assertTrue(Arrays.asList(failure.getSuppressed()).contains(second)); - Assert.assertTrue(graph.closed()); - Assert.assertNull(manager.getScheduler(params)); + Assert.assertFalse(graph.closed()); + Assert.assertSame(scheduler, manager.getScheduler(params)); Assert.assertTrue((boolean) Whitebox.getInternalState(scheduler.serverManager(), "closed")); Assert.assertNull(scheduler.call(owners::get)); Assert.assertNull(owners.get()); Assert.assertEquals(1, workerClosed.get()); Assert.assertEquals(1, callerClosed.get()); Assert.assertEquals(1, otherWorkerClosed.get()); + Mockito.verify(provider, Mockito.never()).close(); + + // A close failure is not an ACK, even if its owner released first. + if (beforeRelease) { + scheduler.call(() -> { + workerOwner.get().close(); + return null; + }); + } + graph.close(); + Assert.assertTrue(graph.closed()); + Assert.assertNull(manager.getScheduler(params)); + Assert.assertEquals(1, otherWorkerClosed.get()); Mockito.verify(provider).close(); } finally { try { diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java new file mode 100644 index 0000000000..1425693fa3 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/StandardTaskAdmissionTest.java @@ -0,0 +1,331 @@ +/* + * 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.hugegraph; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.dist.RegisterUtil; +import org.apache.hugegraph.id.IdGenerator; +import org.apache.hugegraph.job.SysJob; +import org.apache.hugegraph.masterelection.GlobalMasterInfo; +import org.apache.hugegraph.task.HugeTask; +import org.apache.hugegraph.task.StandardTaskScheduler; +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.unit.FakeObjects; +import org.apache.tinkerpop.gremlin.structure.T; +import org.junit.Test; +import org.mockito.Mockito; + +public class StandardTaskAdmissionTest { + + @Test + public void testCancelledWorkerFencesAdmissionUntilActualExit() throws Exception { + try (Fixture fixture = new Fixture("standard_cancel_worker")) { + BlockingJob job = new BlockingJob(); + HugeTask task = fixture.task(job); + fixture.scheduler.schedule(task); + Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS)); + fixture.scheduler.cancel(task); + Assert.assertTrue(task.isCancelled()); + Assert.assertEquals(1, fixture.scheduler.pendingTasks()); + Assert.assertFalse(fixture.scheduler.close()); + HugeTask late = fixture.task(new EmptyJob()); + Assert.assertThrows(IllegalStateException.class, () -> fixture.scheduler.schedule(late)); + Assert.assertThrows(IllegalStateException.class, () -> { + fixture.scheduler.restoreTasks(); + }, error -> Assert.assertContains("closing", error.getMessage())); + Assert.assertEquals(TaskStatus.NEW, late.status()); + job.release.countDown(); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertEquals(TaskStatus.CANCELLED, fixture.scheduler.findTask(task.id()).status()); + Assert.assertTrue(fixture.scheduler.close()); + } + } + + @Test + public void testDispatchFutureCancellationKeepsRunningOwner() throws Exception { + try (Fixture fixture = new Fixture("standard_cancel_future")) { + BlockingJob job = new BlockingJob(); + HugeTask task = fixture.task(job); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS)); + Assert.assertTrue(dispatch.cancel(false)); + Assert.assertFalse(task.isCancelled()); + Assert.assertFalse(fixture.scheduler.close()); + job.release.countDown(); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertEquals(TaskStatus.SUCCESS, fixture.scheduler.findTask(task.id()).status()); + Assert.assertTrue(fixture.scheduler.close()); + } + } + + @Test + public void testCancellationCallbackRetainsItsTransactionAfterWorkerExits() throws Exception { + try (Fixture fixture = new Fixture("standard_cancel_callback")) { + fixture.graph.schema().vertexLabel("person").useCustomizeStringId().create(); + CountDownLatch callback = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + BlockingJob job = new BlockingJob() { + @Override + protected void cancelled() { + this.graph().addVertex(T.id, "callback", T.label, "person"); + callback.countDown(); + await(release); + this.graph().tx().commit(); + super.cancelled(); + } + }; + HugeTask task = fixture.task(job); + fixture.scheduler.schedule(task); + Assert.assertTrue(job.started.await(10L, TimeUnit.SECONDS)); + ExecutorService caller = Executors.newSingleThreadExecutor(); + try { + Future cancellation = caller.submit(() -> fixture.scheduler.cancel(task)); + Assert.assertTrue(callback.await(10L, TimeUnit.SECONDS)); + job.release.countDown(); + fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS); + Assert.assertEquals(1, fixture.scheduler.pendingTasks()); + Assert.assertFalse(fixture.scheduler.close()); + release.countDown(); + cancellation.get(10L, TimeUnit.SECONDS); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertTrue(fixture.graph.vertices("callback").hasNext()); + Assert.assertEquals(TaskStatus.CANCELLED, fixture.scheduler.findTask(task.id()).status()); + } finally { + release.countDown(); + caller.shutdownNow(); + Assert.assertTrue(caller.awaitTermination(10L, TimeUnit.SECONDS)); + } + } + } + + @Test + public void testQueuedCancellationAndSubmissionRejectionReleaseOwners() throws Exception { + try (Fixture fixture = new Fixture("standard_cancel_queued")) { + CountDownLatch occupied = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + fixture.worker.submit(() -> { + occupied.countDown(); + release.await(); + return null; + }); + try { + Assert.assertTrue(occupied.await(10L, TimeUnit.SECONDS)); + BlockingJob job = new BlockingJob(); + Future queued = fixture.scheduler.schedule(fixture.task(job)); + Assert.assertTrue(queued.cancel(false)); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + release.countDown(); + fixture.worker.submit(() -> null).get(10L, TimeUnit.SECONDS); + Assert.assertEquals(1L, job.started.getCount()); + fixture.worker.shutdown(); + Assert.assertTrue(fixture.worker.awaitTermination(10L, TimeUnit.SECONDS)); + Assert.assertThrows(RejectedExecutionException.class, + () -> fixture.scheduler.schedule(fixture.task(new EmptyJob()))); + Assert.assertEquals(0, fixture.scheduler.pendingTasks()); + } finally { + release.countDown(); + } + } + } + + @Test + public void testDependencyRetryDoesNotReleaseEarlierCleanupOwner() throws Exception { + try (Fixture fixture = new Fixture("standard_dependency_owner")) { + CountDownLatch closing = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicBoolean first = new AtomicBoolean(true); + Mockito.doAnswer(invocation -> { + if (Thread.currentThread().getName().equals("standard-test-worker") && + first.compareAndSet(true, false)) { + closing.countDown(); + Assert.assertTrue(release.await(10L, TimeUnit.SECONDS)); + } + return invocation.callRealMethod(); + }).when(fixture.params).closeTx(); + HugeTask dependency = fixture.task(new EmptyJob()); + fixture.scheduler.save(dependency); + HugeTask task = fixture.task(new EmptyJob()); + task.depends(dependency.id()); + fixture.scheduler.schedule(task); + try { + Assert.assertTrue(closing.await(10L, TimeUnit.SECONDS)); + fixture.scheduler.cancel(task); + Assert.assertEquals(1, fixture.scheduler.pendingTasks()); + Assert.assertFalse(fixture.scheduler.close()); + } finally { + release.countDown(); + } + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertEquals(TaskStatus.CANCELLED, fixture.scheduler.findTask(task.id()).status()); + } + } + + @Test + public void testCloseFencesDependencyResubmission() throws Exception { + try (Fixture fixture = new Fixture("standard_dependency_fence")) { + CountDownLatch occupied = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + fixture.worker.submit(() -> { + occupied.countDown(); + release.await(); + return null; + }); + try { + Assert.assertTrue(occupied.await(10L, TimeUnit.SECONDS)); + HugeTask dependency = fixture.task(new EmptyJob()); + fixture.scheduler.save(dependency); + HugeTask task = fixture.task(new EmptyJob()); + task.depends(dependency.id()); + Future dispatch = fixture.scheduler.schedule(task); + Assert.assertFalse(fixture.scheduler.close()); + release.countDown(); + dispatch.get(10L, TimeUnit.SECONDS); + fixture.scheduler.waitUntilAllTasksCompleted(10L); + Assert.assertEquals(TaskStatus.FAILED, fixture.scheduler.findTask(task.id()).status()); + Assert.assertContains("is closing", task.result()); + Assert.assertTrue(fixture.scheduler.close()); + } finally { + release.countDown(); + } + } + } + + @Test + public void testCloseFailureRemainsRetryable() throws Exception { + try (Fixture fixture = new Fixture("standard_close_failure")) { + Mockito.doThrow(new IllegalStateException("close failed")).doCallRealMethod() + .when(fixture.params).closeTx(); + Assert.assertThrows(RuntimeException.class, fixture.scheduler::close); + HugeTask late = fixture.task(new EmptyJob()); + Assert.assertThrows(IllegalStateException.class, () -> fixture.scheduler.schedule(late)); + Assert.assertTrue(fixture.scheduler.close()); + } + } + + private static void await(CountDownLatch latch) { + boolean interrupted = false; + while (true) { + try { + latch.await(); + break; + } catch (InterruptedException ignored) { + interrupted = true; + } + } + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + + private static class Fixture implements AutoCloseable { + + private final HugeGraph graph; + private final HugeGraphParams params; + private final ExecutorService worker; + private final ExecutorService database; + private final StandardTaskScheduler scheduler; + private final List jobs = new ArrayList<>(); + private long nextId = 9998000L; + + private Fixture(String name) { + RegisterUtil.registerBackends(); + HugeConfig config = FakeObjects.newConfig(); + config.setProperty("backend", "memory"); + config.setProperty("serializer", "text"); + config.setProperty("store", name); + this.graph = HugeFactory.open(config); + this.graph.initBackend(); + this.graph.serverStarted(GlobalMasterInfo.master("standard-owner-test")); + this.params = Mockito.spy((HugeGraphParams) Whitebox.getInternalState(this.graph, "params")); + this.worker = Executors.newSingleThreadExecutor(runnable -> + new Thread(runnable, "standard-test-worker")); + this.database = Executors.newSingleThreadExecutor(); + this.scheduler = new StandardTaskScheduler(this.params, this.worker, this.database); + } + + private HugeTask task(TaskCallable job) { + if (job instanceof BlockingJob) { + this.jobs.add((BlockingJob) job); + } + HugeTask task = new HugeTask<>(IdGenerator.of(this.nextId++), null, job); + task.type("test"); + task.name("standard-owner"); + return task; + } + + @Override + public void close() throws Exception { + for (BlockingJob job : this.jobs) { + job.release.countDown(); + } + this.scheduler.close(); + this.worker.shutdown(); + Assert.assertTrue(this.worker.awaitTermination(10L, TimeUnit.SECONDS)); + Assert.assertTrue(this.scheduler.close()); + this.database.shutdownNow(); + Assert.assertTrue(this.database.awaitTermination(10L, TimeUnit.SECONDS)); + this.graph.close(); + HugeFactory.remove(this.graph); + } + } + + public static class EmptyJob extends SysJob { + + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + return null; + } + } + + public static class BlockingJob extends SysJob { + + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + @Override + public String type() { + return "test"; + } + + @Override + public Object execute() { + this.graph().vertices().hasNext(); + this.started.countDown(); + await(this.release); + return null; + } + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java index b83e083a77..c36bed15f0 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java @@ -18,6 +18,9 @@ package org.apache.hugegraph.unit; import org.apache.hugegraph.HugeFactoryTest; +import org.apache.hugegraph.GraphDropPendingTest; +import org.apache.hugegraph.StandardTaskAdmissionTest; +import org.apache.hugegraph.DistributedTaskAdmissionTest; import org.apache.hugegraph.api.auth.GraphSpaceAuthPayloadTest; import org.apache.hugegraph.api.auth.GraphSpaceGroupAPITest; import org.apache.hugegraph.api.cypher.CypherClientTest; @@ -150,6 +153,9 @@ StandardAuthManagerV2Test.class, ContextTaskTest.class, HugeFactoryTest.class, + GraphDropPendingTest.class, + StandardTaskAdmissionTest.class, + DistributedTaskAdmissionTest.class, AuthMetaManagerTest.class, /* api space */ diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java index 4d893e3199..1a0b585b0c 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/space/GraphSpaceAPITest.java @@ -19,7 +19,6 @@ import static org.apache.hugegraph.HugeFactory.SYS_GRAPH; -import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.Collections; import java.util.Map; @@ -37,11 +36,15 @@ import org.apache.hugegraph.auth.HugeUser; import org.apache.hugegraph.auth.RolePermission; import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.config.ServerOptions; +import org.apache.hugegraph.event.EventHub; import org.apache.hugegraph.meta.MetaManager; import org.apache.hugegraph.space.GraphSpace; import org.apache.hugegraph.testutil.Assert; import org.apache.hugegraph.testutil.Whitebox; import org.apache.hugegraph.unit.BaseUnitTest; +import org.apache.hugegraph.unit.FakeObjects; import org.apache.tinkerpop.gremlin.structure.Graph; import org.junit.After; import org.junit.Test; @@ -49,7 +52,6 @@ import jakarta.ws.rs.ForbiddenException; import jakarta.ws.rs.core.SecurityContext; -import sun.misc.Unsafe; public class GraphSpaceAPITest extends BaseUnitTest { @@ -221,7 +223,7 @@ public void testSpaceManagerCannotDeleteSpaceDefaultRole() { @Test public void testPdGraphsIncludesLoadedLocalConfigGraphs() { - GraphManager manager = allocateGraphManager(); + GraphManager manager = newGraphManager(); Whitebox.setInternalState(manager, "PDExist", true); Whitebox.setInternalState(manager, "localGraphs", Collections.emptySet()); @@ -276,7 +278,7 @@ private static GraphManager managerWithDefaultRoleContext(String operator, HugeAuthenticator authenticator = Mockito.mock(HugeAuthenticator.class); Mockito.when(authenticator.authManager()).thenReturn(authManager); - GraphManager manager = allocateGraphManager(); + GraphManager manager = newGraphManager(); Whitebox.setInternalState(manager, "PDExist", true); Whitebox.setInternalState(manager, "authenticator", authenticator); Map spaces = new ConcurrentHashMap<>(); @@ -301,7 +303,7 @@ private static GraphManager managerWithDefaultRoleContext(String operator, private static GraphManager managerWithLoadedDefaultGraph(String operator, boolean admin) { - GraphManager manager = allocateGraphManager(); + GraphManager manager = newGraphManager(); Whitebox.setInternalState(manager, "PDExist", true); Whitebox.setInternalState(manager, "localGraphs", Collections.singleton("hugegraph")); @@ -326,15 +328,11 @@ private static GraphManager managerWithLoadedDefaultGraph(String operator, return manager; } - private static GraphManager allocateGraphManager() { - try { - Field field = Unsafe.class.getDeclaredField("theUnsafe"); - field.setAccessible(true); - Unsafe unsafe = (Unsafe) field.get(null); - return (GraphManager) unsafe.allocateInstance(GraphManager.class); - } catch (Exception e) { - throw new AssertionError(e); - } + private static GraphManager newGraphManager() { + HugeConfig config = FakeObjects.newConfig(); + config.setProperty(ServerOptions.USE_PD.name(), false); + config.setProperty(ServerOptions.GRAPH_LOAD_FROM_LOCAL_CONFIG.name(), false); + return new GraphManager(config, new EventHub("graph-space-api-test")); } private static void setContext(String username) { From 072c0d1b235e218dc510ae1e09ad58477d0eac8d Mon Sep 17 00:00:00 2001 From: imbajin Date: Tue, 13 Oct 2026 02:05:22 +0800 Subject: [PATCH 2/5] fix(task): preserve shutdown and test phases - fence close without holding the cron monitor across queries - test the real cron barrier timeout and close retry - wait for callable entry before asserting task deletion --- .../task/DistributedTaskScheduler.java | 2 +- .../DistributedTaskAdmissionTest.java | 51 ++++++++++++++++++- .../apache/hugegraph/core/TaskCoreTest.java | 6 ++- 3 files changed, 55 insertions(+), 4 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 f29224bc2a..0fe0e5b8f2 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 @@ -125,7 +125,7 @@ private static boolean sleep(long ms) { } } - public synchronized void cronSchedule() { + public void cronSchedule() { // Perform periodic scheduling tasks // Check closed flag first to exit early diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java index c587994473..bddb0c3ec3 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/DistributedTaskAdmissionTest.java @@ -17,6 +17,7 @@ package org.apache.hugegraph; +import java.util.Iterator; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -357,6 +358,43 @@ protected void done() { } } + @Test + public void testBlockedCronStillFencesAdmissionAndTimesOutClose() throws Exception { + try (Fixture fixture = new Fixture("blocked_cron", 1L)) { + CountDownLatch querying = new CountDownLatch(1); + CountDownLatch finishQuery = new CountDownLatch(1); + fixture.scheduler.beforeTaskQuery = () -> { + querying.countDown(); + awaitUninterruptibly(finishQuery); + }; + ExecutorService requests = Executors.newSingleThreadExecutor(); + Future cron = fixture.cron.submit(fixture.scheduler::cronSchedule); + try { + Assert.assertTrue(querying.await(10L, TimeUnit.SECONDS)); + Future close = requests.submit(fixture.scheduler::close); + AtomicBoolean closed = Whitebox.getInternalState(fixture.scheduler, "closed"); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2L); + while (!closed.get() && System.nanoTime() < deadline) { + Thread.yield(); + } + Assert.assertTrue(closed.get()); + Assert.assertThrows(IllegalStateException.class, + () -> fixture.scheduler.schedule(task(true, new AtomicBoolean(), 9999919L))); + // This exercises the real cron barrier, not a mocked drain method. + Assert.assertFalse(close.get(10L, TimeUnit.SECONDS)); + Assert.assertFalse(cron.isDone()); + finishQuery.countDown(); + cron.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(fixture.scheduler.close()); + fixture.assertWorkerClean(); + } finally { + finishQuery.countDown(); + cron.get(10L, TimeUnit.SECONDS); + stop(requests); + } + } + } + private static void awaitUninterruptibly(CountDownLatch release) { while (release.getCount() != 0L) { try { @@ -407,12 +445,16 @@ private static class Fixture implements AutoCloseable { private final Map registry; Fixture(String suffix) { + this(suffix, 600L); + } + + Fixture(String suffix, long schedulePeriod) { RegisterUtil.registerBackends(); HugeConfig config = FakeObjects.newConfig(); config.setProperty("backend", "memory"); config.setProperty("serializer", "text"); config.setProperty("store", "distributed_admission_" + suffix); - config.setProperty("task.schedule_period", 600L); + config.setProperty("task.schedule_period", schedulePeriod); config.setProperty("task.retry", 0); this.graph = HugeFactory.open(config); this.graph.initBackend(); @@ -498,6 +540,13 @@ protected void beforeExecute(Thread thread, Runnable work) { private static class TestScheduler extends DistributedTaskScheduler { private volatile Runnable beforeTaskRead = () -> { }; + private volatile Runnable beforeTaskQuery = () -> { }; + + @Override + protected Iterator> queryTaskWithoutResultByStatus(TaskStatus status) { + this.beforeTaskQuery.run(); + return super.queryTaskWithoutResultByStatus(status); + } @Override public HugeTask task(Id id, boolean withResult) { 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 0622630ecc..ddc653a4c6 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 @@ -234,7 +234,8 @@ public void testDeleteIncompleteTask() { TaskScheduler scheduler = graph.taskScheduler(); Id id = IdGenerator.of(88891); - HugeTask task = new HugeTask<>(id, null, new SleepCallable<>()); + BlockingCallable.reset(); + HugeTask task = new HugeTask<>(id, null, new BlockingCallable<>()); task.type("test"); task.name("delete-incomplete-task"); scheduler.schedule(task); @@ -248,13 +249,14 @@ public void testDeleteIncompleteTask() { e.getMessage()); }); } else { - waitUntilTaskRunning(scheduler); + Assert.assertTrue(BlockingCallable.awaitStarted()); HugeTask deleted = scheduler.delete(id, false); Assert.assertNotNull(deleted); Assert.assertEquals(TaskStatus.DELETING, deleted.status()); Assert.assertEquals(TaskStatus.DELETING, scheduler.task(id).status()); } } finally { + BlockingCallable.release(); try { scheduler.waitUntilAllTasksCompleted(10); } catch (TimeoutException ignored) { From c2897c0f0b2f87e10b86155f3428d0de92d66116 Mon Sep 17 00:00:00 2001 From: imbajin Date: Tue, 13 Oct 2026 03:39:16 +0800 Subject: [PATCH 3/5] fix: align task and ttl test timing - wait for actual ephemeral task drain before absence checks - use current input time for TTL start timestamps - verify exact expiry while retaining all query assertions --- .../main/java/org/apache/hugegraph/core/TaskCoreTest.java | 3 +++ .../main/java/org/apache/hugegraph/core/VertexCoreTest.java | 5 ++++- 2 files changed, 7 insertions(+), 1 deletion(-) 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 ddc653a4c6..3a3837a2e1 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 @@ -606,6 +606,9 @@ public Object execute() throws Exception { Assert.assertEquals(TaskStatus.SUCCESS, task.status()); Assert.assertEquals("{\"k1\":13579,\"k2\":\"24680\"}", task.result()); + // Future.get() publishes the result before done() and native owner cleanup finish. + scheduler.waitUntilAllTasksCompleted(10); + // Ephemeral tasks are not stored in DB, so these should throw NotFoundException Assert.assertThrows(NotFoundException.class, () -> { scheduler.waitUntilTaskCompleted(task.id(), 10); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java index d0c22e2f22..8268291555 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/VertexCoreTest.java @@ -1309,11 +1309,14 @@ public void testAddVertexWithTtl() { @Test public void testAddVertexWithTtlAndTtlStartTime() { + // The input date must not reuse a previously closed transaction's timestamp. + long birth = DateUtil.now().getTime() - 1000L; Vertex vertex = graph().addVertex( T.label, "follower", "name", "Baby", "age", 3, "city", "Beijing", - "birth", graph().now() - 1000L); + "birth", birth); this.commitTx(); + Assert.assertEquals(birth + 3000L, ((HugeElement) vertex).expiredTime()); Iterator vertices = graph().vertices(vertex); Assert.assertTrue(vertices.hasNext()); From 7b0ed580b25be26f522bb4de92ad5cf61581c5f9 Mon Sep 17 00:00:00 2001 From: imbajin Date: Tue, 13 Oct 2026 09:10:06 +0800 Subject: [PATCH 4/5] fix(core): retain task shutdown grace - preserve executor grace with retained schedulers - explain task admission after refused graph deletion - verify shutdown callbacks and retained graph data --- docs/server-runtime.md | 7 ++- .../apache/hugegraph/core/GraphManager.java | 7 ++- .../apache/hugegraph/StandardHugeGraph.java | 12 ++++- .../apache/hugegraph/task/TaskManager.java | 5 +- .../hugegraph/GraphDropPendingTest.java | 16 ++++++- .../org/apache/hugegraph/HugeFactoryTest.java | 48 +++++++++++++++++++ 6 files changed, 85 insertions(+), 10 deletions(-) diff --git a/docs/server-runtime.md b/docs/server-runtime.md index ee17e4287f..0bc0e98ed8 100644 --- a/docs/server-runtime.md +++ b/docs/server-runtime.md @@ -470,8 +470,11 @@ combinations actually used by the application. Stop application writes before deletion and keep them stopped during retries. The local scheduler closes task admission and waits for actual executions, cancellation callbacks and their transaction cleanup before data or metadata is removed. If draining times out, -the graph stays registered with its data and configuration; retry after its owners exit. -Closing the scheduler alone leaves persistent work that has not started in storage. +the graph stays registered with its data and configuration, but task admission remains +closed: new asynchronous tasks are rejected even after the existing owners exit. Keep +application writes stopped and retry deletion after those owners exit. If no destructive +cleanup has begun, restart the Server to resume task admission instead of completing +the deletion. Closing the scheduler alone leaves unstarted persistent work in storage. This is a local shutdown guarantee. For a graph shared by several Servers, stop writes and tasks on every Server before deletion; notification delivery is not a cluster-wide diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java index fd4996890a..c7d821bf38 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java @@ -2265,10 +2265,13 @@ public void dropGraph(String graphSpace, String name, boolean clear) { try { scheduler.waitUntilAllTasksCompleted(g.configuration().get(CoreOptions.TASK_WAIT_TIMEOUT)); } catch (TimeoutException e) { - throw new HugeException("Tasks are still active in graph '%s', please retry later", + throw new HugeException("Tasks are still active in graph '%s'; task admission remains closed. " + + "Retry deletion after owners exit, or restart to resume task admission", e, graphName); } - E.checkState(scheduler.close(), "Tasks are still active in graph '%s', please retry later", + E.checkState(scheduler.close(), + "Tasks are still active in graph '%s'; task admission remains closed. " + + "Retry deletion after owners exit, or restart to resume task admission", graphName); } this.removingGraphs.add(graphName); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java index 18416c07eb..157ec6b74b 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java @@ -1217,8 +1217,16 @@ public void drop() { TaskScheduler scheduler = this.taskManager.getScheduler(this.params); if (scheduler != null && !scheduler.close()) { // Fence admission before waiting for actual task owners to exit. - this.waitUntilAllTasksCompleted(); - E.checkState(scheduler.close(), "Tasks are still active in graph '%s', please retry later", + try { + scheduler.waitUntilAllTasksCompleted(this.configuration().get(CoreOptions.TASK_WAIT_TIMEOUT)); + } catch (TimeoutException e) { + throw new HugeException("Tasks are still active in graph '%s'; task admission remains closed. " + + "Retry deletion after owners exit, or restart to resume task admission", + e, this.spaceGraphName()); + } + E.checkState(scheduler.close(), + "Tasks are still active in graph '%s'; task admission remains closed. " + + "Retry deletion after owners exit, or restart to resume task admission", this.spaceGraphName()); } this.clearBackend(); diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java index 6ad1ebf9e1..6d864d288d 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java @@ -251,8 +251,9 @@ public ServerInfoManager getServerInfoManager(HugeGraphParams graph) { } public void shutdown(long timeout) { - E.checkState(this.schedulers.isEmpty(), - "Task schedulers are still registered: %s", this.schedulers.size()); + if (!this.schedulers.isEmpty()) { + LOG.warn("Stopping task executors with {} schedulers still registered", this.schedulers.size()); + } Throwable ex = null; boolean terminated = this.distributedSchedulerExecutor.isTerminated(); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java index 9dac803821..a7a0c57125 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/GraphDropPendingTest.java @@ -62,7 +62,13 @@ public void testLocalDropKeepsDataAndRegistrationUntilCancelledWorkerExits() thr fixture.scheduler.cancel(job.task()); Assert.assertTrue(job.task().isCancelled()); HugeException refusal = Assert.assertThrows(HugeException.class, fixture::drop); - Assert.assertContains(fixture.pd ? "Tasks are still active" : "Failed to wait", refusal.getMessage()); + Assert.assertContains("Tasks are still active", refusal.getMessage()); + Assert.assertContains("task admission remains closed", refusal.getMessage()); + Assert.assertContains("restart to resume task admission", refusal.getMessage()); + BlockingJob late = new BlockingJob(); + late.release.countDown(); + Assert.assertThrows(IllegalStateException.class, () -> fixture.scheduler.schedule( + new HugeTask<>(IdGenerator.of(9997001L), null, late))); fixture.assertRetained(); job.release.countDown(); fixture.scheduler.waitUntilAllTasksCompleted(10L); @@ -79,7 +85,13 @@ public void testPdDropPreservesMetadataUntilLocalTasksDrain() throws Exception { try (Fixture fixture = new Fixture("pd_pending", true)) { BlockingJob job = fixture.start(); HugeException refusal = Assert.assertThrows(HugeException.class, fixture::drop); - Assert.assertContains(fixture.pd ? "Tasks are still active" : "Failed to wait", refusal.getMessage()); + Assert.assertContains("Tasks are still active", refusal.getMessage()); + Assert.assertContains("task admission remains closed", refusal.getMessage()); + Assert.assertContains("restart to resume task admission", refusal.getMessage()); + BlockingJob late = new BlockingJob(); + late.release.countDown(); + Assert.assertThrows(IllegalStateException.class, () -> fixture.scheduler.schedule( + new HugeTask<>(IdGenerator.of(9997001L), null, late))); fixture.assertRetained(); Assert.assertTrue(fixture.metadataPresent.get()); job.release.countDown(); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java index df9a69b976..e6251122e1 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java @@ -18,6 +18,7 @@ package org.apache.hugegraph; import java.util.ArrayList; +import java.lang.reflect.Constructor; import java.util.Arrays; import java.util.List; import java.util.Map; @@ -376,6 +377,53 @@ public void testIncompleteSchedulerCloseDoesNotQueueOwnerCleanup() { } } + @Test + public void testShutdownDrainsWorkersWithRegisteredScheduler() throws Exception { + Constructor constructor = TaskManager.class.getDeclaredConstructor(int.class); + constructor.setAccessible(true); + TaskManager manager = constructor.newInstance(1); + Map schedulers = Whitebox.getInternalState(manager, "schedulers"); + HugeGraphParams params = Mockito.mock(HugeGraphParams.class); + TaskScheduler scheduler = Mockito.mock(TaskScheduler.class); + schedulers.put(params, scheduler); + ExecutorService workers = Whitebox.getInternalState(manager, "taskExecutor"); + ExecutorService database = Whitebox.getInternalState(manager, "taskDbExecutor"); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicBoolean saved = new AtomicBoolean(); + ExecutorService stoppedWorkers = Mockito.spy(workers); + Mockito.doAnswer(invocation -> { + release.countDown(); + return invocation.callRealMethod(); + }).when(stoppedWorkers).shutdown(); + Whitebox.setInternalState(manager, "taskExecutor", stoppedWorkers); + try { + Future task = workers.submit(() -> { + started.countDown(); + Assert.assertTrue(release.await(10L, TimeUnit.SECONDS)); + database.submit(() -> saved.set(true)).get(10L, TimeUnit.SECONDS); + return null; + }); + Assert.assertTrue(started.await(10L, TimeUnit.SECONDS)); + manager.closeScheduler(params); + Assert.assertSame(scheduler, manager.getScheduler(params)); + manager.shutdown(10L); + task.get(10L, TimeUnit.SECONDS); + Assert.assertTrue(saved.get()); + Assert.assertSame(scheduler, manager.getScheduler(params)); + for (String field : new String[]{"taskExecutor", "taskDbExecutor", "schemaTaskExecutor", + "olapTaskExecutor", "ephemeralTaskExecutor", + "distributedSchedulerExecutor"}) { + ExecutorService executor = Whitebox.getInternalState(manager, field); + Assert.assertTrue(executor.isTerminated()); + } + } finally { + release.countDown(); + schedulers.clear(); + manager.shutdown(10L); + } + } + @Test public void testDistributedCloseRetriesDrainAfterTimeout() throws Exception { HugeGraphParams params = Mockito.mock(HugeGraphParams.class); From 337d7b68f1a054ed2f85a0c651f4f65c54cf3455 Mon Sep 17 00:00:00 2001 From: imbajin Date: Tue, 13 Oct 2026 10:34:16 +0800 Subject: [PATCH 5/5] fix: gate consumer failure in await test - wait for provide to return before injected failure - preserve await exception and timeout assertions - release the existing gate during test cleanup --- .../java/org/apache/hugegraph/unit/util/ConsumersTest.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java index 9db801be19..ea2cb0e0bc 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/util/ConsumersTest.java @@ -86,15 +86,18 @@ public void testAwaitPreservesFatalConsumerError() throws Throwable { @Test(timeout = 1000) public void testAwaitThrowsWhenConsumerThrows() throws Throwable { ExecutorService executor = Executors.newFixedThreadPool(2); + CountDownLatch release = new CountDownLatch(1); try { final String msg = "Injected exception for test"; Consumers consumers = new Consumers<>(executor, v -> { + awaitGate(release); throw new RuntimeException(msg); }); consumers.start("test"); consumers.provide(1); + release.countDown(); try { consumers.await(); @@ -108,6 +111,7 @@ public void testAwaitThrowsWhenConsumerThrows() throws Throwable { root.getMessage().contains(msg)); } } finally { + release.countDown(); executor.shutdownNow(); } }