Skip to content

Commit 8c19b62

Browse files
dpol1dpol1
committed
Merge branch 'master' into fix/3036-meta-listener-reconnect
2 parents dd81bc7 + 63c97ff commit 8c19b62

12 files changed

Lines changed: 777 additions & 47 deletions

File tree

‎hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -102,12 +102,13 @@ public Map<String, Object> list(@Context GraphManager manager,
102102
limit = NO_LIMIT;
103103
List<Id> idList = ids.stream().map(IdGenerator::of)
104104
.collect(Collectors.toList());
105-
iter = scheduler.tasks(idList);
105+
iter = scheduler.tasks(idList, false);
106106
} else {
107107
if (status == null) {
108-
iter = scheduler.tasks(null, limit, page);
108+
iter = scheduler.tasks(null, limit, page, false);
109109
} else {
110-
iter = scheduler.tasks(parseStatus(status), limit, page);
110+
iter = scheduler.tasks(parseStatus(status), limit, page,
111+
false);
111112
}
112113
}
113114

@@ -136,12 +137,17 @@ public Map<String, Object> get(@Context GraphManager manager,
136137
@Parameter(description = "The graph name")
137138
@PathParam("graph") String graph,
138139
@Parameter(description = "The task id")
139-
@PathParam("id") long id) {
140+
@PathParam("id") long id,
141+
@Parameter(description = "Whether to load task result")
142+
@DefaultValue("true")
143+
@QueryParam("with_result")
144+
boolean withResult) {
140145
LOG.debug("Graph [{}] get task: {}", graph, id);
141146

142147
TaskScheduler scheduler = graph(manager, graphSpace, graph)
143148
.taskScheduler();
144-
return scheduler.task(IdGenerator.of(id)).asMap();
149+
return scheduler.task(IdGenerator.of(id), withResult)
150+
.asMap(true, withResult);
145151
}
146152

147153
@DELETE

‎hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java‎

Lines changed: 27 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1326,8 +1326,14 @@ public <V> void save(HugeTask<V> task) {
13261326

13271327
@Override
13281328
public <V> HugeTask<V> task(Id id) {
1329+
return this.task(id, true);
1330+
}
1331+
1332+
@Override
1333+
public <V> HugeTask<V> task(Id id, boolean withResult) {
13291334
return verifyTaskPermission(HugePermission.READ,
1330-
this.taskScheduler.task(id));
1335+
this.taskScheduler.task(id,
1336+
withResult));
13311337
}
13321338

13331339
@Override
@@ -1336,18 +1342,36 @@ public <V> Iterator<HugeTask<V>> tasks(List<Id> ids) {
13361342
this.taskScheduler.tasks(ids));
13371343
}
13381344

1345+
@Override
1346+
public <V> Iterator<HugeTask<V>> tasks(List<Id> ids,
1347+
boolean withResult) {
1348+
return verifyTaskPermission(HugePermission.READ,
1349+
this.taskScheduler.tasks(ids,
1350+
withResult));
1351+
}
1352+
13391353
@Override
13401354
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
13411355
long limit, String page) {
13421356
Iterator<HugeTask<V>> tasks = this.taskScheduler.tasks(status,
1343-
limit, page);
1357+
limit,
1358+
page);
1359+
return verifyTaskPermission(HugePermission.READ, tasks);
1360+
}
1361+
1362+
@Override
1363+
public <V> Iterator<HugeTask<V>> tasks(TaskStatus status,
1364+
long limit, String page,
1365+
boolean withResult) {
1366+
Iterator<HugeTask<V>> tasks = this.taskScheduler.tasks(
1367+
status, limit, page, withResult);
13441368
return verifyTaskPermission(HugePermission.READ, tasks);
13451369
}
13461370

13471371
@Override
13481372
public <V> HugeTask<V> delete(Id id, boolean force) {
13491373
verifyTaskPermission(HugePermission.DELETE,
1350-
this.taskScheduler.task(id));
1374+
this.taskScheduler.task(id, false));
13511375
return this.taskScheduler.delete(id, force);
13521376
}
13531377

‎hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/DistributedTaskScheduler.java‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -306,10 +306,14 @@ protected <V> HugeTask<V> deleteFromDB(Id id) {
306306
Iterator<Vertex> vertices = this.tx().queryTaskInfos(id);
307307
HugeVertex vertex = (HugeVertex) QueryResults.one(vertices);
308308
if (vertex == null) {
309+
this.deleteTaskResultFromTx(id);
309310
return null;
310311
}
311-
HugeTask<V> result = HugeTask.fromVertex(vertex);
312-
this.tx().removeVertex(vertex);
312+
HugeTask<V> result = HugeTask.fromVertex(vertex, false);
313+
// Keep the task vertex as a retryable tombstone until its result
314+
// vertex is removed; cronSchedule() can rediscover DELETING tasks.
315+
this.deleteTaskResultFromTx(id);
316+
this.tx().removeTaskVertex(vertex);
313317
return result;
314318
});
315319
}
@@ -322,6 +326,12 @@ public <V> HugeTask<V> delete(Id id, boolean force) {
322326
this.updateStatus(id, null, TaskStatus.DELETING);
323327
return null;
324328
} else {
329+
HugeTask<?> task = this.taskWithoutResult(id);
330+
if (task != null && task.status() != TaskStatus.DELETING) {
331+
initTaskParams(task);
332+
task.overwriteStatus(TaskStatus.DELETING);
333+
this.save(task);
334+
}
325335
return this.deleteFromDB(id);
326336
}
327337
}
@@ -587,7 +597,7 @@ private void unlockTask(String taskId, LockResult lockResult) {
587597
}
588598
}
589599

