DEV Community

Chen Debra
Chen Debra

Posted on

Inside DolphinScheduler’s Master Startup Process: A Source Code Walkthrough

#apachedolphinscheduler #opensource #github #datascience

Apache DolphinScheduler is a distributed, highly extensible, and visual workflow scheduler designed for creating, scheduling, and monitoring enterprise-scale big data workflows. As a core scheduling component, the Master node receives task commands, schedules DAG workflows, dispatches tasks to Workers, and handles fault tolerance and high availability across the cluster.

This article walks through the Master startup process in DolphinScheduler 3.2.0 from the source-code level. By tracing the startup sequence and examining the key modules involved, you can quickly build an understanding of how the Master service initializes and how its scheduling engine works.

1. How a Manually Triggered Workflow Starts

When you click “Run” in the Web UI to manually trigger a workflow, the workflow does not start executing immediately. Instead, DolphinScheduler first writes a Command to the database.

The entry point is ExecutorController:

@PostMapping(value = "start-process-instance")
public Result startProcessInstance(@RequestBody StartProcessInstanceCommand command) {
    // Parameter validation and permission checks...
    executorService.execProcessInstance(command);
    return Result.success();
}
Enter fullscreen mode Exit fullscreen mode

The subsequent call chain is:

ExecutorServiceImpl#execProcessInstance(...)
→ createCommand(...)
→ CommandServiceImpl#createCommand(Command)
→ CommandMapper#insert(Command)
Enter fullscreen mode Exit fullscreen mode

At this stage, DolphinScheduler only inserts a record into the t_ds_command table. After startup, the Master continuously polls this table and consumes pending commands, after which it actually restores and executes the corresponding workflow instances.

2. MasterServer Startup Entry Point and Overall Flow

The Master startup entry point is the org.apache.dolphinscheduler.server.master.MasterServer class. Its @PostConstruct-annotated run() method initializes and starts the major components in sequence:

@PostConstruct
public void run() throws SchedulerException {
    // 1. Start the RPC services (Server + Client)
    masterRPCServer.start();
    masterRpcClient.start();
​
    // 2. Load Task plugins
    taskPluginManager.loadPlugin();
​
    // 3. Start the registry client (register the Master and monitor cluster changes)
    masterRegistryClient.start();
    masterRegistryClient.setRegistryStoppable(this);
​
    // 4. Start the core scheduling engine
    masterSchedulerBootstrap.start();
​
    // 5. Start the asynchronous event processing service
    eventExecuteService.start();
​
    // 6. Start the failover thread
    failoverExecuteThread.start();
​
    // 7. Start the Quartz scheduler
    schedulerApi.start();
​
    // Add a JVM shutdown hook
    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        if (!ServerLifeCycleManager.isStopped()) {
            close("MasterServer shutdownHook");
        }
    }));
}
Enter fullscreen mode Exit fullscreen mode

Let's walk through each step to understand the implementation details and the design behind them.

3. Starting the Master RPC Services

Inside DolphinScheduler, the Master uses Netty to implement RPC communication. The RPC layer consists of an RPC Server, which receives heartbeats and status updates from Workers, and an RPC Client, which sends task commands to Workers.

3.1 Starting the RPC Server

The MasterRPCServer.start() method initializes the RPC server as follows:

public void start() {
    log.info("Starting Master RPC Server...");
    NettyServerConfig serverConfig = masterConfig.getMasterRpcServerConfig();
    serverConfig.setListenPort(masterConfig.getListenPort());
    this.nettyRemotingServer = new NettyRemotingServer(serverConfig);
    // Register all MasterRpcProcessors to handle incoming messages
    for (MasterRpcProcessor processor : masterRpcProcessors) {
        this.nettyRemotingServer.registerProcessor(processor);
    }
    this.nettyRemotingServer.start();
    log.info("Started Master RPC Server...");
}
Enter fullscreen mode Exit fullscreen mode

When NettyRemotingServer is constructed, DolphinScheduler selects either Epoll or NIO based on the runtime environment and initializes the bossGroup and workerGroup:

