In modern data-driven enterprises, a workflow scheduling system is the "central nervous system" of the data pipeline. From ETL tasks to machine learning training, from report generation to real-time monitoring, almost all critical business processes rely on a stable, efficient, and scalable scheduling engine. The author believes that Apache DolphinScheduler 3.1.9 is a stable and widely used version, so this article focuses on this version, analyzing the relevant processes related to the Master service startup, deeply exploring its core source code, architectural design, module division, and key implementation mechanisms. The goal is to help developers understand how the Master works and lay a foundation for further secondary development or performance optimization. how the Master works This series of articles is divided into three parts: the Master Server startup process, the Worker server startup process, and related process diagrams. This is the first part. 1. Core Overview of Master Server Startup Code Entry: org.apache.dolphinscheduler.server.master.MasterServer#run Code Entry: org.apache.dolphinscheduler.server.master.MasterServer#run Code Entry org.apache.dolphinscheduler.server.master.MasterServer#run public void run() throws SchedulerException {
        // 1. Initialize rpc server
        this.masterRPCServer.start();

        // 2. Install task plugin
        this.taskPluginManager.loadPlugin();

        // 3. Self-tolerant
        this.masterRegistryClient.start();
        this.masterRegistryClient.setRegistryStoppable(this);

        // 4. Master scheduling
        this.masterSchedulerBootstrap.init();
        this.masterSchedulerBootstrap.start();

        // 5. Event execution service
        this.eventExecuteService.start();
        // 6. Fault tolerance mechanism
        this.failoverExecuteThread.start();

        // 7. Quartz scheduling
        this.schedulerApi.start();
        ...
    } public void run() throws SchedulerException {
        // 1. Initialize rpc server
        this.masterRPCServer.start();

        // 2. Install task plugin
        this.taskPluginManager.loadPlugin();

        // 3. Self-tolerant
        this.masterRegistryClient.start();
        this.masterRegistryClient.setRegistryStoppable(this);

        // 4. Master scheduling
        this.masterSchedulerBootstrap.init();
        this.masterSchedulerBootstrap.start();

        // 5. Event execution service
        this.eventExecuteService.start();
        // 6. Fault tolerance mechanism
        this.failoverExecuteThread.start();

        // 7. Quartz scheduling
        this.schedulerApi.start();
        ...
    } 1.1 RPC Startup: Description: Registers processors for relevant commands, such as task execution, task execution results, task termination, etc.
Code Entry: org.apache.dolphinscheduler.server.master.rpc.MasterRPCServer#start Description: Registers processors for relevant commands, such as task execution, task execution results, task termination, etc. Description Code Entry: org.apache.dolphinscheduler.server.master.rpc.MasterRPCServer#start Code Entry org.apache.dolphinscheduler.server.master.rpc.MasterRPCServer#start public void start() {
         ...
        // Task execution request processor
        this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RUNNING, taskExecuteRunningProcessor);
        // Task execution result request processor
        this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RESULT, taskExecuteResponseProcessor);
        // Task termination request processor
        this.nettyRemotingServer.registerProcessor(CommandType.TASK_KILL_RESPONSE, taskKillResponseProcessor);
        ...
        this.nettyRemotingServer.start();
        logger.info("Started Master RPC Server...");
    } public void start() {
         ...
        // Task execution request processor
        this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RUNNING, taskExecuteRunningProcessor);
        // Task execution result request processor
        this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RESULT, taskExecuteResponseProcessor);
        // Task termination request processor
        this.nettyRemotingServer.registerProcessor(CommandType.TASK_KILL_RESPONSE, taskKillResponseProcessor);
        ...
        this.nettyRemotingServer.start();
        logger.info("Started Master RPC Server...");
    } 1.2 Task Plugin Initialization: Description: Task-related template operations such as creating tasks, parsing task parameters, and retrieving task resource information. This plugin needs to be registered on the API, Master, and Worker sides. The role in Master is to set the data source and UDF information. Description: Task-related template operations such as creating tasks, parsing task parameters, and retrieving task resource information. This plugin needs to be registered on the API, Master, and Worker sides. The role in Master is to set the data source and UDF information. Description 1.3 Self-Tolerant (Master Registration): Description: Registers the master’s information to the registry (using Zookeeper as an example), and listens for changes in the registration of itself, other masters, and all worker nodes to perform fault-tolerant processing.
