[Bug-9719][Master] fix failover fail because task plugins has not been loaded (#9720)

dev
caishunfeng 2022-04-24 20:34:21 +08:00 committed by GitHub
parent ebc4253d50
commit 5657cb9aec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 14 additions and 8 deletions

View File

@ -36,6 +36,7 @@ import org.apache.dolphinscheduler.server.master.runner.EventExecuteService;
import org.apache.dolphinscheduler.server.master.runner.FailoverExecuteThread; import org.apache.dolphinscheduler.server.master.runner.FailoverExecuteThread;
import org.apache.dolphinscheduler.server.master.runner.MasterSchedulerService; import org.apache.dolphinscheduler.server.master.runner.MasterSchedulerService;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext; import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.task.TaskPluginManager;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
@ -68,6 +69,9 @@ public class MasterServer implements IStoppable {
@Autowired @Autowired
private MasterRegistryClient masterRegistryClient; private MasterRegistryClient masterRegistryClient;
@Autowired
private TaskPluginManager taskPluginManager;
@Autowired @Autowired
private MasterSchedulerService masterSchedulerService; private MasterSchedulerService masterSchedulerService;
@ -131,6 +135,9 @@ public class MasterServer implements IStoppable {
this.nettyRemotingServer.start(); this.nettyRemotingServer.start();
// install task plugin
this.taskPluginManager.installPlugin();
// self tolerant // self tolerant
this.masterRegistryClient.init(); this.masterRegistryClient.init();
this.masterRegistryClient.start(); this.masterRegistryClient.start();

View File

@ -50,6 +50,9 @@ public class FailoverExecuteThread extends Thread {
@Override @Override
public void run() { public void run() {
// when startup, wait 10s for ready
ThreadUtils.sleep((long) Constants.SLEEP_TIME_MILLIS * 10);
logger.info("failover execute thread started"); logger.info("failover execute thread started");
while (Stopper.isRunning()) { while (Stopper.isRunning()) {
try { try {

View File

@ -33,7 +33,6 @@ import org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutor
import org.apache.dolphinscheduler.server.master.registry.ServerNodeManager; import org.apache.dolphinscheduler.server.master.registry.ServerNodeManager;
import org.apache.dolphinscheduler.service.alert.ProcessAlertManager; import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
import org.apache.dolphinscheduler.service.process.ProcessService; import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.task.TaskPluginManager;
import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.collections4.CollectionUtils;
@ -102,9 +101,6 @@ public class MasterSchedulerService extends Thread {
@Autowired @Autowired
private StateWheelExecuteThread stateWheelExecuteThread; private StateWheelExecuteThread stateWheelExecuteThread;
@Autowired
private TaskPluginManager taskPluginManager;
/** /**
* constructor of MasterSchedulerService * constructor of MasterSchedulerService
*/ */

View File

@ -40,8 +40,6 @@ import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
@Component @Component
@ -86,8 +84,7 @@ public class TaskPluginManager {
return taskChannel.parseParameters(parametersNode); return taskChannel.parseParameters(parametersNode);
} }
@EventListener public void installPlugin() {
public void installPlugin(ApplicationReadyEvent readyEvent) {
final Set<String> names = new HashSet<>(); final Set<String> names = new HashSet<>();
ServiceLoader.load(TaskChannelFactory.class).forEach(factory -> { ServiceLoader.load(TaskChannelFactory.class).forEach(factory -> {

View File

@ -155,6 +155,9 @@ public class WorkerServer implements IStoppable {
this.nettyRemotingServer.start(); this.nettyRemotingServer.start();
// install task plugin
this.taskPluginManager.installPlugin();
// worker registry // worker registry
try { try {
this.workerRegistryClient.registry(); this.workerRegistryClient.registry();