fix bug: send task result to master until success. (#2839)

* fix bug: send task result to master until success.

* fix bug: send task result to master until success.

* add sleep

* set null if send result failed

* set null if send result failed

Co-authored-by: baoliang <baoliang@analysys.com.cn>

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

@@ -102,20 +102,23 @@ public class TaskCallbackService {
         while (Stopper.isRunning()) {
             masterNodes = zookeeperRegistryCenter.getMasterNodesDirectly();
             if (CollectionUtils.isEmpty(masterNodes)) {
+                masterNodes = null;
                 ThreadUtils.sleep(SLEEP_TIME_MILLIS);
-            }else {
-                logger.error("find {} masters for task : {}.",
-                        masterNodes.size(),
-                        taskInstanceId);
-                break;
+                continue;
             }
-        }
-        for(String masterNode : masterNodes){
-            newChannel = nettyRemotingClient.getChannel(Host.of(masterNode));
-            if(newChannel != null){
-                return getRemoteChannel(newChannel, nettyRemoteChannel.getOpaque(), taskInstanceId);
+            logger.info("find {} masters for task : {}.",
+                    masterNodes.size(),
+                    taskInstanceId);
+            for (String masterNode : masterNodes) {
+                newChannel = nettyRemotingClient.getChannel(Host.of(masterNode));
+                if (newChannel != null) {
+                    return getRemoteChannel(newChannel, nettyRemoteChannel.getOpaque(), taskInstanceId);
+                }
             }
+            masterNodes = null;
+            ThreadUtils.sleep(SLEEP_TIME_MILLIS);
         }
+
         throw new IllegalStateException(String.format("all available master nodes : %s are not reachable for task: {}", masterNodes, taskInstanceId));
     }