Refactor worker (#2241)

* let quartz use the same datasource

* move master/worker config from dao.properties to each config
add master/worker registry test

* move mybatis config from application.properties to SpringConnectionFactory

* move mybatis-plus config from application.properties to SpringConnectionFactory

* refactor TaskCallbackService

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackService.java

@@ -21,13 +21,18 @@ package org.apache.dolphinscheduler.server.worker.processor;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
+import org.apache.dolphinscheduler.common.utils.CollectionUtils;
 import org.apache.dolphinscheduler.remote.NettyRemotingClient;
 import org.apache.dolphinscheduler.remote.command.Command;
 import org.apache.dolphinscheduler.remote.config.NettyClientConfig;
+import org.apache.dolphinscheduler.remote.utils.Host;
+import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
@@ -43,6 +48,12 @@ public class TaskCallbackService {
      */
     private static final ConcurrentHashMap<Integer, NettyRemoteChannel> REMOTE_CHANNELS = new ConcurrentHashMap<>();
 
+    /**
+     * zookeeper register center
+     */
+    @Autowired
+    private ZookeeperRegistryCenter zookeeperRegistryCenter;
+
     /**
      * netty remoting client
      */
@@ -75,12 +86,27 @@ public class TaskCallbackService {
         }
         Channel newChannel = nettyRemotingClient.getChannel(nettyRemoteChannel.getHost());
         if(newChannel != null){
-            NettyRemoteChannel remoteChannel = new NettyRemoteChannel(newChannel, nettyRemoteChannel.getOpaque());
+            return getRemoteChannel(newChannel, nettyRemoteChannel.getOpaque(), taskInstanceId);
+        }
+        logger.warn("original master : {} is not reachable, random select master", nettyRemoteChannel.getHost());
+        Set<String> masterNodes = zookeeperRegistryCenter.getMasterNodesDirectly();
+        if(CollectionUtils.isEmpty(masterNodes)){
+            throw new IllegalStateException("no available master node exception");
+        }
+        for(String masterNode : masterNodes){
+            newChannel = nettyRemotingClient.getChannel(Host.of(masterNode));
+            if(newChannel != null){
+                return getRemoteChannel(newChannel, nettyRemoteChannel.getOpaque(), taskInstanceId);
+            }
+        }
+        throw new IllegalStateException(String.format("all available master nodes : %s are not reachable", masterNodes));
+    }
+
+    private NettyRemoteChannel getRemoteChannel(Channel newChannel, long opaque, int taskInstanceId){
+        NettyRemoteChannel remoteChannel = new NettyRemoteChannel(newChannel, opaque);
         addRemoteChannel(taskInstanceId, remoteChannel);
         return remoteChannel;
     }
-        return null;
-    }
 
     /**
      *  remove callback channels
@@ -97,12 +123,8 @@ public class TaskCallbackService {
      */
     public void sendAck(int taskInstanceId, Command command){
         NettyRemoteChannel nettyRemoteChannel = getRemoteChannel(taskInstanceId);
-        if(nettyRemoteChannel == null){
-            //TODO
-        } else{
         nettyRemoteChannel.writeAndFlush(command);
     }
-    }
 
     /**
      *  send result
@@ -112,9 +134,6 @@ public class TaskCallbackService {
      */
     public void sendResult(int taskInstanceId, Command command){
         NettyRemoteChannel nettyRemoteChannel = getRemoteChannel(taskInstanceId);
-        if(nettyRemoteChannel == null){
-            //TODO
-        } else{
         nettyRemoteChannel.writeAndFlush(command).addListener(new ChannelFutureListener(){
 
             @Override
@@ -126,5 +145,4 @@ public class TaskCallbackService {
             }
         });
     }
-    }
 }