fix(server): fix the scheduler and the scheduler selection logic

docker/configs/server1-conf/graphs/hugegraph.properties

@@ -13,7 +13,6 @@ serializer=binary
 pd.peers=127.0.0.1:8686,127.0.0.1:8687,127.0.0.1:8688
 
 # task config
-task.scheduler_type=local
 task.schedule_period=10
 task.retry=0
 task.wait_timeout=10

docker/configs/server2-conf/graphs/hugegraph.properties

@@ -13,7 +13,6 @@ serializer=binary
 pd.peers=127.0.0.1:8686,127.0.0.1:8687,127.0.0.1:8688
 
 # task config
-task.scheduler_type=local
 task.schedule_period=10
 task.retry=0
 task.wait_timeout=10

docker/configs/server3-conf/graphs/hugegraph.properties

@@ -13,7 +13,6 @@ serializer=binary
 pd.peers=127.0.0.1:8686,127.0.0.1:8687,127.0.0.1:8688
 
 # task config
-task.scheduler_type=local
 task.schedule_period=10
 task.retry=0
 task.wait_timeout=10

hugegraph-cluster-test/hugegraph-clustertest-dist/src/assembly/static/conf/hugegraph.properties.template

@@ -45,7 +45,6 @@ store=hugegraph
 pd.peers=$PD_PEERS_LIST$
 
 # task config
-task.scheduler_type=local
 task.schedule_period=10
 task.retry=0
 task.wait_timeout=10

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java