590-
private boolean isLockedTask(String taskId) {
600+
protected boolean isLockedTask(String taskId) {
591601
return MetaManager.instance().isLockedTask(graphSpace,
592602
graphName, taskId);
593603
}
@@ -629,7 +639,7 @@ public void run() {
629639
// 1. start task can be from schedule() & cronSchedule()
630640
// 2. recheck the status of task, in case one same task
631641
// called by both methods at same time;
632-
HugeTask<Object> queryTask = task(this.task.id());
642+
HugeTask<Object> queryTask = task(this.task.id(), false);
633643
if (queryTask != null &&
634644
!TaskStatus.NEW.equals(queryTask.status())) {
635645
return;

‎hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java‎

Lines changed: 68 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
import org.apache.hugegraph.job.EphemeralJob;
4242
import org.apache.hugegraph.job.GremlinJob;
4343
import org.apache.hugegraph.job.schema.SchemaJob;
44+
import org.apache.hugegraph.structure.HugeVertex;
4445
import org.apache.hugegraph.type.define.SerialEnum;
4546
import org.apache.hugegraph.util.Blob;
4647
import org.apache.hugegraph.util.E;
@@ -653,6 +654,11 @@ public Map<String, Object> asMap() {
653654
}
654655

655656
public synchronized Map<String, Object> asMap(boolean withDetails) {
657+
return this.asMap(withDetails, true);
658+
}
659+
660+
public synchronized Map<String, Object> asMap(boolean withDetails,
661+
boolean withResult) {
656662
E.checkState(this.type != null, "Task type can't be null");
657663
E.checkState(this.name != null, "Task name can't be null");
658664

@@ -689,15 +695,45 @@ public synchronized Map<String, Object> asMap(boolean withDetails) {
689695
if (this.input != null) {
690696
map.put(Hidden.unHide(P.INPUT), this.input);
691697
}
692-
if (this.result != null) {
698+
if (withResult && this.result != null) {
693699
map.put(Hidden.unHide(P.RESULT), this.result);
694700
}
695701
}
696702

697703
return map;
698704
}
699705

706+
synchronized HugeTask<V> copyWithoutResult() {
707+
HugeTask<V> task = new HugeTask<>(this.id, this.parent, this.callable);
708+
task.type = this.type;
709+
task.name = this.name;
710+
task.dependencies = this.dependencies == null ?
711+
null : InsertionOrderUtil.newSet(this.dependencies);
712+
task.description = this.description;
713+
task.context = this.context;
714+
task.create = this.create;
715+
task.server = this.server;
716+
task.load = this.load;
717+
task.status = this.status;
718+
task.progress = this.progress;
719+
task.update = this.update;
720+
task.retries = this.retries;
721+
task.input = this.input;
722+
task.result = null;
723+
task.scheduler = this.scheduler;
724+
return task;
725+
}
726+
700727
public static <V> HugeTask<V> fromVertex(Vertex vertex) {
728+
return fromVertex(vertex, true);
729+
}
730+
731+
public static <V> HugeTask<V> fromVertex(Vertex vertex,
732+
boolean withResult) {
733+
if (!withResult && vertex instanceof HugeVertex) {
734+
return fromHugeVertex((HugeVertex) vertex);
735+
}
736+
701737
String callableName = vertex.value(P.CALLABLE);
702738
TaskCallable<V> callable;
703739
try {
@@ -710,11 +746,37 @@ public static <V> HugeTask<V> fromVertex(Vertex vertex) {
710746
for (Iterator<VertexProperty<Object>> iter = vertex.properties();
711747
iter.hasNext(); ) {
712748
VertexProperty<Object> prop = iter.next();
749+
if (!withResult && P.RESULT.equals(prop.key())) {
750+
continue;
751+
}
713752
task.property(prop.key(), prop.value());
714753
}
715754
return task;
716755
}
717756

757+
private static <V> HugeTask<V> fromHugeVertex(HugeVertex vertex) {
758+
String callableName = getPropertyValue(vertex, P.CALLABLE);
759+
TaskCallable<V> callable;
760+
try {
761+
callable = TaskCallable.fromClass(callableName);
762+
} catch (Exception e) {
763+
callable = TaskCallable.empty(e);
764+
}
765+
766+
HugeTask<V> task = new HugeTask<>(vertex.id(), null, callable);
767+
for (String property : P.METADATA_PROPERTIES) {
768+
Object value = getPropertyValue(vertex, property);
769+
if (value != null) {
770+
task.property(property, value);
771+
}
772+
}
773+
return task;
774+
}
775+
776+
private static <V> V getPropertyValue(HugeVertex vertex, String property) {
777+
return vertex.getPropertyValue(vertex.graph().propertyKey(property).id());
778+
}
779+
718780
private static <V> Collector<V, ?, Set<V>> toOrderSet() {
719781
return Collectors.toCollection(InsertionOrderUtil::newSet);
720782
}
@@ -792,6 +854,11 @@ public static final class P {
792854
public static final String DEPENDENCIES = "~task_dependencies";
793855
public static final String SERVER = "~task_server";
794856

857+
private static final String[] METADATA_PROPERTIES = new String[]{
858+
TYPE, NAME, CALLABLE, DESCRIPTION, CONTEXT, STATUS, PROGRESS,
859+
CREATE, UPDATE, RETRIES, DEPENDENCIES, INPUT, SERVER
860+
};
861+
795862
//public static final String PARENT = hide("parent");
796863
//public static final String CHILDREN = hide("children");
797864

0 commit comments

Comments
 (0)