fix commets

apache · Oct 30, 2024 · 07af799 · 07af799
1 parent a7beaa5
commit 07af799
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 31 deletions.
diff --git a/...nsole-service/src/main/java/org/apache/streampark/console/StreamParkConsoleBootstrap.java b/...nsole-service/src/main/java/org/apache/streampark/console/StreamParkConsoleBootstrap.java
@@ -66,9 +66,10 @@ public static void main(String[] args) throws Exception {
     @PostConstruct
     public void init() {
         if (enableHA()) {
-            registryService.startListening();
+            registryService.registry();
+            registryService.doRegister();
             Runtime.getRuntime().addShutdownHook(new Thread(() -> {
-                registryService.deRegistry();
+                registryService.unRegister();
                 log.info("RegistryService close success.");
             }));
         }

diff --git a/...ole-service/src/main/java/org/apache/streampark/console/core/service/RegistryService.java b/...ole-service/src/main/java/org/apache/streampark/console/core/service/RegistryService.java
@@ -19,13 +19,18 @@
 
 public interface RegistryService {
 
+    /**
+     * Registry the service.
+     */
+    void registry();
+
     /**
      * Start the registry service.
      */
-    void startListening();
+    void doRegister();
 
     /**
      * Close the registry service.
      */
-    void deRegistry();
+    void unRegister();
 }
diff --git a/...ce/src/main/java/org/apache/streampark/console/core/service/impl/RegistryServiceImpl.java b/...ce/src/main/java/org/apache/streampark/console/core/service/impl/RegistryServiceImpl.java
@@ -32,8 +32,6 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.PostConstruct;
-
 import java.net.InetAddress;
 import java.util.HashSet;
 import java.util.List;
@@ -52,7 +50,7 @@ public class RegistryServiceImpl implements RegistryService {
     private static final int HEARTBEAT_INTERVAL = 10000;
     private static final int HEARTBEAT_TIMEOUT = 60000;
 
-    private String zk_address;
+    private String zkAddress;
     private ZooKeeper zk;
     private String nodePath;
     private Watcher watcher = event -> {
@@ -70,14 +68,10 @@ public class RegistryServiceImpl implements RegistryService {
 
     private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
 
-    @PostConstruct
     public void registry() {
-        if (!enableHA()) {
-            return;
-        }
         try {
-            zk_address = SystemPropertyUtils.get("high-availability.zookeeper.quorum", "localhost:2181");
-            zk = new ZooKeeper(zk_address, HEARTBEAT_TIMEOUT, watcher);
+            zkAddress = SystemPropertyUtils.get("high-availability.zookeeper.quorum", "localhost:2181");
+            zk = new ZooKeeper(zkAddress, HEARTBEAT_TIMEOUT, watcher);
 
             if (zk.exists(REGISTRY_PATH, false) == null) {
                 zk.create(REGISTRY_PATH, new byte[0], OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
@@ -96,10 +90,7 @@ public void registry() {
     }
 
     @Override
-    public void startListening() {
-        if (!enableHA()) {
-            return;
-        }
+    public void doRegister() {
         try {
             distributedTaskService.init(currentNodes, nodePath);
             startHeartbeat();
@@ -183,7 +174,7 @@ private void reconnectAndRegister() {
         while (retries > 0) {
             try {
                 zk.close();
-                zk = new ZooKeeper(zk_address, HEARTBEAT_TIMEOUT, watcher);
+                zk = new ZooKeeper(zkAddress, HEARTBEAT_TIMEOUT, watcher);
                 zk.create(nodePath, new byte[0], OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                 return;
             } catch (Exception e) {
@@ -200,10 +191,7 @@ private void reconnectAndRegister() {
     }
 
     @Override
-    public void deRegistry() {
-        if (!enableHA()) {
-            return;
-        }
+    public void unRegister() {
         try {
             zk.close();
             scheduler.shutdown();
@@ -213,9 +201,4 @@ public void deRegistry() {
             log.error("Failed to close ZooKeeper client", e);
         }
     }
-
-    public boolean enableHA() {
-        return SystemPropertyUtils.get("high-availability.enable", "false").equals("true");
-    }
-
 }
diff --git a/...service/src/test/java/org/apache/streampark/console/core/service/RegistryServiceTest.java b/...service/src/test/java/org/apache/streampark/console/core/service/RegistryServiceTest.java
@@ -29,10 +29,15 @@ public class RegistryServiceTest {
 
     @Test
     public void testRegister() {
-        SystemPropertyUtils.set("high-availability.enable", "true");
-        registryService.registry();
-        Assertions.assertEquals(1, registryService.getCurrentNodes().size());
-        registryService.deRegistry();
+        if (enableHA()) {
+            registryService.registry();
+            Assertions.assertEquals(1, registryService.getCurrentNodes().size());
+            registryService.unRegister();
+        }
+    }
+
+    public boolean enableHA() {
+        return SystemPropertyUtils.get("high-availability.enable", "false").equals("true");
     }
 
 }