[Feature-3934] Add zookeeper registry for streampark

pom.xml

@@ -103,6 +103,7 @@
         <hadoop.version>3.3.4</hadoop.version>
         <hbase.version>2.1.10</hbase.version>
         <redis.version>3.3.0</redis.version>
+        <zoopkeeper.version>3.6.3</zoopkeeper.version>
         <es.version>6.2.3</es.version>
         <influxdb.version>2.17</influxdb.version>
         <protobuf.version>2.5.0</protobuf.version>
@@ -242,6 +243,12 @@
                 <scope>test</scope>
             </dependency>
 
+            <dependency>
+                <groupId>org.apache.zookeeper</groupId>
+                <artifactId>zookeeper</artifactId>
+                <version>${zoopkeeper.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>redis.clients</groupId>
                 <artifactId>jedis</artifactId>

streampark-console/streampark-console-service/src/main/assembly/conf/config.yaml

@@ -100,11 +100,13 @@ sso:
         # Optional, change by authentication client
         # Please replace and fill in your client config below when enabled SSO
 
-registry:
-  # default using jdbc as registry
-  type: jdbc
-  heartbeat-refresh-interval: 1s
-  session-timeout: 3s
+high-availability:
+  enable: false # true
+  # The list of ZooKeeper quorum peers that coordinate the high-availability
+  # setup. This must be a list of the form:
+  # "host1:clientPort,host2:clientPort,..." (default clientPort: 2181)
+  #
+  zookeeper.quorum: 192.168.100.128:2181,192.168.100.129:2181
 
 network:
   # network interface preferred like eth0, default: empty

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/StreamParkConsoleBootstrap.java

@@ -54,5 +54,4 @@ public static void main(String[] args) throws Exception {
             .sources(StreamParkConsoleBootstrap.class)
             .run(args);
     }
-
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java

@@ -26,9 +26,11 @@
 import org.apache.streampark.common.fs.FsOperator;
 import org.apache.streampark.common.util.AssertUtils;
 import org.apache.streampark.common.util.SystemPropertyUtils;
+import org.apache.streampark.console.base.util.SpringContextUtils;
 import org.apache.streampark.console.base.util.WebUtils;
 import org.apache.streampark.console.core.entity.FlinkEnv;
 import org.apache.streampark.console.core.entity.SparkEnv;
+import org.apache.streampark.console.core.service.RegistryService;
 import org.apache.streampark.console.core.service.SettingService;
 
 import org.apache.commons.lang3.StringUtils;
@@ -81,6 +83,9 @@ public void run(ApplicationArguments args) throws Exception {
         // init InternalConfig
         initConfig();
 
+        // init RegistryService
+        initRegistryService();
+
         boolean isTest = Arrays.asList(context.getEnvironment().getActiveProfiles()).contains("test");
         if (!isTest) {
             // initialize local file system resources
@@ -110,6 +115,18 @@ private void initConfig() {
         overrideSystemProp(ConfigKeys.KEY_HADOOP_USER_NAME(), hadoopUserName);
     }
 
+    private void initRegistryService() {
+        boolean enable = SystemPropertyUtils.get("high-availability.enable", "false").equals("true");
+        if (enable) {
+            RegistryService registryService = SpringContextUtils.getBean(RegistryService.class);
+            registryService.registry();
+            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+                registryService.unRegister();
+                log.info("RegistryService unRegister success");
+            }));
+        }
+    }
+
     private void overrideSystemProp(String key, String defaultValue) {
         String value = context.getEnvironment().getProperty(key, defaultValue);
         log.info("initialize system properties: key:{}, value:{}", key, value);

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/RegistryService.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.console.core.service;
+
+public interface RegistryService {
+
+    /**
+     * Registry the service.
+     */
+    void registry();
+
+    /**
+     * Close the registry service.
+     */
+    void unRegister();
+}

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/RegistryServiceImpl.java

@@ -0,0 +1,206 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.console.core.service.impl;
+
+import org.apache.streampark.common.util.SystemPropertyUtils;
+import org.apache.streampark.console.core.service.DistributedTaskService;
+import org.apache.streampark.console.core.service.RegistryService;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.net.InetAddress;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.zookeeper.ZooDefs.Ids.OPEN_ACL_UNSAFE;
+
+@Slf4j
+@Service
+public class RegistryServiceImpl implements RegistryService {
+
+    private static final String REGISTRY_PATH = "/services";
+    private static final int HEARTBEAT_INTERVAL = 10000;
+    private static final int HEARTBEAT_TIMEOUT = 60000;
+
+    private String zkAddress;
+    private ZooKeeper zk;
+    private String nodePath;
+    private Watcher watcher = event -> {
+        if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged
+            && event.getPath().equals(REGISTRY_PATH)) {
+            handleNodeChanges();
+        }
+    };
+
+    @Getter
+    private Set<String> currentNodes = new HashSet<>();
+
+    @Autowired
+    private DistributedTaskService distributedTaskService;
+
+    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
+
+    public void registry() {
+        try {
+            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);
+            }
+
+            String ip = InetAddress.getLocalHost().getHostAddress();
+            String port = SystemPropertyUtils.get("server.port", "10000");
+            String server_id = ip + ":" + port;
+            nodePath = zk.create(REGISTRY_PATH + "/" + server_id, new byte[0],
+                OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+
+            currentNodes.add(nodePath);
+
+            doRegister();
+        } catch (Exception e) {
+            log.error("Failed to init ZooKeeper client", e);
+        }
+    }
+
+    public void doRegister() {
+        try {
+            distributedTaskService.init(currentNodes, nodePath);
+            startHeartbeat();
+            startHeartbeatChecker();
+            handleNodeChanges();
+            log.info("ZooKeeper client started: {}", nodePath);
+        } catch (Exception e) {
+            log.error("Failed to start ZooKeeper client", e);
+        }
+    }
+
+    private void startHeartbeat() {
+        scheduler.scheduleAtFixedRate(() -> {
+            try {
+                zk.setData(nodePath, new byte[0], -1);
+                log.info("Heartbeat updated for node: {}", nodePath);
+            } catch (KeeperException e) {
+                log.info("Zookeeper session expired, attempting to reconnect...");
+                reconnectAndRegister();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.error("Failed to update heartbeat for node: {}", nodePath, e);
+            }
+        }, 0, HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS);
+    }
+
+    private void startHeartbeatChecker() {
+        scheduler.scheduleAtFixedRate(() -> {
+            try {
+                long now = System.currentTimeMillis();
+                List<String> servers = zk.getChildren(REGISTRY_PATH, false);
+                for (String server : servers) {
+                    String serverPath = REGISTRY_PATH + "/" + server;
+                    Stat stat = zk.exists(serverPath, false);
+                    if (stat != null && (now - stat.getMtime() > HEARTBEAT_TIMEOUT)) {
+                        zk.delete(serverPath, -1);
+                        log.info("Deleted stale node: {}", serverPath);
+                    }
+                }
+            } catch (KeeperException e) {
+                log.info("Zookeeper session expired, attempting to reconnect...");
+                reconnectAndRegister();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.error("Failed to check heartbeat", e);
+            }
+        }, HEARTBEAT_TIMEOUT, HEARTBEAT_TIMEOUT, TimeUnit.MILLISECONDS);
+    }
+
+    private synchronized void handleNodeChanges() {
+        try {
+            List<String> nodes = zk.getChildren(REGISTRY_PATH, true);
+            Set<String> newNodes = new HashSet<>(nodes);
+
+            for (String node : newNodes) {
+                if (!currentNodes.contains(node)) {
+                    log.info("Node added: {}", node);
+                    distributedTaskService.addServer(node);
+                }
+            }
+
+            for (String node : currentNodes) {
+                if (!newNodes.contains(node)) {
+                    log.info("Node removed: {}", node);
+                    distributedTaskService.removeServer(node);
+                }
+            }
+
+            currentNodes = newNodes;
+            log.info("Online servers: {}", currentNodes);
+        } catch (KeeperException e) {
+            log.info("Zookeeper session expired, attempting to reconnect...");
+            reconnectAndRegister();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.error("Failed to handle node changes", e);
+        }
+    }
+
+    private void reconnectAndRegister() {
+        int retries = 5;
+        while (retries > 0) {
+            try {
+                zk.close();
+                zk = new ZooKeeper(zkAddress, HEARTBEAT_TIMEOUT, watcher);
+                zk.create(nodePath, new byte[0], OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+                return;
+            } catch (Exception e) {
+                retries--;
+                log.warn("Retrying connection, attempts left: {}", retries, e);
+                try {
+                    Thread.sleep(3000);
+                } catch (InterruptedException ignored) {
+                    Thread.currentThread().interrupt();
+                }
+            }
+        }
+        log.error("Failed to reconnect and register node after multiple attempts.");
+    }
+
+    @Override
+    public void unRegister() {
+        try {
+            zk.close();
+            scheduler.shutdown();
+            log.info("ZooKeeper client closed: {}", nodePath);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.error("Failed to close ZooKeeper client", e);
+        }
+    }
+}

streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/RegistryServiceTest.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.console.core.service;
+
+import org.apache.streampark.common.util.SystemPropertyUtils;
+import org.apache.streampark.console.core.service.impl.RegistryServiceImpl;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class RegistryServiceTest {
+
+    private final RegistryServiceImpl registryService = new RegistryServiceImpl();
+
+    @Test
+    public void testRegister() {
+        if (enableHA()) {
+            try {
+                registryService.registry();
+            } catch (Exception e) {
+                Assertions.assertEquals(1, registryService.getCurrentNodes().size());
+                registryService.unRegister();
+            }
+        }
+    }
+
+    public boolean enableHA() {
+        return SystemPropertyUtils.get("high-availability.enable", "false").equals("true");
+    }
+
+}

tools/dependencies/known-dependencies.txt

@@ -377,5 +377,7 @@ xml-apis-1.4.01.jar
 xnio-api-3.8.7.Final.jar
 xnio-nio-3.8.7.Final.jar
 xz-1.5.jar
-zookeeper-3.4.14.jar
+zookeeper-3.6.3.jar
 icu4j-67.1.jar
+zookeeper-jute-3.6.3.jar
+netty-transport-native-epoll-4.1.91.Final.jar