[Improvement][Server] Must restart master if Zk reconnect (#5210) (#5211)

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistry.java

@@ -90,10 +90,8 @@ public class MasterRegistry {
                     logger.error("master : {} connection lost from zookeeper", address);
                 } else if (newState == ConnectionState.RECONNECTED) {
                     logger.info("master : {} reconnected to zookeeper", address);
-                    zookeeperRegistryCenter.getRegisterOperator().persistEphemeral(localNodePath, "");
                 } else if (newState == ConnectionState.SUSPENDED) {
                     logger.warn("master : {} connection SUSPENDED ", address);
-                    zookeeperRegistryCenter.getRegisterOperator().persistEphemeral(localNodePath, "");
                 }
             });
         int masterHeartbeatInterval = masterConfig.getMasterHeartbeatInterval();

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistry.java

@@ -111,10 +111,8 @@ public class WorkerRegistry {
                         logger.error("worker : {} connection lost from zookeeper", address);
                     } else if (newState == ConnectionState.RECONNECTED) {
                         logger.info("worker : {} reconnected to zookeeper", address);
-                        zookeeperRegistryCenter.getRegisterOperator().persistEphemeral(workerZKPath, "");
                     } else if (newState == ConnectionState.SUSPENDED) {
                         logger.warn("worker : {} connection SUSPENDED ", address);
-                        zookeeperRegistryCenter.getRegisterOperator().persistEphemeral(workerZKPath, "");
                     }
                 });
             logger.info("worker node : {} registry to ZK {} successfully", address, workerZKPath);

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/ZookeeperOperator.java

@@ -189,18 +189,20 @@ public class ZookeeperOperator implements InitializingBean {
         }
     }
 
-    public void persistEphemeral(final String key, final String value) {
+    public void persistEphemeral(final String path, final String value) {
         try {
-            if (isExisted(key)) {
+            // If the ephemeral node exist and the data is not equals to the given value
+            // delete the old node
+            if (isExisted(path) && !value.equals(get(path))) {
                 try {
-                    zkClient.delete().deletingChildrenIfNeeded().forPath(key);
+                    zkClient.delete().deletingChildrenIfNeeded().forPath(path);
                 } catch (NoNodeException ignore) {
                     //NOP
                 }
             }
-            zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key, value.getBytes(StandardCharsets.UTF_8));
+            zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, value.getBytes(StandardCharsets.UTF_8));
         } catch (final Exception ex) {
-            logger.error("persistEphemeral key : {} , value : {}", key, value, ex);
+            logger.error("persistEphemeral path : {} , value : {}", path, value, ex);
         }
     }