fix(server): fix some issues of the distributed scheduler

Author: Tsukilc

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

@@ -50,6 +50,7 @@
 import org.slf4j.Logger;
 
 public class DistributedTaskScheduler extends TaskAndResultScheduler {
+
     private static final Logger LOG = Log.logger(DistributedTaskScheduler.class);
     private final long schedulePeriod;
     private final ExecutorService taskDbExecutor;

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

@@ -28,7 +28,6 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeoutException;
 
-import com.google.common.collect.ImmutableMap;
 import org.apache.hugegraph.HugeException;
 import org.apache.hugegraph.HugeGraph;
 import org.apache.hugegraph.HugeGraphParams;
@@ -57,6 +56,8 @@
 import org.apache.tinkerpop.gremlin.structure.Vertex;
 import org.slf4j.Logger;
 
+import com.google.common.collect.ImmutableMap;
+
 public class StandardTaskScheduler implements TaskScheduler {
 
     private static final Logger LOG = Log.logger(StandardTaskScheduler.class);
@@ -266,7 +267,7 @@ public synchronized <V> void cancel(HugeTask<V> task) {
         }
 
         throw new HugeException("Can't cancel task '%s' in status %s",
-                task.id(), task.status());
+                                task.id(), task.status());
     }
 
     @Override

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

@@ -46,6 +46,7 @@
  * Base class of task & result scheduler
  */
 public abstract class TaskAndResultScheduler implements TaskScheduler {
+
     /**
      * Which graph the scheduler belongs to
      */
@@ -61,8 +62,8 @@ public abstract class TaskAndResultScheduler implements TaskScheduler {
     private final ServerInfoManager serverManager;
 
     public TaskAndResultScheduler(
-        HugeGraphParams graph,
-        ExecutorService serverInfoDbExecutor) {
+            HugeGraphParams graph,
+            ExecutorService serverInfoDbExecutor) {
         E.checkNotNull(graph, "graph");
 
         this.graph = graph;
@@ -90,7 +91,7 @@ public <V> void save(HugeTask<V> task) {
         // Save result outcome
         if (rawResult != null) {
             HugeTaskResult result =
-                new HugeTaskResult(HugeTaskResult.genId(task.id()));
+                    new HugeTaskResult(HugeTaskResult.genId(task.id()));
             result.result(rawResult);
 
             this.call(() -> {
@@ -164,7 +165,7 @@ protected <V> Iterator<HugeTask<V>> queryTask(Map<String, Object> conditions,
             }
             Iterator<Vertex> vertices = this.tx().queryTaskInfos(query);
             Iterator<HugeTask<V>> tasks =
-                new MapperIterator<>(vertices, HugeTask::fromVertex);
+                    new MapperIterator<>(vertices, HugeTask::fromVertex);
             // Convert iterator to list to avoid across thread tx accessed
             return QueryResults.toList(tasks);
         });
@@ -180,16 +181,16 @@ protected <V> Iterator<HugeTask<V>> queryTask(Map<String, Object> conditions,
 
     protected <V> Iterator<HugeTask<V>> queryTask(List<Id> ids) {
         ListIterator<HugeTask<V>> ts = this.call(
-            () -> {
-                Object[] idArray = ids.toArray(new Id[ids.size()]);
-                Iterator<Vertex> vertices = this.tx()
-                                                .queryTaskInfos(idArray);
-                Iterator<HugeTask<V>> tasks =
-                    new MapperIterator<>(vertices,
-                                         HugeTask::fromVertex);
-                // Convert iterator to list to avoid across thread tx accessed
-                return QueryResults.toList(tasks);
-            });
+                () -> {
+                    Object[] idArray = ids.toArray(new Id[ids.size()]);
+                    Iterator<Vertex> vertices = this.tx()
+                                                    .queryTaskInfos(idArray);
+                    Iterator<HugeTask<V>> tasks =
+                            new MapperIterator<>(vertices,
+                                                 HugeTask::fromVertex);
+                    // Convert iterator to list to avoid across thread tx accessed
+                    return QueryResults.toList(tasks);
+                });
 
         Iterator<HugeTaskResult> results = queryTaskResult(ids);
 
@@ -201,7 +202,7 @@ protected <V> Iterator<HugeTask<V>> queryTask(List<Id> ids) {
 
         return new MapperIterator<>(ts, (task) -> {
             HugeTaskResult taskResult =
-                resultCaches.get(HugeTaskResult.genId(task.id()));
+                    resultCaches.get(HugeTaskResult.genId(task.id()));
             if (taskResult != null) {
                 task.result(taskResult);
             }
@@ -231,7 +232,7 @@ protected <V> Iterator<HugeTask<V>> tasksWithoutResult(List<Id> ids) {
             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, HugeTask::fromVertex);
             // Convert iterator to list to avoid across thread tx accessed
             return QueryResults.toList(tasks);
         });
@@ -254,7 +255,7 @@ protected <V> Iterator<HugeTask<V>> queryTaskWithoutResult(String key,
     }
 
     protected <V> Iterator<HugeTask<V>> queryTaskWithoutResult(Map<String,
-        Object> conditions, long limit, String page) {
+            Object> conditions, long limit, String page) {
         return this.call(() -> {
             ConditionQuery query = new ConditionQuery(HugeType.TASK);
             if (page != null) {
@@ -272,7 +273,7 @@ protected <V> Iterator<HugeTask<V>> queryTaskWithoutResult(Map<String,
             }
             Iterator<Vertex> vertices = this.tx().queryTaskInfos(query);
             Iterator<HugeTask<V>> tasks =
-                new MapperIterator<>(vertices, HugeTask::fromVertex);
+                    new MapperIterator<>(vertices, HugeTask::fromVertex);
             // Convert iterator to list to avoid across thread tx accessed
             return QueryResults.toList(tasks);
         });
@@ -281,7 +282,7 @@ protected <V> Iterator<HugeTask<V>> queryTaskWithoutResult(Map<String,
     protected HugeTaskResult queryTaskResult(Id taskid) {
         HugeTaskResult result = this.call(() -> {
             Iterator<Vertex> vertices =
-                this.tx().queryTaskInfos(HugeTaskResult.genId(taskid));
+                    this.tx().queryTaskInfos(HugeTaskResult.genId(taskid));
             Vertex vertex = QueryResults.one(vertices);
             if (vertex == null) {
                 return null;
@@ -296,12 +297,12 @@ protected HugeTaskResult queryTaskResult(Id taskid) {
     protected Iterator<HugeTaskResult> queryTaskResult(List<Id> taskIds) {
         return this.call(() -> {
             Object[] idArray =
-                taskIds.stream().map(HugeTaskResult::genId).toArray();
+                    taskIds.stream().map(HugeTaskResult::genId).toArray();
             Iterator<Vertex> vertices = this.tx()
                                             .queryTaskInfos(idArray);
             Iterator<HugeTaskResult> tasks =
-                new MapperIterator<>(vertices,
-                                     HugeTaskResult::fromVertex);
+                    new MapperIterator<>(vertices,
+                                         HugeTaskResult::fromVertex);
             // Convert iterator to list to avoid across thread tx accessed
             return QueryResults.toList(tasks);
         });