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.
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
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
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.
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
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
commandtable 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
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
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
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.