This is an automated email from the ASF dual-hosted git repository.
imbajin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hugegraph.git
The following commit(s) were added to refs/heads/master by this push:
new 63c97ff93 fix(server): avoid loading huge task results for metadata
queries (#3060)
63c97ff93 is described below
commit 63c97ff93ff83ae005cad0654f961b250aab3cbc
Author: contrueCT <[email protected]>
AuthorDate: Sun Jun 21 20:27:45 2026 +0800
fix(server): avoid loading huge task results for metadata queries (#3060)
---
.../java/org/apache/hugegraph/api/job/TaskAPI.java | 16 +-
.../apache/hugegraph/auth/HugeGraphAuthProxy.java | 30 +-
.../hugegraph/task/DistributedTaskScheduler.java | 18 +-
.../java/org/apache/hugegraph/task/HugeTask.java | 69 +++-
.../hugegraph/task/StandardTaskScheduler.java | 107 +++++--
.../hugegraph/task/TaskAndResultScheduler.java | 64 +++-
.../org/apache/hugegraph/task/TaskScheduler.java | 14 +
.../org/apache/hugegraph/task/TaskTransaction.java | 29 ++
.../java/org/apache/hugegraph/api/TaskApiTest.java | 28 +-
.../org/apache/hugegraph/core/CoreTestSuite.java | 2 +
.../org/apache/hugegraph/core/TaskCoreTest.java | 99 ++++++
.../hugegraph/task/TaskAndResultSchedulerTest.java | 348 +++++++++++++++++++++
12 files changed, 777 insertions(+), 47 deletions(-)
diff --git
a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java
b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java
index d35cc9a95..7143efb5f 100644
---
a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java
+++
b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java
@@ -102,12 +102,13 @@ public class TaskAPI extends API {
limit = NO_LIMIT;
List<Id> idList = ids.stream().map(IdGenerator::of)
.collect(Collectors.toList());
- iter = scheduler.tasks(idList);
+ iter = scheduler.tasks(idList, false);
} else {
if (status == null) {
- iter = scheduler.tasks(null, limit, page);
+ iter = scheduler.tasks(null, limit, page, false);
} else {
- iter = scheduler.tasks(parseStatus(status), limit, page);
+ iter = scheduler.tasks(parseStatus(status), limit, page,
+ false);
}
}
@@ -136,12 +137,17 @@ public class TaskAPI extends API {
@Parameter(description = "The graph name")
@PathParam("graph") String graph,
@Parameter(description = "The task id")
- @PathParam("id") long id) {
+ @PathParam("id") long id,
+ @Parameter(description = "Whether to load
task result")
+ @DefaultValue("true")
+ @QueryParam("with_result")
+ boolean withResult) {
LOG.debug("Graph [{}] get task: {}", graph, id);
TaskScheduler scheduler = graph(manager, graphSpace, graph)
.taskScheduler();
- return scheduler.task(IdGenerator.of(id)).asMap();
+ return scheduler.task(IdGenerator.of(id), withResult)
+ .asMap(true, withResult);
}
@DELETE
diff --git
a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
index cf390b886..5b117faae 100644
---
a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
+++
b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
@@ -1326,8 +1326,14 @@ public final class HugeGraphAuthProxy implements
HugeGraph {
@Override
public <V> HugeTask<V> task(Id id) {
+ return this.task(id, true);
+ }
+
+ @Override
+ public <V> HugeTask<V> task(Id id, boolean withResult) {
return verifyTaskPermission(HugePermission.READ,
- this.taskScheduler.task(id));
+ this.taskScheduler.task(id,
+ withResult));
}
@Override
@@ -1336,18 +1342,36 @@ public final class HugeGraphAuthProxy implements
HugeGraph {
this.taskScheduler.tasks(ids));
}
+ @Override
+ public <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
+ boolean withResult) {
+ return verifyTaskPermission(HugePermission.READ,
+ this.taskScheduler.tasks(ids,
+ withResult));
+ }
+
@Override
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
long limit, String page) {
Iterator<HugeTask<V>> tasks = this.taskScheduler.tasks(status,
- limit,
page);
+ limit,
+ page);
+ return verifyTaskPermission(HugePermission.READ, tasks);
+ }
+
+ @Override
+ public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
+ long limit, String page,
+ boolean withResult) {
+ Iterator<HugeTask<V>> tasks = this.taskScheduler.tasks(
+ status, limit, page, withResult);
return verifyTaskPermission(HugePermission.READ, tasks);
}
@Override
public <V> HugeTask<V> delete(Id id, boolean force) {
verifyTaskPermission(HugePermission.DELETE,
- this.taskScheduler.task(id));
+ this.taskScheduler.task(id, false));
return this.taskScheduler.delete(id, force);
}
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 b4bba2ea1..3929f8414 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
@@ -306,10 +306,14 @@ public class DistributedTaskScheduler extends
TaskAndResultScheduler {
Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
if (vertex == null) {
+ this.deleteTaskResultFromTx(id);
return null;
}
- HugeTask<V> result = HugeTask.fromVertex(vertex);
- this.tx().removeVertex(vertex);
+ HugeTask<V> result = HugeTask.fromVertex(vertex, false);
+ // Keep the task vertex as a retryable tombstone until its result
+ // vertex is removed; cronSchedule() can rediscover DELETING tasks.
+ this.deleteTaskResultFromTx(id);
+ this.tx().removeTaskVertex(vertex);
return result;
});
}
@@ -322,6 +326,12 @@ public class DistributedTaskScheduler extends
TaskAndResultScheduler {
this.updateStatus(id, null, TaskStatus.DELETING);
return null;
} else {
+ HugeTask<?> task = this.taskWithoutResult(id);
+ if (task != null && task.status() != TaskStatus.DELETING) {
+ initTaskParams(task);
+ task.overwriteStatus(TaskStatus.DELETING);
+ this.save(task);
+ }
return this.deleteFromDB(id);
}
}
@@ -587,7 +597,7 @@ public class DistributedTaskScheduler extends
TaskAndResultScheduler {
}
}
- private boolean isLockedTask(String taskId) {
+ protected boolean isLockedTask(String taskId) {
return MetaManager.instance().isLockedTask(graphSpace,
graphName, taskId);
}
@@ -629,7 +639,7 @@ public class DistributedTaskScheduler extends
TaskAndResultScheduler {
// 1. start task can be from schedule() & cronSchedule()
// 2. recheck the status of task, in case one same task
// called by both methods at same time;
- HugeTask<Object> queryTask = task(this.task.id());
+ HugeTask<Object> queryTask = task(this.task.id(), false);
if (queryTask != null &&
!TaskStatus.NEW.equals(queryTask.status())) {
return;
diff --git
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java
index f9e4f120f..d5d703928 100644
---
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java
+++
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java
@@ -41,6 +41,7 @@ import org.apache.hugegraph.job.ComputerJob;
import org.apache.hugegraph.job.EphemeralJob;
import org.apache.hugegraph.job.GremlinJob;
import org.apache.hugegraph.job.schema.SchemaJob;
+import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.type.define.SerialEnum;
import org.apache.hugegraph.util.Blob;
import org.apache.hugegraph.util.E;
@@ -653,6 +654,11 @@ public class HugeTask<V> extends FutureTask<V> {
}
public synchronized Map<String, Object> asMap(boolean withDetails) {
+ return this.asMap(withDetails, true);
+ }
+
+ public synchronized Map<String, Object> asMap(boolean withDetails,
+ boolean withResult) {
E.checkState(this.type != null, "Task type can't be null");
E.checkState(this.name != null, "Task name can't be null");
@@ -689,7 +695,7 @@ public class HugeTask<V> extends FutureTask<V> {
if (this.input != null) {
map.put(Hidden.unHide(P.INPUT), this.input);
}
- if (this.result != null) {
+ if (withResult && this.result != null) {
map.put(Hidden.unHide(P.RESULT), this.result);
}
}
@@ -697,7 +703,37 @@ public class HugeTask<V> extends FutureTask<V> {
return map;
}
+ synchronized HugeTask<V> copyWithoutResult() {
+ HugeTask<V> task = new HugeTask<>(this.id, this.parent, this.callable);
+ task.type = this.type;
+ task.name = this.name;
+ task.dependencies = this.dependencies == null ?
+ null :
InsertionOrderUtil.newSet(this.dependencies);
+ task.description = this.description;
+ task.context = this.context;
+ task.create = this.create;
+ task.server = this.server;
+ task.load = this.load;
+ task.status = this.status;
+ task.progress = this.progress;
+ task.update = this.update;
+ task.retries = this.retries;
+ task.input = this.input;
+ task.result = null;
+ task.scheduler = this.scheduler;
+ return task;
+ }
+
public static <V> HugeTask<V> fromVertex(Vertex vertex) {
+ return fromVertex(vertex, true);
+ }
+
+ public static <V> HugeTask<V> fromVertex(Vertex vertex,
+ boolean withResult) {
+ if (!withResult && vertex instanceof HugeVertex) {
+ return fromHugeVertex((HugeVertex) vertex);
+ }
+
String callableName = vertex.value(P.CALLABLE);
TaskCallable<V> callable;
try {
@@ -710,11 +746,37 @@ public class HugeTask<V> extends FutureTask<V> {
for (Iterator<VertexProperty<Object>> iter = vertex.properties();
iter.hasNext(); ) {
VertexProperty<Object> prop = iter.next();
+ if (!withResult && P.RESULT.equals(prop.key())) {
+ continue;
+ }
task.property(prop.key(), prop.value());
}
return task;
}
+ private static <V> HugeTask<V> fromHugeVertex(HugeVertex vertex) {
+ String callableName = getPropertyValue(vertex, P.CALLABLE);
+ TaskCallable<V> callable;
+ try {
+ callable = TaskCallable.fromClass(callableName);
+ } catch (Exception e) {
+ callable = TaskCallable.empty(e);
+ }
+
+ HugeTask<V> task = new HugeTask<>(vertex.id(), null, callable);
+ for (String property : P.METADATA_PROPERTIES) {
+ Object value = getPropertyValue(vertex, property);
+ if (value != null) {
+ task.property(property, value);
+ }
+ }
+ return task;
+ }
+
+ private static <V> V getPropertyValue(HugeVertex vertex, String property) {
+ return
vertex.getPropertyValue(vertex.graph().propertyKey(property).id());
+ }
+
private static <V> Collector<V, ?, Set<V>> toOrderSet() {
return Collectors.toCollection(InsertionOrderUtil::newSet);
}
@@ -792,6 +854,11 @@ public class HugeTask<V> extends FutureTask<V> {
public static final String DEPENDENCIES = "~task_dependencies";
public static final String SERVER = "~task_server";
+ private static final String[] METADATA_PROPERTIES = new String[]{
+ TYPE, NAME, CALLABLE, DESCRIPTION, CONTEXT, STATUS, PROGRESS,
+ CREATE, UPDATE, RETRIES, DEPENDENCIES, INPUT, SERVER
+ };
+
//public static final String PARENT = hide("parent");
//public static final String CHILDREN = hide("children");
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 5f60792af..1438249a4 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
@@ -153,7 +153,7 @@ public class StandardTaskScheduler implements TaskScheduler
{
String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null;
do {
Iterator<HugeTask<V>> iter;
- for (iter = this.findTask(status, PAGE_SIZE, page);
+ for (iter = this.findTask(status, PAGE_SIZE, page, false);
iter.hasNext(); ) {
HugeTask<V> task = iter.next();
if (selfServer.equals(task.server())) {
@@ -323,7 +323,8 @@ public class StandardTaskScheduler implements TaskScheduler
{
Collection<HugeServerInfo> serverInfos =
this.serverManager().allServerInfos();
String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null;
do {
- Iterator<HugeTask<Object>> tasks =
this.tasks(TaskStatus.SCHEDULING, PAGE_SIZE, page);
+ Iterator<HugeTask<Object>> tasks =
this.tasks(TaskStatus.SCHEDULING, PAGE_SIZE, page,
+ false);
while (tasks.hasNext()) {
HugeTask<?> task = tasks.next();
if (task.server() != null) {
@@ -365,7 +366,8 @@ public class StandardTaskScheduler implements TaskScheduler
{
protected void executeTasksOnWorker(Id server) {
String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null;
do {
- Iterator<HugeTask<Object>> tasks =
this.tasks(TaskStatus.SCHEDULED, PAGE_SIZE, page);
+ Iterator<HugeTask<Object>> tasks =
this.tasks(TaskStatus.SCHEDULED, PAGE_SIZE, page,
+ false);
while (tasks.hasNext()) {
HugeTask<?> task = tasks.next();
this.initTaskCallable(task);
@@ -394,7 +396,8 @@ public class StandardTaskScheduler implements TaskScheduler
{
protected void cancelTasksOnWorker(Id server) {
String page = this.supportsPaging() ? PageInfo.PAGE_NONE : null;
do {
- Iterator<HugeTask<Object>> tasks =
this.tasks(TaskStatus.CANCELLING, PAGE_SIZE, page);
+ Iterator<HugeTask<Object>> tasks =
this.tasks(TaskStatus.CANCELLING, PAGE_SIZE, page,
+ false);
while (tasks.hasNext()) {
HugeTask<?> task = tasks.next();
Id taskServer = task.server();
@@ -494,24 +497,35 @@ public class StandardTaskScheduler implements
TaskScheduler {
@Override
public <V> HugeTask<V> task(Id id) {
+ return this.task(id, true);
+ }
+
+ @Override
+ public <V> HugeTask<V> task(Id id, boolean withResult) {
E.checkArgumentNotNull(id, "Parameter task id can't be null");
@SuppressWarnings("unchecked")
HugeTask<V> task = (HugeTask<V>) this.tasks.get(id);
if (task != null) {
- return task;
+ return withResult ? task : task.copyWithoutResult();
}
- return this.findTask(id);
+ return this.findTask(id, withResult);
}
@Override
public <V> Iterator<HugeTask<V>> tasks(List<Id> ids) {
+ return this.tasks(ids, true);
+ }
+
+ @Override
+ public <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
+ boolean withResult) {
List<Id> taskIdsNotInMem = new ArrayList<>();
List<HugeTask<V>> taskInMem = new ArrayList<>();
for (Id id : ids) {
@SuppressWarnings("unchecked")
HugeTask<V> task = (HugeTask<V>) this.tasks.get(id);
if (task != null) {
- taskInMem.add(task);
+ taskInMem.add(withResult ? task : task.copyWithoutResult());
} else {
taskIdsNotInMem.add(id);
}
@@ -522,27 +536,37 @@ public class StandardTaskScheduler implements
TaskScheduler {
} else {
iterator = new ExtendableIterator<>(taskInMem.iterator());
}
- iterator.extend(this.findTasks(taskIdsNotInMem));
+ iterator.extend(this.findTasks(taskIdsNotInMem, withResult));
return iterator;
}
@Override
- public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
- long limit, String page) {
+ public <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
+ String page) {
+ return this.tasks(status, limit, page, true);
+ }
+
+ @Override
+ public <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
+ String page, boolean withResult) {
if (status == null) {
- return this.findAllTask(limit, page);
+ return this.findAllTask(limit, page, withResult);
}
- return this.findTask(status, limit, page);
+ return this.findTask(status, limit, page, withResult);
}
public <V> HugeTask<V> findTask(Id id) {
+ return this.findTask(id, true);
+ }
+
+ public <V> HugeTask<V> findTask(Id id, boolean withResult) {
HugeTask<V> result = this.call(() -> {
Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
Vertex vertex = QueryResults.one(vertices);
if (vertex == null) {
return null;
}
- return HugeTask.fromVertex(vertex);
+ return HugeTask.fromVertex(vertex, withResult);
});
if (result == null) {
throw new NotFoundException("Can't find task with id '%s'", id);
@@ -551,23 +575,40 @@ public class StandardTaskScheduler implements
TaskScheduler {
}
public <V> Iterator<HugeTask<V>> findTasks(List<Id> ids) {
- return this.queryTask(ids);
+ return this.findTasks(ids, true);
+ }
+
+ public <V> Iterator<HugeTask<V>> findTasks(List<Id> ids,
+ boolean withResult) {
+ return this.queryTask(ids, withResult);
}
public <V> Iterator<HugeTask<V>> findAllTask(long limit, String page) {
- return this.queryTask(ImmutableMap.of(), limit, page);
+ return this.findAllTask(limit, page, true);
+ }
+
+ public <V> Iterator<HugeTask<V>> findAllTask(long limit, String page,
+ boolean withResult) {
+ return this.queryTask(ImmutableMap.of(), limit, page, withResult);
}
public <V> Iterator<HugeTask<V>> findTask(TaskStatus status,
long limit, String page) {
- return this.queryTask(P.STATUS, status.code(), limit, page);
+ return this.findTask(status, limit, page, true);
+ }
+
+ public <V> Iterator<HugeTask<V>> findTask(TaskStatus status,
+ long limit, String page,
+ boolean withResult) {
+ return this.queryTask(P.STATUS, status.code(), limit, page,
+ withResult);
}
@Override
public <V> HugeTask<V> delete(Id id, boolean force) {
this.checkOnMasterNode("delete");
- HugeTask<?> task = this.task(id);
+ HugeTask<?> task = this.task(id, false);
/*
* The following is out of date when the task running on worker node:
* HugeTask<?> task = this.tasks.get(id);
@@ -592,11 +633,11 @@ public class StandardTaskScheduler implements
TaskScheduler {
if (vertex == null) {
return null;
}
- HugeTask<V> result = HugeTask.fromVertex(vertex);
+ HugeTask<V> result = HugeTask.fromVertex(vertex, false);
E.checkState(force || result.completed(),
"Can't delete incomplete task '%s' in status %s",
id, result.status());
- this.tx().removeVertex(vertex);
+ this.tx().removeTaskVertex(vertex);
return result;
});
}
@@ -674,11 +715,24 @@ public class StandardTaskScheduler implements
TaskScheduler {
private <V> Iterator<HugeTask<V>> queryTask(String key, Object value,
long limit, String page) {
- return this.queryTask(ImmutableMap.of(key, value), limit, page);
+ return this.queryTask(key, value, limit, page, true);
+ }
+
+ private <V> Iterator<HugeTask<V>> queryTask(String key, Object value,
+ long limit, String page,
+ boolean withResult) {
+ return this.queryTask(ImmutableMap.of(key, value), limit, page,
+ withResult);
}
private <V> Iterator<HugeTask<V>> queryTask(Map<String, Object> conditions,
long limit, String page) {
+ return this.queryTask(conditions, limit, page, true);
+ }
+
+ private <V> Iterator<HugeTask<V>> queryTask(Map<String, Object> conditions,
+ long limit, String page,
+ boolean withResult) {
return this.call(() -> {
ConditionQuery query;
if
(this.graph.backendStoreFeatures().supportsTaskAndServerVertex()) {
@@ -701,18 +755,27 @@ public class StandardTaskScheduler implements
TaskScheduler {
}
Iterator<Vertex> vertices = this.tx().queryVertices(query);
Iterator<HugeTask<V>> tasks =
- new MapperIterator<>(vertices, HugeTask::fromVertex);
+ new MapperIterator<>(vertices, vertex -> {
+ return HugeTask.fromVertex(vertex, withResult);
+ });
// Convert iterator to list to avoid across thread tx accessed
return QueryResults.toList(tasks);
});
}
private <V> Iterator<HugeTask<V>> queryTask(List<Id> ids) {
+ return this.queryTask(ids, true);
+ }
+
+ private <V> Iterator<HugeTask<V>> queryTask(List<Id> ids,
+ boolean withResult) {
return this.call(() -> {
Object[] idArray = ids.toArray(new Id[0]);
Iterator<Vertex> vertices = this.tx().queryTaskInfos(idArray);
Iterator<HugeTask<V>> tasks =
- new MapperIterator<>(vertices, HugeTask::fromVertex);
+ new MapperIterator<>(vertices, vertex -> {
+ return HugeTask.fromVertex(vertex, withResult);
+ });
// Convert iterator to list to avoid across thread tx accessed
return QueryResults.toList(tasks);
});
diff --git
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java
index 2ba3fd8a6..26a520bde 100644
---
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java
+++
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskAndResultScheduler.java
@@ -104,22 +104,29 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
@Override
public <V> HugeTask<V> task(Id id) {
+ return this.task(id, true);
+ }
+
+ @Override
+ public <V> HugeTask<V> task(Id id, boolean withResult) {
HugeTask<V> task = this.call(() -> {
Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
Vertex vertex = QueryResults.one(vertices);
if (vertex == null) {
return null;
}
- return HugeTask.fromVertex(vertex);
+ return HugeTask.fromVertex(vertex, withResult);
});
if (task == null) {
throw new NotFoundException("Can't find task with id '%s'", id);
}
- HugeTaskResult taskResult = queryTaskResult(id);
- if (taskResult != null) {
- task.result(taskResult);
+ if (withResult) {
+ HugeTaskResult taskResult = queryTaskResult(id);
+ if (taskResult != null) {
+ task.result(taskResult);
+ }
}
return task;
@@ -127,17 +134,39 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
@Override
public <V> Iterator<HugeTask<V>> tasks(List<Id> ids) {
- return this.tasksWithoutResult(ids);
+ return this.tasks(ids, false);
+ }
+
+ @Override
+ public <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
+ boolean withResult) {
+ if (!withResult) {
+ return this.tasksWithoutResult(ids);
+ }
+ return this.queryTask(ids);
}
@Override
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
String page) {
+ return this.tasks(status, limit, page, false);
+ }
+
+ @Override
+ public <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
+ String page, boolean withResult) {
+ if (!withResult) {
+ if (status == null) {
+ return this.queryTaskWithoutResult(ImmutableMap.of(), limit,
+ page);
+ }
+ return this.queryTaskWithoutResult(HugeTask.P.STATUS,
+ status.code(), limit, page);
+ }
if (status == null) {
- return this.queryTaskWithoutResult(ImmutableMap.of(), limit, page);
+ return this.queryTask(ImmutableMap.of(), limit, page);
}
- return this.queryTaskWithoutResult(HugeTask.P.STATUS, status.code(),
- limit, page);
+ return this.queryTask(HugeTask.P.STATUS, status.code(), limit, page);
}
protected <V> Iterator<HugeTask<V>> queryTask(String key, Object value,
@@ -216,7 +245,7 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
if (vertex == null) {
return null;
}
- return HugeTask.fromVertex(vertex);
+ return HugeTask.fromVertex(vertex, false);
});
return result;
@@ -227,7 +256,9 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
Object[] idArray = ids.toArray(new Id[ids.size()]);
Iterator<Vertex> vertices = this.tx().queryTaskInfos(idArray);
Iterator<HugeTask<V>> tasks =
- new MapperIterator<>(vertices, HugeTask::fromVertex);
+ new MapperIterator<>(vertices, vertex -> {
+ return HugeTask.fromVertex(vertex, false);
+ });
// Convert iterator to list to avoid across thread tx accessed
return QueryResults.toList(tasks);
});
@@ -268,12 +299,23 @@ public abstract class TaskAndResultScheduler implements
TaskScheduler {
}
Iterator<Vertex> vertices = this.tx().queryTaskInfos(query);
Iterator<HugeTask<V>> tasks =
- new MapperIterator<>(vertices, HugeTask::fromVertex);
+ new MapperIterator<>(vertices, vertex -> {
+ return HugeTask.fromVertex(vertex, false);
+ });
// Convert iterator to list to avoid across thread tx accessed
return QueryResults.toList(tasks);
});
}
+ protected void deleteTaskResultFromTx(Id taskId) {
+ Iterator<Vertex> vertices =
+ this.tx().queryTaskInfos(HugeTaskResult.genId(taskId));
+ HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
+ if (vertex != null) {
+ this.tx().removeTaskVertex(vertex);
+ }
+ }
+
protected HugeTaskResult queryTaskResult(Id taskid) {
HugeTaskResult result = this.call(() -> {
Iterator<Vertex> vertices =
diff --git
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskScheduler.java
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskScheduler.java
index af789c523..ea10cd006 100644
---
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskScheduler.java
+++
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskScheduler.java
@@ -49,11 +49,25 @@ public interface TaskScheduler {
<V> HugeTask<V> task(Id id);
+ default <V> HugeTask<V> task(Id id, boolean withResult) {
+ return this.task(id);
+ }
+
<V> Iterator<HugeTask<V>> tasks(List<Id> ids);
+ default <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
+ boolean withResult) {
+ return this.tasks(ids);
+ }
+
<V> Iterator<HugeTask<V>> tasks(TaskStatus status,
long limit, String page);
+ default <V> Iterator<HugeTask<V>> tasks(TaskStatus status, long limit,
+ String page, boolean withResult) {
+ return this.tasks(status, limit, page);
+ }
+
void init();
boolean close();
diff --git
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskTransaction.java
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskTransaction.java
index 2b27019c7..9dd4a46aa 100644
---
a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskTransaction.java
+++
b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskTransaction.java
@@ -31,6 +31,7 @@ import org.apache.hugegraph.schema.IndexLabel;
import org.apache.hugegraph.schema.PropertyKey;
import org.apache.hugegraph.schema.SchemaManager;
import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.structure.HugeIndex;
import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.type.define.Cardinality;
@@ -74,6 +75,34 @@ public class TaskTransaction extends GraphTransaction {
return false;
}
+ public void removeTaskVertex(HugeVertex vertex) {
+ this.checkOwnerThread();
+
+ this.beforeWrite();
+
+ this.doRemove(this.serializer.writeVertex(vertex.prepareRemoved()));
+ if (TASK.equals(vertex.schemaLabel().name())) {
+ this.updateIndex(this.indexLabel(HugeTask.P.STATUS).id(),
+ vertex, true);
+ }
+ this.removeLabelIndex(vertex);
+
+ this.afterWrite();
+ }
+
+ private void removeLabelIndex(HugeVertex vertex) {
+ if (this.store().features().supportsQueryByLabel() ||
+ !vertex.schemaLabel().enableLabelIndex()) {
+ return;
+ }
+
+ HugeIndex index = new HugeIndex(this.graph(),
+ IndexLabel.label(vertex.type()));
+ index.fieldValues(vertex.schemaLabel().id());
+ index.elementIds(vertex.id(), vertex.expiredTime());
+ this.doEliminate(this.serializer.writeIndex(index));
+ }
+
public void initSchema() {
if (this.existVertexLabel(TASK)) {
return;
diff --git
a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/TaskApiTest.java
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/TaskApiTest.java
index 9ed25fd71..f65f651b1 100644
---
a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/TaskApiTest.java
+++
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/TaskApiTest.java
@@ -87,6 +87,28 @@ public class TaskApiTest extends BaseApiTest {
Assert.assertEquals("success", status);
}
+ @Test
+ public void testGetWithoutResult() {
+ int taskId = this.gremlinJob("1 + 2");
+
+ waitTaskSuccess(taskId);
+
+ Response r = client().get(PATH, ImmutableMap.of("limit", -1));
+ String content = assertResponseStatus(200, r);
+ Assert.assertFalse(content, content.contains("task_result"));
+
+ r = client().get(PATH, String.valueOf(taskId));
+ content = assertResponseStatus(200, r);
+ assertJsonContains(content, "task_result");
+
+ r = client().get(PATH + taskId,
+ ImmutableMap.of("with_result", false));
+ content = assertResponseStatus(200, r);
+ assertJsonContains(content, "id");
+ assertJsonContains(content, "task_callable");
+ Assert.assertFalse(content, content.contains("task_result"));
+ }
+
@Test
public void testCancel() {
// create a task
@@ -143,8 +165,12 @@ public class TaskApiTest extends BaseApiTest {
}
private int gremlinJob() {
+ return this.gremlinJob("Thread.sleep(1000L)");
+ }
+
+ private int gremlinJob(String gremlin) {
String body = "{" +
- "\"gremlin\":\"Thread.sleep(1000L)\"," +
+ "\"gremlin\":\"" + gremlin + "\"," +
"\"bindings\":{}," +
"\"language\":\"gremlin-groovy\"," +
"\"aliases\":{}}";
diff --git
a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java
index e78aff7e2..230b8d2d0 100644
---
a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java
+++
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/core/CoreTestSuite.java
@@ -23,6 +23,7 @@ import org.apache.hugegraph.dist.RegisterUtil;
import org.apache.hugegraph.masterelection.GlobalMasterInfo;
import org.apache.hugegraph.meta.MetaManager;
import org.apache.hugegraph.meta.PdMetaDriver;
+import org.apache.hugegraph.task.TaskAndResultSchedulerTest;
import org.apache.hugegraph.testutil.Utils;
import org.apache.hugegraph.util.Log;
import org.junit.AfterClass;
@@ -45,6 +46,7 @@ import org.slf4j.Logger;
PropertyCoreTest.EdgePropertyCoreTest.class,
RestoreCoreTest.class,
TaskCoreTest.class,
+ TaskAndResultSchedulerTest.class,
AuthTest.class,
MultiGraphsTest.class,
RamTableTest.class
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 212ccc058..2d106fc3b 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
@@ -115,6 +115,88 @@ public class TaskCoreTest extends BaseCoreTest {
});
}
+ @Test
+ public void testTaskWithoutResult() throws TimeoutException {
+ HugeGraph graph = graph();
+ TaskScheduler scheduler = graph.taskScheduler();
+
+ Id id = IdGenerator.of(88889);
+ HugeTask<?> task = new HugeTask<>(id, null, new SleepCallable<>());
+ task.type("test");
+ task.name("metadata-task-in-memory");
+ scheduler.schedule(task);
+
+ try {
+ Whitebox.setInternalState(task, "result", "\"in-memory-result\"");
+
+ HugeTask<?> taskWithoutResult = scheduler.task(id, false);
+ Assert.assertEquals("metadata-task-in-memory",
+ taskWithoutResult.name());
+ Assert.assertNull(taskWithoutResult.result());
+
+ Iterator<HugeTask<Object>> iter =
scheduler.tasks(ImmutableList.of(id),
+ false);
+ Assert.assertTrue(iter.hasNext());
+ taskWithoutResult = iter.next();
+ Assert.assertEquals("metadata-task-in-memory",
+ taskWithoutResult.name());
+ Assert.assertNull(taskWithoutResult.result());
+ Assert.assertFalse(iter.hasNext());
+ } finally {
+ Whitebox.setInternalState(task, "result", null);
+ }
+
+ scheduler.waitUntilTaskCompleted(id, 10);
+ scheduler.delete(id, false);
+
+ id = IdGenerator.of(88890);
+ task = new HugeTask<>(id, null, new MetadataResultCallable());
+ task.type("test");
+ task.name("metadata-task");
+ scheduler.schedule(task);
+
+ scheduler.waitUntilTaskCompleted(id, 10);
+
+ HugeTask<?> taskWithResult = scheduler.task(id, true);
+ Assert.assertEquals("\"metadata-result\"", taskWithResult.result());
+
+ HugeTask<?> taskWithoutResult = scheduler.task(id, false);
+ Assert.assertEquals("metadata-task", taskWithoutResult.name());
+ Assert.assertNull(taskWithoutResult.result());
+
+ Iterator<HugeTask<Object>> iter =
scheduler.tasks(ImmutableList.of(id));
+ Assert.assertTrue(iter.hasNext());
+ taskWithResult = iter.next();
+ Assert.assertEquals("metadata-task", taskWithResult.name());
+ Assert.assertEquals("\"metadata-result\"", taskWithResult.result());
+ Assert.assertFalse(iter.hasNext());
+
+ iter = scheduler.tasks(ImmutableList.of(id), false);
+ Assert.assertTrue(iter.hasNext());
+ taskWithoutResult = iter.next();
+ Assert.assertEquals("metadata-task", taskWithoutResult.name());
+ Assert.assertNull(taskWithoutResult.result());
+ Assert.assertFalse(iter.hasNext());
+
+ iter = scheduler.tasks(TaskStatus.SUCCESS, 10, null, false);
+ Assert.assertTrue(iter.hasNext());
+ taskWithoutResult = iter.next();
+ Assert.assertEquals("metadata-task", taskWithoutResult.name());
+ Assert.assertNull(taskWithoutResult.result());
+
+ iter = scheduler.tasks(TaskStatus.SUCCESS, 10, null);
+ Assert.assertTrue(iter.hasNext());
+ taskWithResult = iter.next();
+ Assert.assertEquals("metadata-task", taskWithResult.name());
+ Assert.assertEquals("\"metadata-result\"", taskWithResult.result());
+
+ Id taskId = id;
+ scheduler.delete(taskId, false);
+ Assert.assertThrows(NotFoundException.class, () -> {
+ scheduler.task(taskId);
+ });
+ }
+
@Test
public void testTaskWithFailure() throws TimeoutException {
HugeGraph graph = graph();
@@ -714,4 +796,21 @@ public class TaskCoreTest extends BaseCoreTest {
this.graph().taskScheduler().save(this.task());
}
}
+
+ public static class MetadataResultCallable extends TaskCallable<String> {
+
+ public MetadataResultCallable() {
+ // pass
+ }
+
+ @Override
+ public String call() {
+ return "metadata-result";
+ }
+
+ @Override
+ public void done() {
+ this.graph().taskScheduler().save(this.task());
+ }
+ }
}
diff --git
a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/task/TaskAndResultSchedulerTest.java
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/task/TaskAndResultSchedulerTest.java
new file mode 100644
index 000000000..15fabf130
--- /dev/null
+++
b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/task/TaskAndResultSchedulerTest.java
@@ -0,0 +1,348 @@
+/*
+ * 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.task;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.query.QueryResults;
+import org.apache.hugegraph.core.BaseCoreTest;
+import org.apache.hugegraph.structure.HugeVertex;
+import org.apache.hugegraph.testutil.Assert;
+import org.apache.hugegraph.testutil.Whitebox;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
+
+public class TaskAndResultSchedulerTest extends BaseCoreTest {
+
+ @Test
+ public void testMetadataOnlyReadsSkipLargeCompressedTaskResult() {
+ Id id = IdGenerator.of(88901);
+ TestDistributedTaskScheduler scheduler = this.newScheduler();
+
+ try {
+ scheduler.init();
+
+ String largeResult = largeTaskResult();
+ HugeTask<Object> task = newTask(id, "large-result-task",
+ TaskStatus.SUCCESS, largeResult);
+ scheduler.save(task);
+ Assert.assertTrue(scheduler.taskResultVertexExists(id));
+
+ HugeTask<?> taskWithResult = scheduler.task(id, true);
+ Assert.assertEquals(largeResult, taskWithResult.result());
+ Assert.assertGt(0, scheduler.resultReadCount());
+
+ scheduler.resetResultReadCount();
+ scheduler.forbidResultRead(true);
+
+ HugeTask<?> taskWithoutResult = scheduler.task(id, false);
+ assertTaskWithoutResult(taskWithoutResult, id);
+ Assert.assertEquals(0, scheduler.resultReadCount());
+
+ Iterator<HugeTask<Object>> tasks =
scheduler.tasks(ImmutableList.of(id),
+ false);
+ assertIteratorContainsTaskWithoutResult(tasks, id);
+ Assert.assertEquals(0, scheduler.resultReadCount());
+
+ tasks = scheduler.tasks(TaskStatus.SUCCESS, 10L, null, false);
+ assertIteratorContainsTaskWithoutResult(tasks, id);
+ Assert.assertEquals(0, scheduler.resultReadCount());
+
+ tasks = scheduler.tasks(ImmutableList.of(id));
+ assertIteratorContainsTaskWithoutResult(tasks, id);
+ Assert.assertEquals(0, scheduler.resultReadCount());
+
+ tasks = scheduler.tasks(TaskStatus.SUCCESS, 10L, null);
+ assertIteratorContainsTaskWithoutResult(tasks, id);
+ Assert.assertEquals(0, scheduler.resultReadCount());
+
+ tasks = scheduler.queryMetadataTasksByStatus(TaskStatus.SUCCESS);
+ assertIteratorContainsTaskWithoutResult(tasks, id);
+ Assert.assertEquals(0, scheduler.resultReadCount());
+
+ scheduler.forbidResultRead(false);
+ scheduler.resetResultReadCount();
+
+ tasks = scheduler.tasks(ImmutableList.of(id), true);
+ assertIteratorContainsTaskWithResult(tasks, id, largeResult);
+ Assert.assertGt(0, scheduler.resultReadCount());
+ } finally {
+ scheduler.forbidResultRead(false);
+ scheduler.deleteFromDBForTest(id);
+ scheduler.closeAndShutdown();
+ }
+ }
+
+ @Test
+ public void testDistributedDeleteKeepsTaskResultRecoverable() {
+ Id id = IdGenerator.of(88902);
+ TestDistributedTaskScheduler scheduler = this.newScheduler();
+
+ try {
+ scheduler.init();
+
+ HugeTask<Object> task = newTask(id, "delete-order-task",
+ TaskStatus.SUCCESS, "\"result\"");
+ scheduler.save(task);
+ Assert.assertTrue(scheduler.taskVertexExists(id));
+ Assert.assertTrue(scheduler.taskResultVertexExists(id));
+
+ scheduler.failNextTaskResultDelete();
+ Assert.assertThrows(HugeException.class, () -> {
+ scheduler.delete(id, true);
+ });
+
+ Assert.assertTrue(scheduler.taskVertexExists(id));
+ Assert.assertEquals(TaskStatus.DELETING, scheduler.taskStatus(id));
+ Assert.assertTrue(scheduler.taskResultVertexExists(id));
+
+ scheduler.cronSchedule();
+ Assert.assertFalse(scheduler.taskVertexExists(id));
+ Assert.assertFalse(scheduler.taskResultVertexExists(id));
+ } finally {
+ scheduler.allowTaskResultDelete();
+ scheduler.deleteFromDBForTest(id);
+ scheduler.closeAndShutdown();
+ }
+ }
+
+ private TestDistributedTaskScheduler newScheduler() {
+ return new TestDistributedTaskScheduler(this.params());
+ }
+
+ private static HugeTask<Object> newTask(Id id, String name,
+ TaskStatus status, String result) {
+ HugeTask<Object> task = new HugeTask<>(id, null, new EmptyCallable());
+ task.type("test");
+ task.name(name);
+ task.overwriteStatus(status);
+ Whitebox.setInternalState(task, "result", result);
+ return task;
+ }
+
+ private static void assertTaskWithoutResult(HugeTask<?> task, Id id) {
+ Assert.assertEquals(id, task.id());
+ Assert.assertNull(task.result());
+ }
+
+ private static void assertIteratorContainsTaskWithoutResult(
+ Iterator<HugeTask<Object>> tasks, Id id) {
+ boolean matched = false;
+ while (tasks.hasNext()) {
+ HugeTask<?> task = tasks.next();
+ if (id.equals(task.id())) {
+ assertTaskWithoutResult(task, id);
+ matched = true;
+ }
+ }
+ Assert.assertTrue(matched);
+ }
+
+ private static void assertIteratorContainsTaskWithResult(
+ Iterator<HugeTask<Object>> tasks, Id id, String result) {
+ boolean matched = false;
+ while (tasks.hasNext()) {
+ HugeTask<?> task = tasks.next();
+ if (id.equals(task.id())) {
+ Assert.assertEquals(id, task.id());
+ Assert.assertEquals(result, task.result());
+ matched = true;
+ }
+ }
+ Assert.assertTrue(matched);
+ }
+
+ private static String largeTaskResult() {
+ char[] chars = new char[1024 * 1024 + 257];
+ int value = 0x12345678;
+ for (int i = 0; i < chars.length; i++) {
+ value = value * 1103515245 + 12345;
+ chars[i] = (char) ('a' + ((value >>> 16) & 0x0F));
+ }
+ return new String(chars);
+ }
+
+ public static class EmptyCallable extends TaskCallable<Object> {
+
+ public EmptyCallable() {
+ // pass
+ }
+
+ @Override
+ public Object call() {
+ return null;
+ }
+ }
+
+ private static class TestDistributedTaskScheduler
+ extends DistributedTaskScheduler {
+
+ private final ScheduledThreadPoolExecutor schedulerExecutor;
+ private final ExecutorService taskDbExecutor;
+ private final ExecutorService schemaTaskExecutor;
+ private final ExecutorService olapTaskExecutor;
+ private final ExecutorService gremlinTaskExecutor;
+ private final ExecutorService ephemeralTaskExecutor;
+ private final ExecutorService serverInfoDbExecutor;
+ private final AtomicInteger resultReadCount;
+ private volatile boolean forbidResultRead;
+ private volatile boolean failNextTaskResultDelete;
+
+ TestDistributedTaskScheduler(HugeGraphParams graph) {
+ this(graph, new ScheduledThreadPoolExecutor(1),
+ Executors.newSingleThreadExecutor(),
+ Executors.newSingleThreadExecutor(),
+ Executors.newSingleThreadExecutor(),
+ Executors.newSingleThreadExecutor(),
+ Executors.newSingleThreadExecutor(),
+ Executors.newSingleThreadExecutor());
+ }
+
+ private TestDistributedTaskScheduler(
+ HugeGraphParams graph,
+ ScheduledThreadPoolExecutor schedulerExecutor,
+ ExecutorService taskDbExecutor,
+ ExecutorService schemaTaskExecutor,
+ ExecutorService olapTaskExecutor,
+ ExecutorService gremlinTaskExecutor,
+ ExecutorService ephemeralTaskExecutor,
+ ExecutorService serverInfoDbExecutor) {
+ super(graph, schedulerExecutor, taskDbExecutor, schemaTaskExecutor,
+ olapTaskExecutor, gremlinTaskExecutor, ephemeralTaskExecutor,
+ serverInfoDbExecutor);
+ this.schedulerExecutor = schedulerExecutor;
+ this.taskDbExecutor = taskDbExecutor;
+ this.schemaTaskExecutor = schemaTaskExecutor;
+ this.olapTaskExecutor = olapTaskExecutor;
+ this.gremlinTaskExecutor = gremlinTaskExecutor;
+ this.ephemeralTaskExecutor = ephemeralTaskExecutor;
+ this.serverInfoDbExecutor = serverInfoDbExecutor;
+ this.resultReadCount = new AtomicInteger();
+ this.forbidResultRead = false;
+ this.failNextTaskResultDelete = false;
+ }
+
+ @Override
+ protected HugeTaskResult queryTaskResult(Id taskid) {
+ this.resultReadCount.incrementAndGet();
+ this.checkResultReadAllowed();
+ return super.queryTaskResult(taskid);
+ }
+
+ @Override
+ protected Iterator<HugeTaskResult> queryTaskResult(List<Id> taskIds) {
+ this.resultReadCount.incrementAndGet();
+ this.checkResultReadAllowed();
+ return super.queryTaskResult(taskIds);
+ }
+
+ @Override
+ protected void deleteTaskResultFromTx(Id taskId) {
+ if (this.failNextTaskResultDelete) {
+ this.failNextTaskResultDelete = false;
+ throw new HugeException("Mock task result delete failure");
+ }
+ super.deleteTaskResultFromTx(taskId);
+ }
+
+ @Override
+ protected boolean isLockedTask(String taskId) {
+ return false;
+ }
+
+ public <V> HugeTask<V> deleteFromDBForTest(Id id) {
+ return super.deleteFromDB(id);
+ }
+
+ public Iterator<HugeTask<Object>> queryMetadataTasksByStatus(
+ TaskStatus status) {
+ return super.queryTaskWithoutResultByStatus(status);
+ }
+
+ public boolean taskVertexExists(Id id) {
+ return this.vertexExists(id);
+ }
+
+ public boolean taskResultVertexExists(Id id) {
+ return this.vertexExists(HugeTaskResult.genId(id));
+ }
+
+ public TaskStatus taskStatus(Id id) {
+ return this.task(id, false).status();
+ }
+
+ public int resultReadCount() {
+ return this.resultReadCount.get();
+ }
+
+ public void resetResultReadCount() {
+ this.resultReadCount.set(0);
+ }
+
+ public void forbidResultRead(boolean forbid) {
+ this.forbidResultRead = forbid;
+ }
+
+ public void failNextTaskResultDelete() {
+ this.failNextTaskResultDelete = true;
+ }
+
+ public void allowTaskResultDelete() {
+ this.failNextTaskResultDelete = false;
+ }
+
+ public void closeAndShutdown() {
+ try {
+ this.close();
+ } finally {
+ this.schedulerExecutor.shutdownNow();
+ this.taskDbExecutor.shutdownNow();
+ this.schemaTaskExecutor.shutdownNow();
+ this.olapTaskExecutor.shutdownNow();
+ this.gremlinTaskExecutor.shutdownNow();
+ this.ephemeralTaskExecutor.shutdownNow();
+ this.serverInfoDbExecutor.shutdownNow();
+ }
+ }
+
+ private boolean vertexExists(Object id) {
+ return this.call(() -> {
+ Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
+ HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
+ return vertex != null;
+ });
+ }
+
+ private void checkResultReadAllowed() {
+ if (this.forbidResultRead) {
+ throw new AssertionError("Unexpected task result read");
+ }
+ }
+ }
+}