public NettyRemotingServer(NettyServerConfig config) {
    if (Epoll.isAvailable()) {
        bossGroup = new EpollEventLoopGroup(1, bossFactory);
        workGroup = new EpollEventLoopGroup(config.getWorkerThread(), workerFactory);
    } else {
        bossGroup = new NioEventLoopGroup(1, bossFactory);
        workGroup = new NioEventLoopGroup(config.getWorkerThread(), workerFactory);
    }
    // Initialize ServerBootstrap and register handlers
}
Enter fullscreen mode Exit fullscreen mode

During start(), ServerBootstrap.bind(...) binds the server to the configured port. The default RPC port is 5678:

serverBootstrap
    .group(bossGroup, workGroup)
    .channel(NettyUtils.getServerSocketChannelClass())
    .childHandler(new ChannelInitializer<SocketChannel>() {
        protected void initChannel(SocketChannel ch) {
            initNettyChannel(ch);
        }
    });
ChannelFuture future = serverBootstrap.bind(serverConfig.getListenPort()).sync();
Enter fullscreen mode Exit fullscreen mode

Once the server successfully binds to the port, the Master can receive RPC requests from Workers and other components.

3.2 Starting the RPC Client

MasterRpcClient.start() initializes the NettyRemotingClient, but it does not proactively establish connections to Workers:

public void start() {
    client = new NettyRemotingClient(masterConfig.getMasterRpcClientConfig());
    log.info("Success initialized MasterRPCClient...");
}
Enter fullscreen mode Exit fullscreen mode

The NettyRemotingClient constructor also selects Epoll or NIO depending on the environment. It initializes the workerGroup, callbackExecutor, and responseFutureExecutor, and starts the response-future scanning mechanism:

bootstrap
    .group(workerGroup)
    .channel(NettyUtils.getSocketChannelClass())
    .handler(new ChannelInitializer<SocketChannel>() {
        public void initChannel(SocketChannel ch) {
            ch.pipeline()
              .addLast(new IdleStateHandler(...))
              .addLast(new NettyDecoder(), clientHandler, encoder);
        }
    });
responseFutureExecutor.scheduleWithFixedDelay(ResponseFuture::scanFutureTable, 0, 1, TimeUnit.SECONDS);
Enter fullscreen mode Exit fullscreen mode

In other words, the Master initializes the client-side RPC infrastructure during startup. Actual connections to Workers are established when the Master needs to dispatch tasks.

4. Plugin Loading

DolphinScheduler uses Java SPI to dynamically load Task plugins, enabling integration with multiple execution engines such as Hive, Spark, and Flink.

The implementation of TaskPluginManager.loadPlugin() is:

public void loadPlugin() {
    PrioritySPIFactory<TaskChannelFactory> factory = new PrioritySPIFactory<>(TaskChannelFactory.class);
    for (Map.Entry<String, TaskChannelFactory> entry : factory.getSPIMap().entrySet()) {
        String name = entry.getKey();
        TaskChannelFactory plugin = entry.getValue();
        taskChannelFactoryMap.put(name, plugin);
        taskChannelMap.put(name, plugin.create());
    }
}
Enter fullscreen mode Exit fullscreen mode

Internally, PrioritySPIFactory uses ServiceLoader.load(spiClass) to scan implementations registered under META-INF/services and handles conflicts between implementations with the same name:

for (T impl : ServiceLoader.load(spiClass)) {
    String key = impl.getIdentify().getName();
    if (map.containsKey(key)) resolveConflict(impl);
    else map.put(key, impl);
}
Enter fullscreen mode Exit fullscreen mode

This SPI-based architecture allows Task-related capabilities to be extended without tightly coupling the core scheduling engine to every specific execution technology.

5. Registry Client Initialization and Heartbeat Maintenance

The Master communicates with ZooKeeper, or another supported registry, through masterRegistryClient. This component handles several key responsibilities:

  1. Register the current Master node: Create an ephemeral node under /dolphinscheduler/master and store heartbeat information.
  2. Maintain heartbeats: Periodically update node information to prevent the cluster from treating the Master as unavailable.
  3. Monitor cluster changes: Subscribe to /dolphinscheduler/servers so the Master can dynamically detect Master and Worker nodes joining or leaving the cluster.

