Add MasterCoordinator

Author: ruanwenjun

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/TaskGroupQueueStatus.java

@@ -28,6 +28,7 @@ public enum TaskGroupQueueStatus {
 
     WAIT_QUEUE(-1, "wait queue"),
     ACQUIRE_SUCCESS(1, "acquire success"),
+    @Deprecated
     RELEASE(2, "release");
 
     @EnumValue

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/model/MasterHeartBeat.java

@@ -24,4 +24,6 @@
 @NoArgsConstructor
 public class MasterHeartBeat extends BaseHeartBeat implements HeartBeat {
 
+    private boolean isCoordinator;
+
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/MasterServer.java

@@ -33,6 +33,7 @@
 import org.apache.dolphinscheduler.scheduler.api.SchedulerApi;
 import org.apache.dolphinscheduler.server.master.cluster.ClusterManager;
 import org.apache.dolphinscheduler.server.master.cluster.ClusterStateMonitors;
+import org.apache.dolphinscheduler.server.master.engine.MasterCoordinator;
 import org.apache.dolphinscheduler.server.master.engine.WorkflowEngine;
 import org.apache.dolphinscheduler.server.master.engine.system.SystemEventBus;
 import org.apache.dolphinscheduler.server.master.engine.system.SystemEventBusFireWorker;
@@ -95,6 +96,9 @@ public class MasterServer implements IStoppable {
     @Autowired
     private SystemEventBusFireWorker systemEventBusFireWorker;
 
+    @Autowired
+    private MasterCoordinator masterCoordinator;
+
     public static void main(String[] args) {
         MasterServerMetrics.registerUncachedException(DefaultUncaughtExceptionHandler::getUncaughtExceptionCount);
 
@@ -122,6 +126,8 @@ public void initialized() {
         this.masterRegistryClient.start();
         this.masterRegistryClient.setRegistryStoppable(this);
 
+        this.masterCoordinator.start();
+
         this.clusterManager.start();
         this.clusterStateMonitors.start();
 
@@ -173,6 +179,7 @@ public void close(String cause) {
                 WorkflowEngine workflowEngine1 = workflowEngine;
                 SchedulerApi closedSchedulerApi = schedulerApi;
                 MasterRpcServer closedRpcServer = masterRPCServer;
+                MasterCoordinator closeMasterCoordinator = masterCoordinator;
                 MasterRegistryClient closedMasterRegistryClient = masterRegistryClient;
                 // close spring Context and will invoke method with @PreDestroy annotation to destroy beans.
                 // like ServerNodeManager,HostManager,TaskResponseService,CuratorZookeeperClient,etc

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/MasterCoordinator.java

@@ -0,0 +1,81 @@
+/*
+ * 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.dolphinscheduler.server.master.engine;
+
+import org.apache.dolphinscheduler.registry.api.Registry;
+import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
+import org.apache.dolphinscheduler.registry.api.ha.AbstractHAServer;
+import org.apache.dolphinscheduler.registry.api.ha.AbstractServerStatusChangeListener;
+import org.apache.dolphinscheduler.server.master.config.MasterConfig;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+/**
+ * The MasterCoordinator is singleton at the clusters, which is used to do some control work, e.g manage the {@link TaskGroupCoordinator}
+ */
+@Slf4j
+@Component
+public class MasterCoordinator extends AbstractHAServer {
+
+    @Autowired
+    private TaskGroupCoordinator taskGroupCoordinator;
+
+    public MasterCoordinator(final Registry registry, final MasterConfig masterConfig) {
+        super(
+                registry,
+                RegistryNodeType.MASTER_COORDINATOR.getRegistryPath(),
+                masterConfig.getMasterAddress());
+
+        addServerStatusChangeListener(new AbstractServerStatusChangeListener() {
+
+            @Override
+            public void changeToActive() {
+                onActive();
+            }
+
+            @Override
+            public void changeToStandBy() {
+                onStandBy();
+            }
+        });
+    }
+
+    @Override
+    public void start() {
+        super.start();
+        log.info("MasterCoordinator started...");
+    }
+
+    @Override
+    public void close() {
+        taskGroupCoordinator.close();
+        log.info("MasterCoordinator shutdown...");
+    }
+
+    private void onActive() {
+        taskGroupCoordinator.start();
+    }
+
+    private void onStandBy() {
+        taskGroupCoordinator.close();
+    }
+
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/TaskGroupCoordinator.java

@@ -37,8 +37,6 @@
 import org.apache.dolphinscheduler.extract.master.transportor.TaskGroupSlotAcquireSuccessNotifyResponse;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
-import org.apache.dolphinscheduler.registry.api.RegistryClient;
-import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
 import org.apache.dolphinscheduler.server.master.utils.TaskGroupUtils;
 
 import org.apache.commons.collections4.CollectionUtils;
@@ -47,6 +45,7 @@
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -79,10 +78,7 @@
  */
 @Slf4j
 @Component
-public class TaskGroupCoordinator extends BaseDaemonThread implements AutoCloseable {
-
-    @Autowired
-    private RegistryClient registryClient;
+public class TaskGroupCoordinator implements AutoCloseable {
 
     @Autowired
     private TaskGroupDao taskGroupDao;
@@ -98,38 +94,42 @@ public class TaskGroupCoordinator extends BaseDaemonThread implements AutoClosea
 
     private boolean flag = true;
 
-    private static final int DEFAULT_LIMIT = 1000;
+    private Thread internalThread;
 
-    public TaskGroupCoordinator() {
-        super("TaskGroupCoordinator");
-    }
+    private static final int DEFAULT_LIMIT = 1000;
 
-    @Override
     public synchronized void start() {
         log.info("TaskGroupCoordinator starting...");
+        if (flag) {
+            throw new IllegalStateException("TaskGroupCoordinator is already started");
+        }
+        if (internalThread != null) {
+            throw new IllegalStateException("InternalThread is already started");
+        }
         flag = true;
-        super.start();
+        internalThread = new BaseDaemonThread(this::doStart) {
+        };
+        internalThread.start();
         log.info("TaskGroupCoordinator started...");
     }
 
-    @Override
-    public void run() {
+    private void doStart() {
+        // Sleep 1 minutes here to make sure the previous task group slot has been released.
+        // This step is not necessary, since the wakeup operation is idempotent, but we can avoid confusion warning.
+        ThreadUtils.sleep(TimeUnit.MINUTES.toMillis(1));
+
         while (flag) {
             try {
-                registryClient.getLock(RegistryNodeType.MASTER_TASK_GROUP_COORDINATOR_LOCK.getRegistryPath());
-                try {
-                    StopWatch taskGroupCoordinatorRoundCost = StopWatch.createStarted();
+                final StopWatch taskGroupCoordinatorRoundCost = StopWatch.createStarted();
 
-                    amendTaskGroupUseSize();
-                    amendTaskGroupQueueStatus();
-                    dealWithForceStartTaskGroupQueue();
-                    dealWithWaitingTaskGroupQueue();
+                //
+                amendTaskGroupUseSize();
+                amendTaskGroupQueueStatus();
+                dealWithForceStartTaskGroupQueue();
+                dealWithWaitingTaskGroupQueue();
 
-                    taskGroupCoordinatorRoundCost.stop();
-                    log.debug("TaskGroupCoordinator round cost: {}/ms", taskGroupCoordinatorRoundCost.getTime());
-                } finally {
-                    registryClient.releaseLock(RegistryNodeType.MASTER_TASK_GROUP_COORDINATOR_LOCK.getRegistryPath());
-                }
+                taskGroupCoordinatorRoundCost.stop();
+                log.debug("TaskGroupCoordinator round cost: {}/ms", taskGroupCoordinatorRoundCost.getTime());
             } catch (Throwable e) {
                 log.error("TaskGroupCoordinator error", e);
             } finally {
@@ -212,7 +212,6 @@ private void amendTaskGroupQueueStatus(List<TaskGroupQueue> taskGroupQueues) {
                 log.warn("The TaskInstance: {} state: {} finished, will release the TaskGroupQueue: {}",
                         taskInstance.getName(), taskInstance.getState(), taskGroupQueue);
                 deleteTaskGroupQueueSlot(taskGroupQueue);
-                continue;
             }
         }
     }
@@ -226,7 +225,7 @@ private void dealWithForceStartTaskGroupQueue() {
         int limit = DEFAULT_LIMIT;
         StopWatch taskGroupCoordinatorRoundTimeCost = StopWatch.createStarted();
         while (true) {
-            List<TaskGroupQueue> taskGroupQueues =
+            final List<TaskGroupQueue> taskGroupQueues =
                     taskGroupQueueDao.queryWaitNotifyForceStartTaskGroupQueue(minTaskGroupQueueId, limit);
             if (CollectionUtils.isEmpty(taskGroupQueues)) {
                 break;
@@ -245,7 +244,7 @@ private void dealWithForceStartTaskGroupQueue(List<TaskGroupQueue> taskGroupQueu
         // Find the force start task group queue(Which is inQueue and forceStart is YES)
         // Notify the related waiting task instance
         // Set the taskGroupQueue status to RELEASE and remove it from queue
-        for (TaskGroupQueue taskGroupQueue : taskGroupQueues) {
+        for (final TaskGroupQueue taskGroupQueue : taskGroupQueues) {
             try {
                 LogUtils.setTaskInstanceIdMDC(taskGroupQueue.getTaskId());
                 // notify the waiting task instance
@@ -488,8 +487,21 @@ private void deleteTaskGroupQueueSlot(TaskGroupQueue taskGroupQueue) {
     }
 
     @Override
-    public void close() throws Exception {
+    public synchronized void close() {
+        if (!flag) {
+            log.error("TaskGroupCoordinator is already closed");
+            return;
+        }
         flag = false;
+        try {
+            if (internalThread != null) {
+                internalThread.interrupt();
+                internalThread.join();
+            }
+        } catch (Exception ex) {
+            log.error("Close internalThread failed", ex);
+        }
+        internalThread = null;
         log.info("TaskGroupCoordinator closed");
     }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterHeartBeatTask.java

@@ -30,6 +30,7 @@
 import org.apache.dolphinscheduler.registry.api.utils.RegistryUtils;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.config.MasterServerLoadProtection;
+import org.apache.dolphinscheduler.server.master.engine.MasterCoordinator;
 import org.apache.dolphinscheduler.server.master.metrics.MasterServerMetrics;
 
 import lombok.NonNull;
@@ -44,17 +45,21 @@ public class MasterHeartBeatTask extends BaseHeartBeatTask<MasterHeartBeat> {
 
     private final RegistryClient registryClient;
 
+    private final MasterCoordinator masterCoordinator;
+
     private final String heartBeatPath;
 
     private final int processId;
 
     public MasterHeartBeatTask(@NonNull MasterConfig masterConfig,
                                @NonNull MetricsProvider metricsProvider,
-                               @NonNull RegistryClient registryClient) {
+                               @NonNull RegistryClient registryClient,
+                               @NonNull MasterCoordinator masterCoordinator) {
         super("MasterHeartBeatTask", masterConfig.getMaxHeartbeatInterval().toMillis());
         this.masterConfig = masterConfig;
         this.metricsProvider = metricsProvider;
         this.registryClient = registryClient;
+        this.masterCoordinator = masterCoordinator;
         this.heartBeatPath = masterConfig.getMasterRegistryPath();
         this.processId = OSUtils.getProcessID();
     }
@@ -75,6 +80,7 @@ public MasterHeartBeat getHeartBeat() {
                 .serverStatus(serverStatus)
                 .host(NetUtils.getHost())
                 .port(masterConfig.getListenPort())
+                .isCoordinator(masterCoordinator.isActive())
                 .build();
     }
 

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistryClient.java

@@ -30,6 +30,7 @@
 import org.apache.dolphinscheduler.registry.api.RegistryException;
 import org.apache.dolphinscheduler.registry.api.enums.RegistryNodeType;
 import org.apache.dolphinscheduler.server.master.config.MasterConfig;
+import org.apache.dolphinscheduler.server.master.engine.MasterCoordinator;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -53,11 +54,15 @@ public class MasterRegistryClient implements AutoCloseable {
     @Autowired
     private MetricsProvider metricsProvider;
 
+    @Autowired
+    private MasterCoordinator masterCoordinator;
+
     private MasterHeartBeatTask masterHeartBeatTask;
 
     public void start() {
         try {
-            this.masterHeartBeatTask = new MasterHeartBeatTask(masterConfig, metricsProvider, registryClient);
+            this.masterHeartBeatTask =
+                    new MasterHeartBeatTask(masterConfig, metricsProvider, registryClient, masterCoordinator);
             // master registry
             registry();
             registryClient.addConnectionStateListener(new MasterConnectionStateListener(registryClient));

dolphinscheduler-registry/dolphinscheduler-registry-api/src/main/java/org/apache/dolphinscheduler/registry/api/enums/RegistryNodeType.java

@@ -28,6 +28,7 @@ public enum RegistryNodeType {
 
     MASTER("Master", "/nodes/master"),
     MASTER_FAILOVER_LOCK("MasterFailoverLock", "/lock/master-failover"),
+    MASTER_COORDINATOR("MasterCoordinator", "/nodes/master-coordinator"),
     MASTER_TASK_GROUP_COORDINATOR_LOCK("TaskGroupCoordinatorLock", "/lock/master-task-group-coordinator"),
     MASTER_SERIAL_COORDINATOR_LOCK("SerialWorkflowCoordinator", "/lock/master-serial-workflow-coordinator"),