Code Entry: org.apache.dolphinscheduler.server.master.registry.MasterRegistryClient#start Description: Registers the master’s information to the registry (using Zookeeper as an example), and listens for changes in the registration of itself, other masters, and all worker nodes to perform fault-tolerant processing. Description Code Entry: org.apache.dolphinscheduler.server.master.registry.MasterRegistryClient#start Code Entry org.apache.dolphinscheduler.server.master.registry.MasterRegistryClient#start public void start() {
        try {
            this.masterHeartBeatTask = new MasterHeartBeatTask(masterConfig, registryClient);
            // 1. Register itself to the registry;
            registry();
            // 2. Listen to the connection state with the registry;
            registryClient.addConnectionStateListener(
                    new MasterConnectionStateListener(masterConfig, registryClient, masterConnectStrategy));
            // 3. Listen to the status of other masters and workers in the registry and perform fault-tolerant work
            registryClient.subscribe(REGISTRY_DOLPHINSCHEDULER_NODE, new MasterRegistryDataListener());
        } catch (Exception e) {
            throw new RegistryException("Master registry client start up error", e);
        }
    } public void start() {
        try {
            this.masterHeartBeatTask = new MasterHeartBeatTask(masterConfig, registryClient);
            // 1. Register itself to the registry;
            registry();
            // 2. Listen to the connection state with the registry;
            registryClient.addConnectionStateListener(
                    new MasterConnectionStateListener(masterConfig, registryClient, masterConnectStrategy));
            // 3. Listen to the status of other masters and workers in the registry and perform fault-tolerant work
            registryClient.subscribe(REGISTRY_DOLPHINSCHEDULER_NODE, new MasterRegistryDataListener());
        } catch (Exception e) {
            throw new RegistryException("Master registry client start up error", e);
        }
    } 1.4 Master Scheduling: Description: A scanning thread that periodically checks the command table in the database and performs different operations based on command types. This is the core logic for workflow startup, instance fault tolerance, etc.
Code Entry: org.apache.dolphinscheduler.server.master.runner.MasterSchedulerBootstrap#run Description: A scanning thread that periodically checks the command table in the database and performs different operations based on command types. This is the core logic for workflow startup, instance fault tolerance, etc. Description command Code Entry: org.apache.dolphinscheduler.server.master.runner.MasterSchedulerBootstrap#run Code Entry org.apache.dolphinscheduler.server.master.runner.MasterSchedulerBootstrap#run public void run() {
        while (!ServerLifeCycleManager.isStopped()) {
            try {
                // If the server is not in running status, it cannot consume commands
                if (!ServerLifeCycleManager.isRunning()) {
                    logger.warn("The current server {} is not at running status, cannot consume commands.", this.masterAddress);
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                }

                // Handle workload overload (CPU/memory)
                boolean isOverload = OSUtils.isOverload(masterConfig.getMaxCpuLoadAvg(), masterConfig.getReservedMemory());
                if (isOverload) {
                    logger.warn("The current server {} is overload, cannot consume commands.", this.masterAddress);
                    MasterServerMetrics.incMasterOverload();
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                    continue;
                }

                // Get commands from the database
                List<Command> commands = findCommands();
                if (CollectionUtils.isEmpty(commands)) {
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                    continue;
                }

                // Convert commands to process instances and handle the workflow logic
                List<ProcessInstance> processInstances = command2ProcessInstance(commands);
                if (CollectionUtils.isEmpty(processInstances)) {
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                    continue;
                }

                // Handle workflow instance execution
                processInstances.forEach(processInstance -> {
                    try {
                        LoggerUtils.setWorkflowInstanceIdMDC(processInstance.getId());
                        WorkflowExecuteRunnable workflowRunnable = new WorkflowExecuteRunnable(processInstance, ...);
                        processInstanceExecCacheManager.cache(processInstance.getId(), workflowRunnable);
                        workflowEventQueue.addEvent(new WorkflowEvent(WorkflowEventType.START_WORKFLOW, processInstance.getId()));
                    } finally {
                        LoggerUtils.removeWorkflowInstanceIdMDC();
                    }
                });
            } catch (InterruptedException interruptedException) {
                logger.warn("Master schedule bootstrap interrupted, close the loop", interruptedException);
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                logger.error("Master schedule workflow error", e);
                ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
            }
        }
    } public void run() {
        while (!ServerLifeCycleManager.isStopped()) {
            try {
                // If the server is not in running status, it cannot consume commands
                if (!ServerLifeCycleManager.isRunning()) {
                    logger.warn("The current server {} is not at running status, cannot consume commands.", this.masterAddress);
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                }

                // Handle workload overload (CPU/memory)
                boolean isOverload = OSUtils.isOverload(masterConfig.getMaxCpuLoadAvg(), masterConfig.getReservedMemory());
                if (isOverload) {
                    logger.warn("The current server {} is overload, cannot consume commands.", this.masterAddress);
                    MasterServerMetrics.incMasterOverload();
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                    continue;
                }

                // Get commands from the database
                List<Command> commands = findCommands();
                if (CollectionUtils.isEmpty(commands)) {
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                    continue;
                }

                // Convert commands to process instances and handle the workflow logic
                List<ProcessInstance> processInstances = command2ProcessInstance(commands);
                if (CollectionUtils.isEmpty(processInstances)) {
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                    continue;
                }

                // Handle workflow instance execution
                processInstances.forEach(processInstance -> {
                    try {
                        LoggerUtils.setWorkflowInstanceIdMDC(processInstance.getId());
                        WorkflowExecuteRunnable workflowRunnable = new WorkflowExecuteRunnable(processInstance, ...);
                        processInstanceExecCacheManager.cache(processInstance.getId(), workflowRunnable);
                        workflowEventQueue.addEvent(new WorkflowEvent(WorkflowEventType.START_WORKFLOW, processInstance.getId()));
                    } finally {
                        LoggerUtils.removeWorkflowInstanceIdMDC();
                    }
                });
            } catch (InterruptedException interruptedException) {
                logger.warn("Master schedule bootstrap interrupted, close the loop", interruptedException);
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                logger.error("Master schedule workflow error", e);
                ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
            }
        }
    } 1.5 Event Execution Service: Description: Responsible for polling the event queue of the workflow instance. Events such as workflow submission failures or task state changes are handled here.