@@ -556,9 +556,9 @@ public class ServerOptions extends OptionHolder {
     public static final ConfigOption<String> SERVER_ID =
             new ConfigOption<>(
                     "server.id",
-                    "The id of hugegraph-server.",
-                    disallowEmpty(),
-                    "server-1"
+                    "The id of hugegraph-server, auto-generated if not specified.",
+                    null,
+                    ""
             );
     public static final ConfigOption<String> SERVER_ROLE =
             new ConfigOption<>(

hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java

@@ -33,6 +33,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -68,6 +69,7 @@
 import org.apache.hugegraph.config.TypedOption;
 import org.apache.hugegraph.event.EventHub;
 import org.apache.hugegraph.exception.ExistedException;
+import org.apache.hugegraph.exception.NotFoundException;
 import org.apache.hugegraph.exception.NotSupportException;
 import org.apache.hugegraph.io.HugeGraphSONModule;
 import org.apache.hugegraph.k8s.K8sDriver;
@@ -195,7 +197,17 @@ public final class GraphManager {
     public GraphManager(HugeConfig conf, EventHub hub) {
         LOG.info("Init graph manager");
         E.checkArgumentNotNull(conf, "The config can't be null");
+
+        // Auto-generate server.id if not configured.
+        // Random generation is to prevent duplicate id error reports.This id is currently
+        // meaningless and needs to be completely removed serverInfoManager in
+        // the future
         String server = conf.get(ServerOptions.SERVER_ID);
+        if (StringUtils.isEmpty(server)) {
+            server = "server-" + UUID.randomUUID().toString().substring(0, 8);
+            LOG.info("Auto-generated server.id: {}", server);
+            conf.setProperty(ServerOptions.SERVER_ID.name(), server);
+        }
         String role = conf.get(ServerOptions.SERVER_ROLE);
 
         this.config = conf;
@@ -206,10 +218,6 @@ public GraphManager(HugeConfig conf, EventHub hub) {
                 conf.get(ServerOptions.SERVER_DEPLOY_IN_K8S);
         this.startIgnoreSingleGraphError = conf.get(
                 ServerOptions.SERVER_START_IGNORE_SINGLE_GRAPH_ERROR);
-        E.checkArgument(server != null && !server.isEmpty(),
-                        "The server name can't be null or empty");
-        E.checkArgument(role != null && !role.isEmpty(),
-                        "The server role can't be null or empty");
         this.graphsDir = conf.get(ServerOptions.GRAPHS);
         this.cluster = conf.get(ServerOptions.CLUSTER);
         this.graphSpaces = new ConcurrentHashMap<>();
@@ -276,7 +284,7 @@ private static String serviceId(String graphSpace, Service.ServiceType type,
                      .replace("_", "-").toLowerCase();
     }
 
-    private boolean usePD() {
+    public boolean usePD() {
         return this.PDExist;
     }
 
@@ -1557,6 +1565,14 @@ private void loadGraph(String name, String graphConfPath) {
         String raftGroupPeers = this.conf.get(ServerOptions.RAFT_GROUP_PEERS);
         config.addProperty(ServerOptions.RAFT_GROUP_PEERS.name(),
                            raftGroupPeers);
+
+        // Transfer `pd.peers` from server config to graph config
+        // Only inject if not already configured in graph config
+        if (!config.containsKey("pd.peers")) {
+            String pdPeers = this.conf.get(ServerOptions.PD_PEERS);
+            config.addProperty("pd.peers", pdPeers);
+        }
+
         this.transferRoleWorkerConfig(config);
 
         Graph graph = GraphFactory.open(config);
@@ -1637,10 +1653,6 @@ private void checkBackendVersionOrExit(HugeConfig config) {
     private void initNodeRole() {
         String id = config.get(ServerOptions.SERVER_ID);
         String role = config.get(ServerOptions.SERVER_ROLE);
-        E.checkArgument(StringUtils.isNotEmpty(id),
-                        "The server name can't be null or empty");
-        E.checkArgument(StringUtils.isNotEmpty(role),
-                        "The server role can't be null or empty");
 
         NodeRole nodeRole = NodeRole.valueOf(role.toUpperCase());
         boolean supportRoleElection = !nodeRole.computer() &&
@@ -1960,7 +1972,7 @@ public HugeGraph graph(String graphSpace, String name) {
         } else if (graph instanceof HugeGraph) {
             return (HugeGraph) graph;
         }
-        throw new NotSupportException("graph instance of %s", graph.getClass());
+        throw new NotFoundException(String.format("Graph '%s' does not exist", name));
     }
 
     public void dropGraphLocal(String name) {

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java

@@ -176,7 +176,6 @@ public class StandardHugeGraph implements HugeGraph {
     private final BackendStoreProvider storeProvider;
     private final TinkerPopTransaction tx;
     private final RamTable ramtable;
-    private final String schedulerType;
     private volatile boolean started;
     private volatile boolean closed;
     private volatile GraphMode mode;
@@ -229,7 +228,6 @@ public StandardHugeGraph(HugeConfig config) {
         this.closed = false;
         this.mode = GraphMode.NONE;
         this.readMode = GraphReadMode.OLTP_ONLY;
-        this.schedulerType = config.get(CoreOptions.SCHEDULER_TYPE);
 
         LockUtil.init(this.spaceGraphName());
 
@@ -315,6 +313,7 @@ public String backend() {
         return this.storeProvider.type();
     }
 
+    @Override
     public BackendStoreInfo backendStoreInfo() {
         // Just for trigger Tx.getOrNewTransaction, then load 3 stores
         // TODO: pass storeProvider.metaStore()
@@ -465,6 +464,7 @@ public void updateTime(Date updateTime) {
         this.updateTime = updateTime;
     }
 
+    @Override
     public void waitStarted() {
         // Just for trigger Tx.getOrNewTransaction, then load 3 stores
         this.schemaTransaction();
@@ -1629,7 +1629,9 @@ public <T> void submitEphemeralJob(EphemeralJob<T> job) {
 
         @Override
         public String schedulerType() {
-            return StandardHugeGraph.this.schedulerType;
+            // Use distributed scheduler for hstore backend, otherwise use local
+            // After the merger of rocksdb and hstore, consider whether to change this logic
+            return StandardHugeGraph.this.isHstore() ? "distributed" : "local";
         }
     }
 

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java

@@ -303,13 +303,7 @@ public class CoreOptions extends OptionHolder {
                     rangeInt(1, 500),
                     1
             );
-    public static final ConfigOption<String> SCHEDULER_TYPE =
-            new ConfigOption<>(
-                    "task.scheduler_type",
-                    "The type of scheduler used in distribution system.",
-                    allowValues("local", "distributed"),
-                    "local"
-            );
+
     public static final ConfigOption<Boolean> TASK_SYNC_DELETION =
             new ConfigOption<>(
                     "task.sync_deletion",

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/GlobalMasterInfo.java

@@ -22,7 +22,7 @@
 import org.apache.hugegraph.type.define.NodeRole;
 import org.apache.hugegraph.util.E;
 
-// TODO: rename to GlobalNodeRoleInfo
+// TODO: We need to completely delete the startup of master-worker
 public final class GlobalMasterInfo {
 
     private static final NodeInfo NO_MASTER = new NodeInfo(false, "");

hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleListener.java

@@ -17,12 +17,12 @@
 
 package org.apache.hugegraph.masterelection;
 
-import java.util.Objects;
-
 import org.apache.hugegraph.task.TaskManager;
 import org.apache.hugegraph.util.Log;
 import org.slf4j.Logger;
 
+import java.util.Objects;
+
 public class StandardRoleListener implements RoleListener {
 
     private static final Logger LOG = Log.logger(StandardRoleListener.class);
@@ -36,7 +36,6 @@ public class StandardRoleListener implements RoleListener {
     public StandardRoleListener(TaskManager taskManager,
                                 GlobalMasterInfo roleInfo) {
         this.taskManager = taskManager;
-        this.taskManager.enableRoleElection();
         this.roleInfo = roleInfo;
         this.selfIsMaster = false;
     }

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

@@ -19,7 +19,9 @@
 
 import java.util.Iterator;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -48,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;
@@ -118,6 +121,11 @@ private static boolean sleep(long ms) {
     public void cronSchedule() {
         // Perform periodic scheduling tasks
 
+        // Check closed flag first to exit early
+        if (this.closed.get()) {
+            return;
+        }
+
         if (!this.graph.started() || this.graph.closed()) {
             return;
         }
@@ -253,6 +261,10 @@ public <V> Future<?> schedule(HugeTask<V> task) {
             return this.ephemeralTaskExecutor.submit(task);
         }
 
+        // Validate task state before saving to ensure correct exception type
+        E.checkState(task.type() != null, "Task type can't be null");
+        E.checkState(task.name() != null, "Task name can't be null");
+
         // Process schema task
         // Handle gremlin task
         // Handle OLAP calculation tasks
@@ -284,14 +296,41 @@ protected <V> void initTaskParams(HugeTask<V> task) {
         }
     }
 
+    /**
+     * Note: This method will update the status of the input task.
+     *
+     * @param task
+     * @param <V>
+     */
     @Override
     public <V> void cancel(HugeTask<V> task) {
-        // Update status to CANCELLING
-        if (!task.completed()) {
-            // Task not completed, can only execute status not CANCELLING
-            this.updateStatus(task.id(), null, TaskStatus.CANCELLING);
+        E.checkArgumentNotNull(task, "Task can't be null");
+
+        if (task.completed() || task.cancelling()) {
+            return;
+        }
+
+        LOG.info("Cancel task '{}' in status {}", task.id(), task.status());
+
+        // 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);
+            }
+            LOG.info("Cancel local running task '{}' result: {}", task.id(), cancelled);
+            return;
+        }
+
+        // Task not running locally, update status to CANCELLING
+        // for cronSchedule() or other nodes to handle
+        TaskStatus currentStatus = task.status();
+        if (!this.updateStatus(task.id(), currentStatus, TaskStatus.CANCELLING)) {
+            LOG.info("Failed to cancel task '{}', status may have changed from {}",
+                     task.id(), currentStatus);
         } else {
-            LOG.info("cancel task({}) error, task has completed", task.id());
+            task.overwriteStatus(TaskStatus.CANCELLING);
         }
     }
 
@@ -316,14 +355,18 @@ protected <V> HugeTask<V> deleteFromDB(Id id) {
 
     @Override
     public <V> HugeTask<V> delete(Id id, boolean force) {
-        if (!force) {
-            // Change status to DELETING, perform the deletion operation through automatic
-            // scheduling.
+        HugeTask<?> task = this.taskWithoutResult(id);
+
+        if (!force && !task.completed()) {
+            // Check task status: can't delete running tasks without force
             this.updateStatus(id, null, TaskStatus.DELETING);
             return null;
-        } else {
-            return this.deleteFromDB(id);
+            // Already in DELETING status, delete directly from DB
+            // Completed tasks can also be deleted directly
         }
+
+        // Delete from DB directly for completed/DELETING tasks or force=true
+        return this.deleteFromDB(id);
     }
 
     @Override
@@ -353,6 +396,18 @@ public boolean close() {
             cronFuture.cancel(false);
         }
 
+        // Wait for cron task to complete to ensure all transactions are closed
+        try {
+            cronFuture.get(schedulePeriod + 5, TimeUnit.SECONDS);
+        } catch (CancellationException e) {
+            // Task was cancelled, this is expected
+            LOG.debug("Cron task was cancelled");
+        } catch (TimeoutException e) {
+            LOG.warn("Cron task did not complete in time when closing scheduler");
+        } catch (ExecutionException | InterruptedException e) {
+            LOG.warn("Exception while waiting for cron task to complete", e);
+        }
+
         if (!this.taskDbExecutor.isShutdown()) {
             this.call(() -> {
                 try {
@@ -363,7 +418,10 @@ public boolean close() {
                 this.graph.closeTx();
             });
         }
-        return true;
+
+        //todo: serverInfoManager section should be removed in the future.
+        return this.serverManager().close();
+        //return true;
     }
 
     @Override

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

@@ -209,14 +209,6 @@ public static HugeServerInfo fromVertex(Vertex vertex) {
         return serverInfo;
     }
 
-    public <V> boolean suitableFor(HugeTask<V> task, long now) {
-        if (task.computer() != this.role.computer()) {
-            return false;
-        }
-        return this.updateTime.getTime() + EXPIRED_INTERVAL >= now &&
-               this.load() + task.load() <= this.maxLoad;
-    }
-
     public static Schema schema(HugeGraphParams graph) {
         return new Schema(graph);
     }