The startup logic is:

public void start() {
    this.masterHeartBeatTask = new MasterHeartBeatTask(masterConfig, registryClient);
    registry(); // Register the Master and start the heartbeat
    registryClient.addConnectionStateListener(new MasterConnectionStateListener(...));
    registryClient.subscribe(RegistryNodeType.ALL_SERVERS.getRegistryPath(), new MasterRegistryDataListener());
}
Enter fullscreen mode Exit fullscreen mode

The core registration logic is:

void registry() {
    registryClient.remove(masterPath);
    registryClient.persistEphemeral(masterPath, JSONUtils.toJsonString(heartbeat));
    while (!registryClient.checkNodeExists(host, MASTER)) {
        ThreadUtils.sleep(3000);
    }
    masterHeartBeatTask.start();
}
Enter fullscreen mode Exit fullscreen mode

The MasterRegistryDataListener handles registry events through handleMasterEvent() and handleWorkerEvent(), triggering failover processing or resource cleanup when cluster membership changes.

6. Starting the Core Scheduling Engine

The core scheduling engine is started by MasterSchedulerBootstrap, which brings together three major components:

  • Command recovery: Queries all pending Commands, creates the corresponding WorkflowExecuteRunnable instances, and adds them to the execution cache and event queue.
  • Event loop: WorkflowEventLooper continuously consumes events from workflowEventQueue and invokes the appropriate handler based on the event type.
  • Task executor: MasterTaskExecutorBootstrap starts the thread pools and queues responsible for consuming pending tasks and dispatching them to Workers through RPC.

The startup sequence is:

public synchronized void start() {
    super.start();              // part1: recover and submit Commands
    workflowEventLooper.start(); // part2: start the event loop
    masterTaskExecutorBootstrap.start(); // part3: dispatch tasks
}
Enter fullscreen mode Exit fullscreen mode

6.1 Recovering Commands

The recovery process can be illustrated as follows:

List<Command> commands = findCommands();
commands.parallelStream().forEach(cmd -> {
    Optional<WorkflowExecuteRunnable> opt = factory.create(cmd);
    if (opt.isPresent()) {
        cacheManager.cache(id, runnable);
        workflowEventQueue.addEvent(new WorkflowEvent(START_WORKFLOW, id));
    }
});
Enter fullscreen mode Exit fullscreen mode

The Master retrieves pending Commands, creates the corresponding workflow execution objects, caches them, and then pushes START_WORKFLOW events into the workflow event queue.

This is the key transition from a command stored in the database to an actual workflow execution process.

6.2 The Event Loop

WorkflowEventLooper implements Runnable:

public void run() {
    while (RUNNING) {
        WorkflowEvent event = queue.poolEvent();
        try (MDCAutoClosableContext ctx = setWorkflowIdMDC(...)) {
            handlerMap.get(event.getType()).handle(event);
        }
    }
}
Enter fullscreen mode Exit fullscreen mode

When a START_WORKFLOW event is received, it is routed to WorkflowStartHandler, which invokes WorkflowExecuteRunnable.call():

public WorkflowStartStatus startWorkflow() {
    initTaskQueue();     // Initialize the DAG task queue
    submitPostNode(null); // Submit the first node
    return SUCCESS;
}
Enter fullscreen mode Exit fullscreen mode

At this point, the workflow execution engine begins processing the DAG and submitting its executable tasks.

6.3 Task Dispatch

MasterTaskExecutorBootstrap starts three processing loops:

globalTaskDispatchLooper.start();
masterDelayTaskLooper.start();
asyncMasterTaskDelayLooper.start();
Enter fullscreen mode Exit fullscreen mode

Among them, globalTaskDispatchLooper retrieves DefaultTaskExecuteRunnable instances from globalTaskDispatchWaitingQueue and invokes taskDispatcher.dispatch():

Message msg = taskDispatchRequest.convert2Command();
masterRpcClient.sendSyncCommand(host, msg, timeout);
Enter fullscreen mode Exit fullscreen mode

The task dispatch request is converted into an RPC message and sent synchronously to the target Worker.

This is the point where the Master moves from workflow-level orchestration to actual task execution on a Worker node.

7. Event Processing Service

Event processing is divided into two main categories:

  • Workflow events: Events such as task status changes and workflow blocking.
  • Stream events: Custom events associated with streaming tasks.

EventExecuteService.start() starts the corresponding thread pools and continuously consumes events from the stateEvents and taskEvents queues:

workflowEventHandler(); // Submit to workflowExecuteThreadPool
streamTaskEventHandler(); // Submit to streamTaskExecuteThreadPool
Enter fullscreen mode Exit fullscreen mode

By processing events asynchronously through dedicated thread pools, the Master can separate event production from event handling and avoid coupling the execution of different types of events too tightly.

8. Failover Processing

The failoverExecuteThread periodically checks the health of Master and Worker nodes. When a node becomes unavailable, DolphinScheduler handles two types of failover scenarios: Master Failover and Worker Failover.

8.1 Master Failover

When a Master node goes down, the registry removes its node. Other Master nodes detect the change and trigger failover processing:

failoverService.failoverServerWhenDown(serverHost, MASTER);
Enter fullscreen mode Exit fullscreen mode

doFailoverMaster then:

  • Queries the ProcessInstances that require failover.
  • Calls processService.processNeedFailoverProcessInstances(processInstance) for each instance.
  • Writes the required execution commands back to t_ds_command.

This allows another available Master to take over the affected workflow instances and continue the scheduling process.

8.2 Worker Failover

When a Worker node goes down:

failoverService.failoverServerWhenDown(workerHost, WORKER);
Enter fullscreen mode Exit fullscreen mode

failoverWorker queries the TaskInstance records for tasks that were running on the failed Worker. For each unfinished task, it:

  • Sets state = NEED_FAULT_TOLERANCE.
  • Updates the database through taskInstanceDao.upsert.
  • Submits a TaskStateEvent to workflowExecuteThreadPool, allowing the task to be dispatched again to another available Worker.

Through this mechanism, task execution can recover from Worker failures without requiring the workflow itself to be restarted from scratch.

9. Starting the Quartz Scheduler

In addition to manually triggered workflows, the Master uses Quartz to support scheduled execution.

SchedulerApi injects an org.quartz.Scheduler instance:

@Override
public void start() throws SchedulerException {
    scheduler.start();
}
Enter fullscreen mode Exit fullscreen mode

The Quartz Scheduler is used to trigger scheduled jobs periodically, including tasks such as workflow dependency handling and scheduled workflow execution.

This complements the command-driven execution path used for manually triggered workflows.

10. Summary and Key Takeaways

This article has walked through the complete startup and execution flow of the Master node in DolphinScheduler 3.2.0, from manually writing a Command to the database, to MasterServer initialization, RPC service setup, plugin loading, registry integration, core scheduling engine startup, event processing, failover handling, and Quartz-based scheduled execution.

At the architectural level, the Master demonstrates a modular and decoupled design:

  • Network communication: RPC communication is implemented with Netty.
  • Plugin extensibility: Java SPI is used to dynamically load TaskChannel implementations.
  • High availability: Registry-based heartbeats and the Failover service provide fault-tolerance capabilities.
  • Scheduling engine: Parallel processing and an event-driven architecture work together to drive workflow execution.
  • Scheduled execution: Quartz provides additional scheduling capabilities for time-based tasks.

At the source-code level, the core logic is built around database operations, queues, event processing, and RPC messaging. Developers can continue tracing the call chain to explore the implementation details that are not covered here, and combine the source code with the DolphinScheduler community documentation for a deeper understanding of how the Master works.

Once you follow the startup path from MasterServer.run() through command recovery, event handling, task dispatch, and failover, the Master is no longer a black box: it becomes a clearly connected execution pipeline that turns workflow commands into reliable task execution across the cluster.

Top comments (0)