Code Entry: org.apache.dolphinscheduler.server.master.runner.EventExecuteService#run Description: Responsible for polling the event queue of the workflow instance. Events such as workflow submission failures or task state changes are handled here. Description Code Entry: org.apache.dolphinscheduler.server.master.runner.EventExecuteService#run Code Entry org.apache.dolphinscheduler.server.master.runner.EventExecuteService#run public void run() {
        while (!ServerLifeCycleManager.isStopped()) {
            try {
                // Handle workflow execution events
                workflowEventHandler();
                // Handle stream task execution events
                streamTaskEventHandler();
                TimeUnit.MILLISECONDS.sleep(Constants.SLEEP_TIME_MILLIS_SHORT);
            } ...
        }
    } public void run() {
        while (!ServerLifeCycleManager.isStopped()) {
            try {
                // Handle workflow execution events
                workflowEventHandler();
                // Handle stream task execution events
                streamTaskEventHandler();
                TimeUnit.MILLISECONDS.sleep(Constants.SLEEP_TIME_MILLIS_SHORT);
            } ...
        }
    } 1.6 Fault Tolerance Mechanism: Description: Responsible for Master and Worker fault tolerance.
Code Entry: org.apache.dolphinscheduler.server.master.service.MasterFailoverService#checkMasterFailover Description: Responsible for Master and Worker fault tolerance. Description Code Entry: org.apache.dolphinscheduler.server.master.service.MasterFailoverService#checkMasterFailover Code Entry org.apache.dolphinscheduler.server.master.service.MasterFailoverService#checkMasterFailover public void checkMasterFailover() {
        List<String> needFailoverMasterHosts = processService.queryNeedFailoverProcessInstanceHost()
                .stream()
                .filter(host -> localAddress.equals(host) || !registryClient.checkNodeExists(host, NodeType.MASTER))
                .distinct()
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(needFailoverMasterHosts)) {
            return;
        }

        for (String needFailoverMasterHost : needFailoverMasterHosts) {
            failoverMaster(needFailoverMasterHost);
        }
    } public void checkMasterFailover() {
        List<String> needFailoverMasterHosts = processService.queryNeedFailoverProcessInstanceHost()
                .stream()
                .filter(host -> localAddress.equals(host) || !registryClient.checkNodeExists(host, NodeType.MASTER))
                .distinct()
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(needFailoverMasterHosts)) {
            return;
        }

        for (String needFailoverMasterHost : needFailoverMasterHosts) {
            failoverMaster(needFailoverMasterHost);
        }
    } Conclusion: The article provides an in-depth look at the Apache DolphinScheduler 3.1.9 Master service startup process, fault tolerance mechanisms, and the overall architecture. Further articles will explore the Worker startup process and interactions between Master and Worker.

The code in this story is for educational purposes. The readers are solely responsible for whatever they build with it.

Dissecting the Master Server: How DolphinScheduler Powers Workflow Scheduling

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

A 10-Minute Deep Dive Into the Core Architecture of Apache SeaTunnel and DataX

How to Seamlessly Transfer From Airflow to Apache DolphinScheduler With Air2phin

A Comprehensive Guide to Building DolphinScheduler 3.2.0 Production-Grade Cluster Deployment

How to Regularly Delete Log Instances in DolphinScheduler

How to Enable Auto-Start for Apache DolphinScheduler

Solutions for Upgrading Apache DolphinScheduler from Version 1.3.4 to 3.1.2

A 10-Minute Deep Dive Into the Core Architecture of Apache SeaTunnel and DataX

How to Seamlessly Transfer From Airflow to Apache DolphinScheduler With Air2phin

A Comprehensive Guide to Building DolphinScheduler 3.2.0 Production-Grade Cluster Deployment

How to Regularly Delete Log Instances in DolphinScheduler

How to Enable Auto-Start for Apache DolphinScheduler

Solutions for Upgrading Apache DolphinScheduler from Version 1.3.4 to 3.1.2

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps