[Fix-#4840][worker] fix master fault tolerance when startup (#4845)

* [Fix-#4840][worker] fix master fault tolerance when startup

* [Fix-#4840][worker] move masterRegistry.unRegistry to zkMasterClient.close

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

@@ -25,12 +25,14 @@ import org.apache.dolphinscheduler.server.master.config.MasterConfig;
 import org.apache.dolphinscheduler.server.master.processor.TaskAckProcessor;
 import org.apache.dolphinscheduler.server.master.processor.TaskKillResponseProcessor;
 import org.apache.dolphinscheduler.server.master.processor.TaskResponseProcessor;
-import org.apache.dolphinscheduler.server.master.registry.MasterRegistry;
 import org.apache.dolphinscheduler.server.master.runner.MasterSchedulerService;
 import org.apache.dolphinscheduler.server.worker.WorkerServer;
 import org.apache.dolphinscheduler.server.zk.ZKMasterClient;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 import org.apache.dolphinscheduler.service.quartz.QuartzExecutors;
+
+import javax.annotation.PostConstruct;
+
 import org.quartz.SchedulerException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,8 +42,6 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.FilterType;
 
-import javax.annotation.PostConstruct;
-
 
 
 
@@ -73,12 +73,6 @@ public class MasterServer {
      */
     private NettyRemotingServer nettyRemotingServer;
 
-    /**
-     * master registry
-     */
-    @Autowired
-    private MasterRegistry masterRegistry;
-
     /**
      * zk master client
      */
@@ -117,9 +111,6 @@ public class MasterServer {
         this.nettyRemotingServer.registerProcessor(CommandType.TASK_KILL_RESPONSE, new TaskKillResponseProcessor());
         this.nettyRemotingServer.start();
 
-        // register
-        this.masterRegistry.registry();
-
         // self tolerant
         this.zkMasterClient.start();
 
@@ -178,7 +169,6 @@ public class MasterServer {
             //
             this.masterSchedulerService.close();
             this.nettyRemotingServer.close();
-            this.masterRegistry.unRegistry();
             this.zkMasterClient.close();
             //close quartz
             try{

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/zk/ZKMasterClient.java

@@ -31,6 +31,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.remote.utils.Host;
 import org.apache.dolphinscheduler.server.builder.TaskExecutionContextBuilder;
 import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
+import org.apache.dolphinscheduler.server.master.registry.MasterRegistry;
 import org.apache.dolphinscheduler.server.utils.ProcessUtils;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 import org.apache.dolphinscheduler.service.zk.AbstractZKClient;
@@ -66,6 +67,12 @@ public class ZKMasterClient extends AbstractZKClient {
     @Autowired
     private ProcessService processService;
 
+    /**
+     * master registry
+     */
+    @Autowired
+    private MasterRegistry masterRegistry;
+
     public void start() {
 
         InterProcessMutex mutex = null;
@@ -75,6 +82,9 @@ public class ZKMasterClient extends AbstractZKClient {
             mutex = new InterProcessMutex(getZkClient(), znodeLock);
             mutex.acquire();
 
+            //  Master registry
+            masterRegistry.registry();
+
             // init system znode
             this.initSystemZNode();
 
@@ -98,6 +108,7 @@ public class ZKMasterClient extends AbstractZKClient {
     @Override
     public void close() {
         super.close();
+        masterRegistry.unRegistry();
     }
 
     /**