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

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

@@ -58,7 +58,6 @@
 import org.apache.hugegraph.backend.BackendException;
 import org.apache.hugegraph.backend.cache.Cache;
 import org.apache.hugegraph.backend.cache.CacheManager;
-import org.apache.hugegraph.backend.id.IdGenerator;
 import org.apache.hugegraph.backend.store.AbstractBackendStoreProvider;
 import org.apache.hugegraph.backend.store.BackendStoreInfo;
 import org.apache.hugegraph.config.ConfigOption;
@@ -68,6 +67,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,8 +195,6 @@ public final class GraphManager {
     public GraphManager(HugeConfig conf, EventHub hub) {
         LOG.info("Init graph manager");
         E.checkArgumentNotNull(conf, "The config can't be null");
-        String server = conf.get(ServerOptions.SERVER_ID);
-        String role = conf.get(ServerOptions.SERVER_ROLE);
 
         this.config = conf;
         this.url = conf.get(ServerOptions.REST_SERVER_URL);
@@ -206,10 +204,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<>();
@@ -1557,6 +1551,9 @@ private void loadGraph(String name, String graphConfPath) {
         String raftGroupPeers = this.conf.get(ServerOptions.RAFT_GROUP_PEERS);
         config.addProperty(ServerOptions.RAFT_GROUP_PEERS.name(),
                            raftGroupPeers);
+
+        this.transferPdPeersConfig(config);
+
         this.transferRoleWorkerConfig(config);
 
         Graph graph = GraphFactory.open(config);
@@ -1575,6 +1572,19 @@ private void loadGraph(String name, String graphConfPath) {
         }
     }
 
+    private void transferPdPeersConfig(HugeConfig config) {
+        if (config.containsKey(CoreOptions.PD_PEERS.name())) {
+            return;
+        }
+
+        String backend = config.get(CoreOptions.BACKEND);
+        boolean needPdPeers = this.conf.get(ServerOptions.USE_PD) ||
+                              "hstore".equals(backend);
+        if (needPdPeers) {
+            config.addProperty(CoreOptions.PD_PEERS.name(), this.pdPeers);
+        }
+    }
+
     private void transferRoleWorkerConfig(HugeConfig config) {
         config.setProperty(RoleElectionOptions.NODE_EXTERNAL_URL.name(),
                            this.conf.get(ServerOptions.REST_SERVER_URL));
@@ -1635,23 +1645,18 @@ private void checkBackendVersionOrExit(HugeConfig config) {
     }
 
     private void initNodeRole() {
-        String id = config.get(ServerOptions.SERVER_ID);
+        boolean enableRoleElection = config.get(
+                ServerOptions.ENABLE_SERVER_ROLE_ELECTION);
+        if (enableRoleElection) {
+            LOG.warn("The server.role_election option is deprecated and no " +
+                     "longer supported (removed with server_info persistence). " +
+                     "Nodes operate as WORKER with local scheduling. Set " +
+                     "enable_server_role_election=false to suppress this warning.");
+        }
+
         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() &&
-                                      this.supportRoleElection() &&
-                                      config.get(ServerOptions.ENABLE_SERVER_ROLE_ELECTION);
-        if (supportRoleElection) {
-            // Init any server as Worker role, then do role election
-            nodeRole = NodeRole.WORKER;
-        }
-
-        this.globalNodeRoleInfo.initNodeId(IdGenerator.of(id));
         this.globalNodeRoleInfo.initNodeRole(nodeRole);
     }
 
@@ -1663,7 +1668,7 @@ private void serverStarted(HugeConfig conf) {
         }
         if (!this.globalNodeRoleInfo.nodeRole().computer() && this.supportRoleElection() &&
             config.get(ServerOptions.ENABLE_SERVER_ROLE_ELECTION)) {
-            this.initRoleStateMachine();
+            LOG.info("Skip role state machine init (deprecated with server_info)");
         }
     }
 
@@ -1937,26 +1942,29 @@ public Set<String> getServiceUrls(String graphSpace, String service,
     public HugeGraph graph(String graphSpace, String name) {
         String key = String.join(DELIMITER, graphSpace, name);
         Graph graph = this.graphs.get(key);
-        if (graph == null && isPDEnabled()) {
-            Map<String, Map<String, Object>> configs =
-                    this.metaManager.graphConfigs(graphSpace);
-            // If current server registered graph space is not DEFAULT, only load graph creation
-            // under registered graph space
-            if (!configs.containsKey(key) ||
-                (!"DEFAULT".equals(this.serviceGraphSpace) &&
-                 !graphSpace.equals(this.serviceGraphSpace))) {
-                return null;
+        if (graph == null) {
+            if (isPDEnabled()) {
+                Map<String, Map<String, Object>> configs =
+                        this.metaManager.graphConfigs(graphSpace);
+                // If current server registered graph space is not DEFAULT, only load graph creation
+                // under registered graph space
+                if (!configs.containsKey(key) ||
+                    (!"DEFAULT".equals(this.serviceGraphSpace) &&
+                     !graphSpace.equals(this.serviceGraphSpace))) {
+                    return null;
+                }
+                Map<String, Object> config = configs.get(key);
+                String creator = String.valueOf(config.get("creator"));
+                Date createTime = parseDate(config.get("create_time"));
+                Date updateTime = parseDate(config.get("update_time"));
+                HugeGraph graph1 = this.createGraph(graphSpace, name,
+                                                    creator, config, false);
+                graph1.createTime(createTime);
+                graph1.updateTime(updateTime);
+                this.graphs.put(key, graph1);
+                return graph1;
             }
-            Map<String, Object> config = configs.get(key);
-            String creator = String.valueOf(config.get("creator"));
-            Date createTime = parseDate(config.get("create_time"));
-            Date updateTime = parseDate(config.get("update_time"));
-            HugeGraph graph1 = this.createGraph(graphSpace, name,
-                                                creator, config, false);
-            graph1.createTime(createTime);
-            graph1.updateTime(updateTime);
-            this.graphs.put(key, graph1);
-            return graph1;
+            throw new NotFoundException(String.format("Graph '%s' does not exist", name));
         } else if (graph instanceof HugeGraph) {
             return (HugeGraph) graph;
         }

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

@@ -177,7 +177,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;
@@ -226,11 +225,20 @@ public StandardHugeGraph(HugeConfig config) {
 
         this.taskManager = TaskManager.instance();
         this.name = config.get(CoreOptions.STORE);
+
+        // Fail fast if user still has the removed config key in their properties file
+        if (config.containsKey("task.scheduler_type")) {
+            throw new HugeException(
+                "Config key 'task.scheduler_type' has been removed. " +
+                "The scheduler is now auto-selected based on backend type " +
+                "(hstore → distributed, others → local). " +
+                "Please remove this key from your hugegraph.properties.");
+        }
+
         this.started = false;
         this.closed = false;
         this.mode = GraphMode.NONE;
         this.readMode = GraphReadMode.OLTP_ONLY;
-        this.schedulerType = config.get(CoreOptions.SCHEDULER_TYPE);
 
         // Init process-wide static configs before lock, so that validation
         // failures won't leave stale lock groups in LockManager.
@@ -323,6 +331,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()
@@ -340,11 +349,10 @@ public void serverStarted(GlobalMasterInfo nodeInfo) {
         LOG.info("Init system info for graph '{}'", this.spaceGraphName());
         this.initSystemInfo();
 
-        LOG.info("Init server info [{}-{}] for graph '{}'...",
-                 nodeInfo.nodeId(), nodeInfo.nodeRole(), this.spaceGraphName());
-        this.serverInfoManager().initServerInfo(nodeInfo);
-
-        this.initRoleStateMachine(nodeInfo.nodeId());
+        if (nodeInfo != null && nodeInfo.nodeId() != null) {
+            this.serverInfoManager().initServerInfo(nodeInfo);
+            this.initRoleStateMachine(nodeInfo.nodeId());
+        }
 
         // TODO: check necessary?
         LOG.info("Check olap property-key tables for graph '{}'", this.spaceGraphName());
@@ -473,6 +481,7 @@ public void updateTime(Date updateTime) {
         this.updateTime = updateTime;
     }
 
+    @Override
     public void waitStarted() {
         // Just for trigger Tx.getOrNewTransaction, then load 3 stores
         this.schemaTransaction();
@@ -489,9 +498,7 @@ public void initBackend() {
         try {
             this.storeProvider.init();
             /*
-             * NOTE: The main goal is to write the serverInfo to the central
-             * node, such as etcd, and also create the system schema in memory,
-             * which has no side effects
+             * NOTE: Create system schema in memory, which has no side effects.
              */
             this.initSystemInfo();
         } finally {
@@ -532,8 +539,7 @@ public void truncateBackend() {
         LockUtil.lock(this.spaceGraphName(), LockUtil.GRAPH_LOCK);
         try {
             this.storeProvider.truncate();
-            // TODO: remove this after serverinfo saved in etcd
-            this.serverStarted(this.serverInfoManager().globalNodeRoleInfo());
+            this.serverStarted(null);
         } finally {
             LockUtil.unlock(this.spaceGraphName(), LockUtil.GRAPH_LOCK);
         }
@@ -555,7 +561,6 @@ public KvStore kvStore() {
     public void initSystemInfo() {
         try {
             this.taskScheduler().init();
-            this.serverInfoManager().init();
             this.authManager().init();
         } finally {
             this.closeTx();
@@ -1640,7 +1645,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

@@ -313,13 +313,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

@@ -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;
     }