You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2024/10/27 16:10:42 UTC
(incubator-streampark) branch remove-registry created (now 16347c9cc)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a change to branch remove-registry
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
at 16347c9cc [Improve] remove registry module
This branch includes the following new commits:
new 16347c9cc [Improve] remove registry module
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(incubator-streampark) 01/01: [Improve] remove registry module
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch remove-registry
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
commit 16347c9cc3329528578e75bfcadab9ebb6dea3cf
Author: benjobs <be...@apache.org>
AuthorDate: Mon Oct 28 00:10:14 2024 +0800
[Improve] remove registry module
---
streampark-console/pom.xml | 1 -
.../streampark-console-registry/README.md | 37 --
.../streampark-console-registry/pom.xml | 92 -----
.../streampark-registry-api/pom.xml | 39 ---
.../streampark/registry/api/ConnectStrategy.java | 31 --
.../registry/api/ConnectStrategyProperties.java | 33 --
.../registry/api/ConnectionListener.java | 28 --
.../streampark/registry/api/ConnectionState.java | 28 --
.../org/apache/streampark/registry/api/Event.java | 131 -------
.../apache/streampark/registry/api/Registry.java | 112 ------
.../streampark/registry/api/RegistryClient.java | 214 ------------
.../registry/api/RegistryConfiguration.java | 33 --
.../streampark/registry/api/RegistryException.java | 29 --
.../streampark/registry/api/StrategyType.java | 25 --
.../streampark/registry/api/SubscribeListener.java | 23 --
.../registry/api/enums/RegistryNodeType.java | 35 --
.../registry/api/enums/ServerStatusEnum.java | 25 --
.../registry/api/lifecycle/ServerLifeCycle.java | 40 ---
.../api/lifecycle/ServerLifeCycleException.java | 29 --
.../api/lifecycle/ServerLifeCycleManager.java | 83 -----
.../registry/api/model/BaseHeartBeat.java | 43 ---
.../registry/api/model/BaseHeartBeatTask.java | 95 -----
.../registry/api/model/ConsoleHeartBeat.java | 27 --
.../streampark/registry/api/model/HeartBeat.java | 26 --
.../streampark/registry/api/model/Server.java | 46 ---
.../registry/api/sql/ClasspathSqlScriptParser.java | 147 --------
.../registry/api/sql/SqlScriptParser.java | 27 --
.../registry/api/sql/SqlScriptRunner.java | 65 ----
.../registry/api/thread/BaseDaemonThread.java | 38 --
.../thread/DefaultUncaughtExceptionHandler.java | 47 ---
.../registry/api/thread/ThreadUtils.java | 64 ----
.../streampark-registry-core/README.md | 83 -----
.../streampark-registry-core/pom.xml | 113 ------
.../streampark/registry/core/JdbcRegistry.java | 272 ---------------
.../core/JdbcRegistryAutoConfiguration.java | 120 -------
.../registry/core/JdbcRegistryProperties.java | 97 ------
.../registry/core/JdbcRegistryThreadFactory.java | 31 --
.../apache/streampark/registry/core/LockUtils.java | 34 --
.../registry/core/client/IJdbcRegistryClient.java | 106 ------
.../registry/core/client/JdbcRegistryClient.java | 129 -------
.../core/client/JdbcRegistryClientIdentify.java | 33 --
.../mapper/JdbcRegistryClientHeartbeatMapper.java | 33 --
.../mapper/JdbcRegistryDataChanceEventMapper.java | 41 ---
.../core/mapper/JdbcRegistryDataMapper.java | 51 ---
.../core/mapper/JdbcRegistryLockMapper.java | 39 ---
.../core/model/DO/JdbcRegistryClientHeartbeat.java | 46 ---
.../registry/core/model/DO/JdbcRegistryData.java | 48 ---
.../core/model/DO/JdbcRegistryDataChanceEvent.java | 48 ---
.../registry/core/model/DO/JdbcRegistryLock.java | 45 ---
.../registry/core/model/DTO/DataType.java | 25 --
.../model/DTO/JdbcRegistryClientHeartbeatDTO.java | 100 ------
.../model/DTO/JdbcRegistryDataChanceEventDTO.java | 78 -----
.../core/model/DTO/JdbcRegistryDataDTO.java | 69 ----
.../core/model/DTO/JdbcRegistryLockDTO.java | 63 ----
.../repository/JdbcRegistryClientRepository.java | 66 ----
.../JdbcRegistryDataChanceEventRepository.java | 65 ----
.../repository/JdbcRegistryDataRepository.java | 72 ----
.../repository/JdbcRegistryLockRepository.java | 55 ---
.../core/server/ConnectionStateListener.java | 27 --
.../core/server/IJdbcRegistryDataManager.java | 55 ---
.../core/server/IJdbcRegistryLockManager.java | 37 --
.../registry/core/server/IJdbcRegistryServer.java | 111 ------
.../core/server/IRegistryRowChangeNotifier.java | 35 --
.../server/JdbcRegistryDataChangeListener.java | 28 --
.../core/server/JdbcRegistryDataManager.java | 264 --------------
.../core/server/JdbcRegistryLockManager.java | 150 --------
.../registry/core/server/JdbcRegistryServer.java | 386 ---------------------
.../core/server/JdbcRegistryServerState.java | 46 ---
.../src/main/resources/META-INF/spring.factories | 19 -
.../src/main/resources/mysql_registry_init.sql | 69 ----
.../main/resources/postgresql_registry_init.sql | 65 ----
.../registry/core/JdbcRegistryTestCase.java | 64 ----
.../core/JdbcRegistryThreadFactoryTest.java | 43 ---
.../streampark/registry/core/LockUtilsTest.java | 38 --
.../registry/core/MysqlJdbcRegistryTestCase.java | 77 ----
.../core/PostgresqlJdbcRegistryTestCase.java | 73 ----
.../src/test/resources/application-mysql.yaml | 38 --
.../src/test/resources/application-postgresql.yaml | 36 --
.../src/test/resources/application.yaml | 24 --
.../src/test/resources/logback.xml | 21 --
.../streampark-registry-it/pom.xml | 62 ----
.../plugin/registry/RegistryTestCase.java | 260 --------------
.../console/StreamParkConsoleBootstrap.java | 52 +--
.../console/core/config/ConsoleConfig.java | 10 +-
.../core/registry/ConsoleConnectStrategy.java | 24 --
.../registry/ConsoleConnectionStateListener.java | 54 ---
.../core/registry/ConsoleRegistryClient.java | 203 -----------
.../core/registry/ConsoleRegistryDataListener.java | 66 ----
.../console/core/registry/ConsoleStopStrategy.java | 52 ---
.../console/core/task/ConsoleHeartBeatTask.java | 72 ----
90 files changed, 2 insertions(+), 6214 deletions(-)
diff --git a/streampark-console/pom.xml b/streampark-console/pom.xml
index 8fdb2a4fc..59488748e 100644
--- a/streampark-console/pom.xml
+++ b/streampark-console/pom.xml
@@ -31,7 +31,6 @@
<modules>
<module>streampark-console-service</module>
- <module>streampark-console-registry</module>
</modules>
<profiles>
diff --git a/streampark-console/streampark-console-registry/README.md b/streampark-console/streampark-console-registry/README.md
deleted file mode 100644
index 1545f5f41..000000000
--- a/streampark-console/streampark-console-registry/README.md
+++ /dev/null
@@ -1,37 +0,0 @@
-## Function
-
-SP use registry to do the below three things:
-
-1. Store the metadata of master/worker so that it can get notify when nodes up and down.
-2. Store the metadata of worker to do load balance.
-3. Acquire a global lock when do failover.
-
-So for SP, the registry need to notify the server when the server subscribe data have added/deleted/updated, support a way to create/release a global lock,
-delete the server's metadata when server down.
-
-## How to use
-
-At present, we have implements three registry: JDBC(Default, Support MySql and PostgreSQL),Zookeeper. If you
-want to use them, you should config it at resource/application.yaml. The configuration details
-can be viewed in the README of plugin under Module streampark-registry-plugins
-
-## Module
-
-### streampark-registry-all
-
-This module is used for exporting the implementation of registry.
-If you want to add new registry,you should add the dependency in the pom.xml
-
-### streampark-registry-api
-
-This module contains the relevant interfaces involved in the use of the registry.
-The following are several important interfaces
-1. Registry Interface: If you want to implement your own registry, you just need to implement this interface
-2. ConnectionListener Interface: This interface is responsible for the connection status between the client and the registry,
-The connection state can be viewed in ConnectionState.java
-3. SubscribeListener Interface: This interface is responsible for monitoring the state changes of child nodes under the specified prefix.
-Event content can be viewed in event.java
-
-### streampark-registry-plugins
-
-This module contains all registry implementations in SP
diff --git a/streampark-console/streampark-console-registry/pom.xml b/streampark-console/streampark-console-registry/pom.xml
deleted file mode 100644
index c29bb0154..000000000
--- a/streampark-console/streampark-console-registry/pom.xml
+++ /dev/null
@@ -1,92 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ 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.
- -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-console</artifactId>
- <version>2.2.0-SNAPSHOT</version>
- </parent>
- <artifactId>streampark-console-registry</artifactId>
- <packaging>pom</packaging>
- <name>StreamPark : Registry</name>
-
- <modules>
- <module>streampark-registry-api</module>
- <module>streampark-registry-core</module>
- <module>streampark-registry-it</module>
- </modules>
-
- <dependencies>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-context</artifactId>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-autoconfigure</artifactId>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>javax.annotation</groupId>
- <artifactId>javax.annotation-api</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.projectlombok</groupId>
- <artifactId>lombok</artifactId>
- </dependency>
-
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-collections4</artifactId>
- </dependency>
-
- <!-- test dependencies on TestContainers -->
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>testcontainers</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>mysql</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>postgresql</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
-
-</project>
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/pom.xml b/streampark-console/streampark-console-registry/streampark-registry-api/pom.xml
deleted file mode 100644
index e4de468e4..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/pom.xml
+++ /dev/null
@@ -1,39 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ Licensed to 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. Apache Software Foundation (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.
- -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-console-registry</artifactId>
- <version>2.2.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>streampark-registry-api</artifactId>
- <name>StreamPark : Console Registry API</name>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-common_${scala.binary.version}</artifactId>
- </dependency>
- </dependencies>
-
-</project>
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectStrategy.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectStrategy.java
deleted file mode 100644
index 7aeaac4bc..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectStrategy.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.registry.api;
-
-/**
- * This interface defined a method to be executed when the server disconnected from registry.
- */
-public interface ConnectStrategy {
-
- void disconnect();
-
- void reconnect();
-
- StrategyType getStrategyType();
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectStrategyProperties.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectStrategyProperties.java
deleted file mode 100644
index 30a384a78..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectStrategyProperties.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.registry.api;
-
-import lombok.Getter;
-import lombok.Setter;
-
-import java.time.Duration;
-
-@Getter
-@Setter
-public class ConnectStrategyProperties {
-
- private StrategyType strategy = StrategyType.STOP;
-
- private Duration maxWaitingTime = Duration.ofSeconds(0);
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectionListener.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectionListener.java
deleted file mode 100644
index ebdc56eb3..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectionListener.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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.registry.api;
-
-/**
- * when the connect state between client and registry center changed,
- * the {@code onUpdate} function is triggered
- */
-@FunctionalInterface
-public interface ConnectionListener {
-
- void onUpdate(ConnectionState newState);
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectionState.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectionState.java
deleted file mode 100644
index dfcdbc10f..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/ConnectionState.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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.registry.api;
-
-/**
- * Connection state between client and registry center(Etcd, MySql, Zookeeper)
- */
-public enum ConnectionState {
- CONNECTED,
- RECONNECTED,
- SUSPENDED,
- DISCONNECTED
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/Event.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/Event.java
deleted file mode 100644
index deb9d5f7b..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/Event.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * 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.registry.api;
-
-public class Event {
-
- // The prefix which is watched
- private String key;
- // The full path where the event was generated
- private String path;
- // The value corresponding to the path
- private String data;
- // The event type {ADD, REMOVE, UPDATE}
- private Type type;
-
- public Event(String key, String path, String data, Type type) {
- this.key = key;
- this.path = path;
- this.data = data;
- this.type = type;
- }
-
- public Event() {
- }
-
- public static EventBuilder builder() {
- return new EventBuilder();
- }
-
- public String key() {
- return this.key;
- }
-
- public String path() {
- return this.path;
- }
-
- public String data() {
- return this.data;
- }
-
- public Type type() {
- return this.type;
- }
-
- public Event key(String key) {
- this.key = key;
- return this;
- }
-
- public Event path(String path) {
- this.path = path;
- return this;
- }
-
- public Event data(String data) {
- this.data = data;
- return this;
- }
-
- public Event type(Type type) {
- this.type = type;
- return this;
- }
-
- public String toString() {
- return "Event(key=" + this.key() + ", path=" + this.path() + ", data=" + this.data() + ", type=" + this.type()
- + ")";
- }
-
- public enum Type {
- ADD,
- REMOVE,
- UPDATE
- }
-
- public static class EventBuilder {
-
- private String key;
- private String path;
- private String data;
- private Type type;
-
- EventBuilder() {
- }
-
- public EventBuilder key(String key) {
- this.key = key;
- return this;
- }
-
- public EventBuilder path(String path) {
- this.path = path;
- return this;
- }
-
- public EventBuilder data(String data) {
- this.data = data;
- return this;
- }
-
- public EventBuilder type(Type type) {
- this.type = type;
- return this;
- }
-
- public Event build() {
- return new Event(key, path, data, type);
- }
-
- public String toString() {
- return "Event.EventBuilder(key=" + this.key + ", path=" + this.path + ", data=" + this.data + ", type="
- + this.type + ")";
- }
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/Registry.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/Registry.java
deleted file mode 100644
index 2f349be1d..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/Registry.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * 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.registry.api;
-
-import lombok.NonNull;
-
-import java.io.Closeable;
-import java.time.Duration;
-import java.util.Collection;
-
-/**
- * The SPI interface for registry center, each registry plugin should implement this interface.
- */
-public interface Registry extends Closeable {
-
- /**
- * Start the registry, once started, the registry will connect to the registry center.
- */
- void start();
-
- /**
- * Whether the registry is connected
- *
- * @return true if connected, false otherwise.
- */
- boolean isConnected();
-
- /**
- * Connect to the registry, will wait in the given timeout
- *
- * @param timeout max timeout, if timeout <= 0 will wait indefinitely.
- * @throws RegistryException cannot connect in the given timeout
- */
- void connectUntilTimeout(@NonNull Duration timeout) throws RegistryException;
-
- /**
- * Subscribe the path, when the path has expose {@link Event}, the listener will be triggered.
- * <p>
- * The sub path will also be watched, if the sub path has event, the listener will be triggered.
- *
- * @param path the path to subscribe
- * @param listener the listener to be triggered
- */
- void subscribe(String path, SubscribeListener listener);
-
- /**
- * Add a connection listener to collection.
- */
- void addConnectionStateListener(ConnectionListener listener);
-
- /**
- * Get the value of the key, if key not exist will throw {@link RegistryException}
- */
- String get(String key) throws RegistryException;
-
- /**
- * Put the key-value pair into the registry
- *
- * @param key the key, cannot be null
- * @param value the value, cannot be null
- * @param deleteOnDisconnect if true, when the connection state is disconnected, the key will be deleted
- */
- void put(String key, String value, boolean deleteOnDisconnect);
-
- /**
- * Delete the key from the registry
- */
- void delete(String key);
-
- /**
- * Return the children of the key
- */
- Collection<String> children(String key);
-
- /**
- * Check if the key exists
- *
- * @param key the key to check
- * @return true if the key exists
- */
- boolean exists(String key);
-
- /**
- * Acquire the lock of the prefix {@param key}
- */
- boolean acquireLock(String key);
-
- /**
- * Acquire the lock of the prefix {@param key}, if acquire in the given timeout return true, else return false.
- */
- boolean acquireLock(String key, long timeout);
-
- /**
- * Release the lock of the prefix {@param key}
- */
- boolean releaseLock(String key);
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/RegistryClient.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/RegistryClient.java
deleted file mode 100644
index 0dbd96adc..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/RegistryClient.java
+++ /dev/null
@@ -1,214 +0,0 @@
-/*
- * 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.registry.api;
-
-import org.apache.streampark.common.IStoppable;
-import org.apache.streampark.common.constants.Constants;
-import org.apache.streampark.common.utils.JSONUtils;
-import org.apache.streampark.registry.api.enums.RegistryNodeType;
-import org.apache.streampark.registry.api.model.ConsoleHeartBeat;
-import org.apache.streampark.registry.api.model.Server;
-
-import org.apache.commons.lang3.StringUtils;
-
-import com.google.common.base.Strings;
-import lombok.NonNull;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-
-import java.io.IOException;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import static com.google.common.base.Preconditions.checkArgument;
-
-@Component
-@Slf4j
-public class RegistryClient {
-
- private IStoppable stoppable;
-
- private static final String EMPTY = "";
-
- private final Registry registry;
-
- public RegistryClient(Registry registry) {
- this.registry = registry;
- if (!registry.exists(RegistryNodeType.CONSOLE_SERVER.getRegistryPath())) {
- registry.put(RegistryNodeType.CONSOLE_SERVER.getRegistryPath(), EMPTY, false);
- }
- }
-
- public boolean isConnected() {
- return registry.isConnected();
-
- }
-
- public void connectUntilTimeout(@NonNull Duration duration) throws RegistryException {
- registry.connectUntilTimeout(duration);
- }
-
- public List<Server> getServerList(RegistryNodeType registryNodeType) {
- Map<String, String> serverMaps = getServerMaps(registryNodeType);
-
- List<Server> serverList = new ArrayList<>();
- for (Map.Entry<String, String> entry : serverMaps.entrySet()) {
- String serverPath = entry.getKey();
- String heartBeatJson = entry.getValue();
- if (StringUtils.isEmpty(heartBeatJson)) {
- log.error("The heartBeatJson is empty, serverPath: {}", serverPath);
- continue;
- }
- Server server = new Server();
- switch (registryNodeType) {
- case CONSOLE_SERVER:
- ConsoleHeartBeat consoleHeartBeat =
- JSONUtils.parseObject(heartBeatJson, ConsoleHeartBeat.class);
- server.setCreateTime(new Date(consoleHeartBeat.getStartupTime()));
- server.setLastHeartbeatTime(new Date(consoleHeartBeat.getReportTime()));
- server.setId(consoleHeartBeat.getProcessId());
- server.setHost(consoleHeartBeat.getHost());
- server.setPort(consoleHeartBeat.getPort());
- break;
- default:
- log.warn("unknown registry node type: {}", registryNodeType);
- }
-
- server.setResInfo(heartBeatJson);
- // todo: add host, port in heartBeat Info, so that we don't need to parse this again
- server.setZkDirectory(registryNodeType.getRegistryPath() + "/" + serverPath);
- serverList.add(server);
- }
- return serverList;
- }
-
- /**
- * Return server host:port -> value
- */
- public Map<String, String> getServerMaps(RegistryNodeType nodeType) {
- Map<String, String> serverMap = new HashMap<>();
- try {
- Collection<String> serverList = getServerNodes(nodeType);
- for (String server : serverList) {
- serverMap.putIfAbsent(server, get(nodeType.getRegistryPath() + Constants.SINGLE_SLASH + server));
- }
- } catch (Exception e) {
- log.error("get server list failed", e);
- }
-
- return serverMap;
- }
-
- public boolean checkNodeExists(String host, RegistryNodeType nodeType) {
- return getServerMaps(nodeType).keySet()
- .stream()
- .anyMatch(it -> it.contains(host));
- }
-
- public Collection<String> getConsoleNodesDirectly() {
- return getChildrenKeys(RegistryNodeType.CONSOLE_SERVER.getRegistryPath());
- }
-
- /**
- * get host ip:port, path format: parentPath/ip:port
- *
- * @param path path
- * @return host ip:port, string format: parentPath/ip:port
- */
- public String getHostByEventDataPath(String path) {
- checkArgument(!Strings.isNullOrEmpty(path), "path cannot be null or empty");
-
- final String[] pathArray = path.split(Constants.SINGLE_SLASH);
-
- checkArgument(pathArray.length >= 1, "cannot parse path: %s", path);
-
- return pathArray[pathArray.length - 1];
- }
-
- public void close() throws IOException {
- registry.close();
- }
-
- public void persistEphemeral(String key, String value) {
- registry.put(key, value, true);
- }
-
- public void remove(String key) {
- registry.delete(key);
- }
-
- public String get(String key) {
- return registry.get(key);
- }
-
- public void subscribe(String path, SubscribeListener listener) {
- registry.subscribe(path, listener);
- }
-
- public void addConnectionStateListener(ConnectionListener listener) {
- registry.addConnectionStateListener(listener);
- }
-
- public boolean exists(String key) {
- return registry.exists(key);
- }
-
- public boolean getLock(String key) {
- return registry.acquireLock(key);
- }
-
- public boolean releaseLock(String key) {
- return registry.releaseLock(key);
- }
-
- public void setStoppable(IStoppable stoppable) {
- this.stoppable = stoppable;
- }
-
- public IStoppable getStoppable() {
- return stoppable;
- }
-
- public boolean isConsolePath(String path) {
- return path != null
- && path.startsWith(RegistryNodeType.CONSOLE_SERVER.getRegistryPath() + Constants.SINGLE_SLASH);
- }
-
- public Collection<String> getChildrenKeys(final String key) {
- return registry.children(key);
- }
-
- public Set<String> getServerNodeSet(RegistryNodeType nodeType) {
- try {
- return new HashSet<>(getServerNodes(nodeType));
- } catch (Exception e) {
- throw new RegistryException("Failed to get server node: " + nodeType, e);
- }
- }
-
- private Collection<String> getServerNodes(RegistryNodeType nodeType) {
- return getChildrenKeys(nodeType.getRegistryPath());
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/RegistryConfiguration.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/RegistryConfiguration.java
deleted file mode 100644
index 629b148cc..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/RegistryConfiguration.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.registry.api;
-
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-@Configuration
-public class RegistryConfiguration {
-
- @Bean
- @ConditionalOnMissingBean
- public RegistryClient registryClient(Registry registry) {
- return new RegistryClient(registry);
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/RegistryException.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/RegistryException.java
deleted file mode 100644
index 8611f102c..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/RegistryException.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * 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.registry.api;
-
-public final class RegistryException extends RuntimeException {
-
- public RegistryException(String message, Throwable cause) {
- super(message, cause);
- }
-
- public RegistryException(String message) {
- super(message);
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/StrategyType.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/StrategyType.java
deleted file mode 100644
index 450a4a272..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/StrategyType.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.registry.api;
-
-public enum StrategyType {
-
- STOP,
- WAITING,
- ;
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/SubscribeListener.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/SubscribeListener.java
deleted file mode 100644
index 9c3ab224e..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/SubscribeListener.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * 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.registry.api;
-
-public interface SubscribeListener {
-
- void notify(Event event);
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/enums/RegistryNodeType.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/enums/RegistryNodeType.java
deleted file mode 100644
index 8d7086334..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/enums/RegistryNodeType.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.registry.api.enums;
-
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-
-@Getter
-@AllArgsConstructor
-public enum RegistryNodeType {
-
- ALL_SERVERS("nodes", "/nodes"),
- CONSOLE_SERVER("ConsoleServer", "/nodes/console-server"),
- CONSOLE_NODE_LOCK("ConsoleNodeLock", "/lock/console"),
- ;
-
- private final String name;
-
- private final String registryPath;
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/enums/ServerStatusEnum.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/enums/ServerStatusEnum.java
deleted file mode 100644
index def84ca4a..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/enums/ServerStatusEnum.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.registry.api.enums;
-
-public enum ServerStatusEnum {
-
- NORMAL,
- BUSY,
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/lifecycle/ServerLifeCycle.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/lifecycle/ServerLifeCycle.java
deleted file mode 100644
index 3ad4b0fc7..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/lifecycle/ServerLifeCycle.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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.registry.api.lifecycle;
-
-import lombok.Getter;
-
-/**
- * This enum is used to represent the server status, include master/worker.
- */
-@Getter
-public enum ServerLifeCycle {
-
- RUNNING(0, "The current server is running"),
- WAITING(1, "The current server is waiting, this means it cannot work"),
- STOPPED(2, "The current server is stopped"),
- ;
-
- private final int code;
- private final String desc;
-
- ServerLifeCycle(int code, String desc) {
- this.code = code;
- this.desc = desc;
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/lifecycle/ServerLifeCycleException.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/lifecycle/ServerLifeCycleException.java
deleted file mode 100644
index 8fcbf8279..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/lifecycle/ServerLifeCycleException.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * 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.registry.api.lifecycle;
-
-public class ServerLifeCycleException extends Exception {
-
- public ServerLifeCycleException(String message) {
- super(message);
- }
-
- public ServerLifeCycleException(String message, Throwable throwable) {
- super(message, throwable);
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/lifecycle/ServerLifeCycleManager.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/lifecycle/ServerLifeCycleManager.java
deleted file mode 100644
index d70133617..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/lifecycle/ServerLifeCycleManager.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.registry.api.lifecycle;
-
-import lombok.Getter;
-import lombok.experimental.UtilityClass;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-@UtilityClass
-public class ServerLifeCycleManager {
-
- @Getter
- private static volatile ServerLifeCycle serverLifeCycle = ServerLifeCycle.RUNNING;
-
- @Getter
- private static long serverStartupTime = System.currentTimeMillis();
-
- public static boolean isRunning() {
- return serverLifeCycle == ServerLifeCycle.RUNNING;
- }
-
- public static boolean isStopped() {
- return serverLifeCycle == ServerLifeCycle.STOPPED;
- }
-
- /**
- * Change the current server state to {@link ServerLifeCycle#WAITING}, only {@link ServerLifeCycle#RUNNING} can change to {@link ServerLifeCycle#WAITING}.
- *
- * @throws ServerLifeCycleException if change failed.
- */
- public static synchronized void toWaiting() throws ServerLifeCycleException {
- if (isStopped()) {
- throw new ServerLifeCycleException("The current server is already stopped, cannot change to waiting");
- }
-
- if (serverLifeCycle == ServerLifeCycle.WAITING) {
- log.warn("The current server is already at waiting status, cannot change to waiting");
- return;
- }
- serverLifeCycle = ServerLifeCycle.WAITING;
- }
-
- /**
- * Recover from {@link ServerLifeCycle#WAITING} to {@link ServerLifeCycle#RUNNING}.
- */
- public static synchronized void recoverFromWaiting() throws ServerLifeCycleException {
- if (isStopped()) {
- throw new ServerLifeCycleException("The current server is already stopped, cannot recovery");
- }
-
- if (serverLifeCycle == ServerLifeCycle.RUNNING) {
- log.warn("The current server status is already running, cannot recover form waiting");
- return;
- }
- serverStartupTime = System.currentTimeMillis();
- serverLifeCycle = ServerLifeCycle.RUNNING;
- }
-
- public static synchronized boolean toStopped() {
- if (serverLifeCycle == ServerLifeCycle.STOPPED) {
- return false;
- }
- serverLifeCycle = ServerLifeCycle.STOPPED;
- return true;
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/BaseHeartBeat.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/BaseHeartBeat.java
deleted file mode 100644
index ca9c98fab..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/BaseHeartBeat.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.registry.api.model;
-
-import org.apache.streampark.registry.api.enums.ServerStatusEnum;
-
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-import lombok.experimental.SuperBuilder;
-
-@Getter
-@Setter
-@SuperBuilder
-@NoArgsConstructor
-@AllArgsConstructor
-public class BaseHeartBeat implements HeartBeat {
-
- protected int processId;
- protected long startupTime;
- protected long reportTime;
- protected ServerStatusEnum serverStatusEnum;
-
- protected String host;
- protected int port;
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/BaseHeartBeatTask.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/BaseHeartBeatTask.java
deleted file mode 100644
index 0585bc83a..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/BaseHeartBeatTask.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.registry.api.model;
-
-import org.apache.streampark.registry.api.lifecycle.ServerLifeCycleManager;
-import org.apache.streampark.registry.api.thread.BaseDaemonThread;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public abstract class BaseHeartBeatTask<T extends HeartBeat> extends BaseDaemonThread {
-
- private static final long DEFAULT_HEARTBEAT_SCAN_INTERVAL = 1_000L;
-
- private final String threadName;
- private final long heartBeatInterval;
-
- protected boolean runningFlag;
-
- protected long lastWriteTime = 0L;
-
- protected T lastHeartBeat = null;
-
- public BaseHeartBeatTask(String threadName, long heartBeatInterval) {
- super(threadName);
- this.threadName = threadName;
- this.heartBeatInterval = heartBeatInterval;
- this.runningFlag = true;
- }
-
- @Override
- public synchronized void start() {
- log.info("Starting {}...", threadName);
- super.start();
- log.info("Started {}, heartBeatInterval: {}...", threadName, heartBeatInterval);
- }
-
- @Override
- public void run() {
- while (runningFlag) {
- try {
- if (!ServerLifeCycleManager.isRunning()) {
- log.info("The current server status is {}, will not write heartBeatInfo into registry",
- ServerLifeCycleManager.getServerLifeCycle());
- continue;
- }
- T heartBeat = getHeartBeat();
- // if first time or heartBeat status changed, write heartBeatInfo into registry
- if (System.currentTimeMillis() - lastWriteTime >= heartBeatInterval
- || !lastHeartBeat.getServerStatusEnum().equals(heartBeat.getServerStatusEnum())) {
- lastHeartBeat = heartBeat;
- writeHeartBeat(heartBeat);
- lastWriteTime = System.currentTimeMillis();
- }
- } catch (Exception ex) {
- log.error("{} task execute failed", threadName, ex);
- } finally {
- try {
- Thread.sleep(DEFAULT_HEARTBEAT_SCAN_INTERVAL);
- } catch (InterruptedException e) {
- handleInterruptException(e);
- }
- }
- }
- }
-
- public void shutdown() {
- runningFlag = false;
- log.warn("{} finished...", threadName);
- }
-
- private void handleInterruptException(InterruptedException ex) {
- log.warn("{} has been interrupted", threadName, ex);
- Thread.currentThread().interrupt();
- }
-
- public abstract T getHeartBeat();
-
- public abstract void writeHeartBeat(T heartBeat);
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/ConsoleHeartBeat.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/ConsoleHeartBeat.java
deleted file mode 100644
index 8f6a9529b..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/ConsoleHeartBeat.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * 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.registry.api.model;
-
-import lombok.NoArgsConstructor;
-import lombok.experimental.SuperBuilder;
-
-@SuperBuilder
-@NoArgsConstructor
-public class ConsoleHeartBeat extends BaseHeartBeat implements HeartBeat {
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/HeartBeat.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/HeartBeat.java
deleted file mode 100644
index 490ac12ed..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/HeartBeat.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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.registry.api.model;
-
-import org.apache.streampark.registry.api.enums.ServerStatusEnum;
-
-public interface HeartBeat {
-
- ServerStatusEnum getServerStatusEnum();
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/Server.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/Server.java
deleted file mode 100644
index 2571169fa..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/model/Server.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.registry.api.model;
-
-import lombok.Getter;
-import lombok.Setter;
-
-import java.util.Date;
-
-@Getter
-@Setter
-public class Server {
-
- private int id;
-
- private String host;
-
- private int port;
-
- private String zkDirectory;
-
- /**
- * resource info: CPU and memory
- */
- private String resInfo;
-
- private Date createTime;
-
- private Date lastHeartbeatTime;
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/sql/ClasspathSqlScriptParser.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/sql/ClasspathSqlScriptParser.java
deleted file mode 100644
index 3b414802d..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/sql/ClasspathSqlScriptParser.java
+++ /dev/null
@@ -1,147 +0,0 @@
-/*
- * 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.registry.api.sql;
-
-import org.apache.commons.lang3.StringUtils;
-
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.core.io.ClassPathResource;
-import org.springframework.core.io.Resource;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.io.LineNumberReader;
-import java.io.Reader;
-import java.nio.charset.Charset;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-@Slf4j
-public class ClasspathSqlScriptParser implements SqlScriptParser {
-
- private final String sqlScriptPath;
-
- private final Charset charset;
-
- public ClasspathSqlScriptParser(String sqlScriptPath) {
- this.sqlScriptPath = sqlScriptPath;
- this.charset = StandardCharsets.UTF_8;
- }
-
- @Override
- public List<String> getAllSql() throws IOException {
- Resource sqlScriptResource = new ClassPathResource(sqlScriptPath);
- if (!sqlScriptResource.exists()) {
- log.warn("The sql script file {} doesn't exist", sqlScriptPath);
- return Collections.emptyList();
- }
- List<String> result = new ArrayList<>();
- try (
- InputStream inputStream = sqlScriptResource.getInputStream();
- Reader sqlScriptReader = new InputStreamReader(inputStream, charset);
- LineNumberReader lineNumberReader = new LineNumberReader(sqlScriptReader)) {
- String sql;
- do {
- sql = parseNextSql(lineNumberReader);
- if (StringUtils.isNotBlank(sql)) {
- result.add(sql);
- }
- } while (StringUtils.isNotBlank(sql));
- }
- return result;
- }
-
- private String parseNextSql(LineNumberReader lineNumberReader) throws IOException {
- String line;
- while ((line = lineNumberReader.readLine()) != null) {
- String trimLine = line.trim();
- if (StringUtils.isEmpty(trimLine) || isComment(trimLine)) {
- // Skip the empty line, comment line
- continue;
- }
- if (trimLine.startsWith("/*")) {
- skipLicenseHeader(lineNumberReader);
- continue;
- }
- if (trimLine.startsWith("delimiter")) {
- // begin to parse processor, until delimiter ;
- String[] split = trimLine.split(" ");
- if (split[1].equals(";")) {
- continue;
- }
- return parseProcedure(lineNumberReader, split[1]);
- }
- // begin to parse sql until;
- List<String> sqlLines = new ArrayList<>();
- sqlLines.add(line);
- while (!line.endsWith(";")) {
- line = lineNumberReader.readLine();
- if (line == null) {
- break;
- }
- if (StringUtils.isBlank(line)) {
- continue;
- }
- sqlLines.add(line);
- }
- return String.join("\n", sqlLines);
- }
- return null;
- }
-
- private void skipLicenseHeader(LineNumberReader lineNumberReader) throws IOException {
- String line;
- while ((line = lineNumberReader.readLine()) != null) {
- String trimLine = line.trim();
- if (StringUtils.isEmpty(trimLine) || isComment(trimLine)) {
- // Skip the empty line, comment line
- continue;
- }
- if (line.startsWith("*/")) {
- break;
- }
- }
- }
-
- private String parseProcedure(LineNumberReader lineNumberReader, String delimiter) throws IOException {
- List<String> sqlLines = new ArrayList<>();
- // begin to parse processor, until delimiter ;
- String line;
- while (true) {
- line = lineNumberReader.readLine();
- if (line == null) {
- break;
- }
- if (StringUtils.isBlank(line)) {
- continue;
- }
- if (line.trim().startsWith(delimiter)) {
- break;
- }
- sqlLines.add(line);
- }
- return String.join("\n", sqlLines);
- }
-
- private boolean isComment(String line) {
- return line.startsWith("--") || line.startsWith("//");
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/sql/SqlScriptParser.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/sql/SqlScriptParser.java
deleted file mode 100644
index bcf43710f..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/sql/SqlScriptParser.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * 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.registry.api.sql;
-
-import java.io.IOException;
-import java.util.List;
-
-public interface SqlScriptParser {
-
- List<String> getAllSql() throws IOException;
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/sql/SqlScriptRunner.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/sql/SqlScriptRunner.java
deleted file mode 100644
index 450e4e428..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/sql/SqlScriptRunner.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.registry.api.sql;
-
-import org.apache.commons.lang3.StringUtils;
-
-import lombok.extern.slf4j.Slf4j;
-
-import javax.sql.DataSource;
-
-import java.io.IOException;
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.List;
-
-@Slf4j
-public class SqlScriptRunner {
-
- private final DataSource dataSource;
-
- private final SqlScriptParser sqlScriptParser;
-
- /**
- * @param dataSource DataSource which used to execute the sql script.
- * @param sqlScriptFilePath Sqk script file path, the path should under classpath.
- */
- public SqlScriptRunner(DataSource dataSource, String sqlScriptFilePath) {
- this.dataSource = dataSource;
- this.sqlScriptParser = new ClasspathSqlScriptParser(sqlScriptFilePath);
- }
-
- public void execute() throws SQLException, IOException {
- List<String> allSql = sqlScriptParser.getAllSql();
- try (Connection connection = dataSource.getConnection()) {
- for (String sql : allSql) {
- if (StringUtils.isBlank(sql)) {
- continue;
- }
- try (Statement statement = connection.createStatement()) {
- // Since some sql doesn't have result so we believe if there is no exception then we think the sql
- // execute success.
- statement.execute(sql);
- log.info("Execute sql: {} success", sql);
- }
- }
- }
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/thread/BaseDaemonThread.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/thread/BaseDaemonThread.java
deleted file mode 100644
index d156273a8..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/thread/BaseDaemonThread.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.registry.api.thread;
-
-/**
- * All thread used in StreamPark should extend with this class to avoid the server hang issue.
- */
-public abstract class BaseDaemonThread extends Thread {
-
- protected BaseDaemonThread(Runnable runnable) {
- super(runnable);
- this.setDaemon(true);
- this.setUncaughtExceptionHandler(DefaultUncaughtExceptionHandler.getInstance());
- }
-
- protected BaseDaemonThread(String threadName) {
- super();
- this.setName(threadName);
- this.setDaemon(true);
- this.setUncaughtExceptionHandler(DefaultUncaughtExceptionHandler.getInstance());
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/thread/DefaultUncaughtExceptionHandler.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/thread/DefaultUncaughtExceptionHandler.java
deleted file mode 100644
index 79df8d8ce..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/thread/DefaultUncaughtExceptionHandler.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.registry.api.thread;
-
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.concurrent.atomic.LongAdder;
-
-@Slf4j
-public class DefaultUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
-
- private static final DefaultUncaughtExceptionHandler INSTANCE = new DefaultUncaughtExceptionHandler();
-
- private static final LongAdder uncaughtExceptionCount = new LongAdder();
-
- private DefaultUncaughtExceptionHandler() {
- }
-
- public static DefaultUncaughtExceptionHandler getInstance() {
- return INSTANCE;
- }
-
- public static long getUncaughtExceptionCount() {
- return uncaughtExceptionCount.longValue();
- }
-
- @Override
- public void uncaughtException(Thread t, Throwable e) {
- uncaughtExceptionCount.add(1);
- log.error("Caught an exception in {}.", t, e);
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/thread/ThreadUtils.java b/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/thread/ThreadUtils.java
deleted file mode 100644
index 9e92fdb96..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-api/src/main/java/org/apache/streampark/registry/api/thread/ThreadUtils.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.registry.api.thread;
-
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import lombok.experimental.UtilityClass;
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-
-@UtilityClass
-@Slf4j
-public class ThreadUtils {
-
- public static ThreadPoolExecutor newDaemonFixedThreadExecutor(String threadName, int threadsNum) {
- return (ThreadPoolExecutor) Executors.newFixedThreadPool(threadsNum, newDaemonThreadFactory(threadName));
- }
-
- public static ScheduledExecutorService newSingleDaemonScheduledExecutorService(String threadName) {
- return Executors.newSingleThreadScheduledExecutor(newDaemonThreadFactory(threadName));
- }
-
- public static ScheduledExecutorService newDaemonScheduledExecutorService(String threadName, int threadsNum) {
- return Executors.newScheduledThreadPool(threadsNum, newDaemonThreadFactory(threadName));
- }
-
- public static ThreadFactory newDaemonThreadFactory(String threadName) {
- return new ThreadFactoryBuilder()
- .setDaemon(true)
- .setNameFormat(threadName)
- .setUncaughtExceptionHandler(DefaultUncaughtExceptionHandler.getInstance())
- .build();
- }
-
- /**
- * Sleep in given mills, this is not accuracy.
- */
- public static void sleep(final long millis) {
- try {
- Thread.sleep(millis);
- } catch (final InterruptedException interruptedException) {
- Thread.currentThread().interrupt();
- log.error("Current thread sleep error", interruptedException);
- }
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/README.md b/streampark-console/streampark-console-registry/streampark-registry-core/README.md
deleted file mode 100644
index a9bfc05b9..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/README.md
+++ /dev/null
@@ -1,83 +0,0 @@
-# Introduction
-
-This module is the jdbc registry plugin module, this plugin will use jdbc as the registry center. Will use the database
-configuration same as streampark in config.yaml default.
-
-# How to use
-
-1. Initialize the database table
-
-- If you use Mysql you can directly execute the sql script `src/main/resources/mysql_registry_init.sql`.
-
-- If you use Postgresql you can directly execute the sql script `src/main/resources/postgresql_registry_init.sql`.
-
-2. Change the config
-
-You need to set the registry properties in master/worker/api's application.yml
-
-```yaml
-registry:
- type: jdbc
-```
-
-After do this two steps, you can start your streampark cluster, your cluster will use mysql as registry center to
-store server metadata.
-
-NOTE: You need to add `mysql-connector-java.jar` into SP classpath if you use mysql database, since this plugin will not
-bundle this driver in distribution.
-
-## Optional configuration
-
-```yaml
-registry:
- type: jdbc
- # Used to schedule refresh the heartbeat.
- heartbeat-refresh-interval: 3s
- # Once the client's heartbeat is not refresh in this time, the server will consider the client is offline.
- session-timeout: 60s
- # The hikari configuration, default will use the same datasource pool as streampark.
- hikari-config:
- jdbc-url: jdbc:mysql://127.0.0.1:3306/streampark
- username: root
- password: root
- maximum-pool-size: 5
- connection-timeout: 9000
- idle-timeout: 600000
-```
-
-## Use different database configuration for jdbc registry center
-
-You need to set the registry properties in console config.yaml
-
-### Use MySql as registry center
-
-```yaml
-registry:
- type: jdbc
- heartbeat-refresh-interval: 3s
- session-timeout: 60s
- hikari-config:
- jdbc-url: jdbc:mysql://127.0.0.1:3306/streampark
- username: root
- password: root
- maximum-pool-size: 5
- connection-timeout: 9000
- idle-timeout: 600000
-```
-
-### Use PostgreSQL as registry center
-
-```yaml
-registry:
- type: jdbc
- heartbeat-refresh-interval: 3s
- session-timeout: 60s
- hikari-config:
- jdbc-url: jdbc:postgresql://localhost:5432/streampark
- username: root
- password: root
- maximum-pool-size: 5
- connection-timeout: 9000
- idle-timeout: 600000
-```
-
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/pom.xml b/streampark-console/streampark-console-registry/streampark-registry-core/pom.xml
deleted file mode 100644
index a4877f699..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/pom.xml
+++ /dev/null
@@ -1,113 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ Licensed to 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. Apache Software Foundation (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.
- -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-console-registry</artifactId>
- <version>2.2.0-SNAPSHOT</version>
- </parent>
- <artifactId>streampark-registry-core</artifactId>
- <name>StreamPark : Console Registry Core</name>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-registry-api</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-common_${scala.binary.version}</artifactId>
- </dependency>
- <dependency>
- <groupId>com.zaxxer</groupId>
- <artifactId>HikariCP</artifactId>
- </dependency>
-
- <dependency>
- <groupId>mysql</groupId>
- <artifactId>mysql-connector-java</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.postgresql</groupId>
- <artifactId>postgresql</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </dependency>
-
- <dependency>
- <groupId>com.baomidou</groupId>
- <artifactId>mybatis-plus</artifactId>
- </dependency>
-
- <dependency>
- <groupId>com.baomidou</groupId>
- <artifactId>mybatis-plus-boot-starter</artifactId>
- <exclusions>
- <exclusion>
- <groupId>org.apache.logging.log4j</groupId>
- <artifactId>log4j-to-slf4j</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.hibernate.validator</groupId>
- <artifactId>hibernate-validator</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-registry-it</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>mysql</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>postgresql</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.apache.httpcomponents.client5</groupId>
- <artifactId>httpclient5</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-test</artifactId>
- <scope>test</scope>
- </dependency>
-
- </dependencies>
-
-</project>
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistry.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistry.java
deleted file mode 100644
index 58f5f972e..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistry.java
+++ /dev/null
@@ -1,272 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.apache.streampark.registry.api.ConnectionListener;
-import org.apache.streampark.registry.api.ConnectionState;
-import org.apache.streampark.registry.api.Event;
-import org.apache.streampark.registry.api.Registry;
-import org.apache.streampark.registry.api.RegistryException;
-import org.apache.streampark.registry.api.SubscribeListener;
-import org.apache.streampark.registry.core.client.JdbcRegistryClient;
-import org.apache.streampark.registry.core.model.DTO.DataType;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataDTO;
-import org.apache.streampark.registry.core.server.ConnectionStateListener;
-import org.apache.streampark.registry.core.server.IJdbcRegistryServer;
-import org.apache.streampark.registry.core.server.JdbcRegistryDataChangeListener;
-
-import org.apache.commons.lang3.StringUtils;
-
-import lombok.NonNull;
-import lombok.extern.slf4j.Slf4j;
-
-import java.time.Duration;
-import java.util.Collection;
-import java.util.List;
-import java.util.Optional;
-import java.util.stream.Collectors;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-/**
- * This is one of the implementation of {@link Registry}, with this implementation, you need to rely on mysql database to
- * store the streampark master/worker's metadata and do the server registry/unRegistry.
- */
-@Slf4j
-public final class JdbcRegistry implements Registry {
-
- private final JdbcRegistryProperties jdbcRegistryProperties;
- private final JdbcRegistryClient jdbcRegistryClient;
-
- private final IJdbcRegistryServer jdbcRegistryServer;
-
- JdbcRegistry(JdbcRegistryProperties jdbcRegistryProperties, IJdbcRegistryServer jdbcRegistryServer) {
- this.jdbcRegistryProperties = jdbcRegistryProperties;
- this.jdbcRegistryServer = jdbcRegistryServer;
- this.jdbcRegistryClient = new JdbcRegistryClient(jdbcRegistryProperties, jdbcRegistryServer);
- log.info("Initialize Jdbc Registry...");
- }
-
- @Override
- public void start() {
- log.info("Starting Jdbc Registry...");
- jdbcRegistryServer.start();
- jdbcRegistryClient.start();
- log.info("Started Jdbc Registry...");
- }
-
- @Override
- public boolean isConnected() {
- return jdbcRegistryClient.isConnectivity();
- }
-
- @Override
- public void connectUntilTimeout(@NonNull Duration timeout) throws RegistryException {
- long beginTimeMillis = System.currentTimeMillis();
- long endTimeMills = timeout.getSeconds() <= 0 ? Long.MAX_VALUE : beginTimeMillis + timeout.toMillis();
- while (true) {
- if (System.currentTimeMillis() > endTimeMills) {
- throw new RegistryException(
- String.format("Cannot connect to jdbc registry in %s s", timeout.getSeconds()));
- }
- if (jdbcRegistryClient.isConnectivity()) {
- return;
- }
- try {
- Thread.sleep(jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis());
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RegistryException("Cannot connect to jdbc registry due to interrupted exception", e);
- }
- }
- }
-
- @Override
- public void subscribe(String path, SubscribeListener listener) {
- checkNotNull(path);
- checkNotNull(listener);
- jdbcRegistryClient.subscribeJdbcRegistryDataChange(new JdbcRegistryDataChangeListener() {
-
- @Override
- public void onJdbcRegistryDataChanged(String key, String value) {
- if (!key.startsWith(path)) {
- return;
- }
- Event event = Event.builder()
- .key(key)
- .path(path)
- .data(value)
- .type(Event.Type.UPDATE)
- .build();
- listener.notify(event);
- }
-
- @Override
- public void onJdbcRegistryDataDeleted(String key) {
- if (!key.startsWith(path)) {
- return;
- }
- Event event = Event.builder()
- .key(key)
- .path(key)
- .type(Event.Type.REMOVE)
- .build();
- listener.notify(event);
- }
-
- @Override
- public void onJdbcRegistryDataAdded(String key, String value) {
- if (!key.startsWith(path)) {
- return;
- }
- Event event = Event.builder()
- .key(key)
- .path(key)
- .data(value)
- .type(Event.Type.ADD)
- .build();
- listener.notify(event);
- }
- });
- }
-
- @Override
- public void addConnectionStateListener(ConnectionListener listener) {
- checkNotNull(listener);
- jdbcRegistryClient.subscribeConnectionStateChange(new ConnectionStateListener() {
-
- @Override
- public void onConnected() {
- listener.onUpdate(ConnectionState.CONNECTED);
- }
-
- @Override
- public void onDisConnected() {
- listener.onUpdate(ConnectionState.DISCONNECTED);
- }
-
- @Override
- public void onReconnected() {
- listener.onUpdate(ConnectionState.RECONNECTED);
- }
- });
- }
-
- @Override
- public String get(String key) {
- try {
- // get the key value
- // Directly get from the db?
- Optional<JdbcRegistryDataDTO> jdbcRegistryDataOptional = jdbcRegistryClient.getJdbcRegistryDataByKey(key);
- if (!jdbcRegistryDataOptional.isPresent()) {
- throw new RegistryException("key: " + key + " not exist");
- }
- return jdbcRegistryDataOptional.get().getDataValue();
- } catch (RegistryException registryException) {
- throw registryException;
- } catch (Exception e) {
- throw new RegistryException(String.format("Get key: %s error", key), e);
- }
- }
-
- @Override
- public void put(String key, String value, boolean deleteOnDisconnect) {
- try {
- DataType dataType = deleteOnDisconnect ? DataType.EPHEMERAL : DataType.PERSISTENT;
- jdbcRegistryClient.putJdbcRegistryData(key, value, dataType);
- } catch (Exception ex) {
- throw new RegistryException(String.format("put key:%s, value:%s error", key, value), ex);
- }
- }
-
- @Override
- public void delete(String key) {
- try {
- jdbcRegistryClient.deleteJdbcRegistryDataByKey(key);
- } catch (Exception e) {
- throw new RegistryException(String.format("Delete key: %s error", key), e);
- }
- }
-
- @Override
- public Collection<String> children(String key) {
- try {
- List<JdbcRegistryDataDTO> children = jdbcRegistryClient.listJdbcRegistryDataChildren(key);
- return children
- .stream()
- .map(JdbcRegistryDataDTO::getDataKey)
- .filter(fullPath -> fullPath.length() > key.length())
- .map(fullPath -> StringUtils.substringBefore(fullPath.substring(key.length() + 1), "/"))
- .distinct()
- .collect(Collectors.toList());
- } catch (Exception e) {
- throw new RegistryException(String.format("Get key: %s children error", key), e);
- }
- }
-
- @Override
- public boolean exists(String key) {
- try {
- return jdbcRegistryClient.existJdbcRegistryDataKey(key);
- } catch (Exception e) {
- throw new RegistryException(String.format("Check key: %s exist error", key), e);
- }
- }
-
- @Override
- public boolean acquireLock(String key) {
- try {
- jdbcRegistryClient.acquireJdbcRegistryLock(key);
- return true;
- } catch (RegistryException e) {
- throw e;
- } catch (Exception e) {
- throw new RegistryException(String.format("Acquire lock: %s error", key), e);
- }
- }
-
- @Override
- public boolean acquireLock(String key, long timeout) {
- try {
- return jdbcRegistryClient.acquireJdbcRegistryLock(key, timeout);
- } catch (RegistryException e) {
- throw e;
- } catch (Exception e) {
- throw new RegistryException(String.format("Acquire lock: %s error", key), e);
- }
- }
-
- @Override
- public boolean releaseLock(String key) {
- jdbcRegistryClient.releaseJdbcRegistryLock(key);
- return true;
- }
-
- @Override
- public void close() {
- log.info("Closing Jdbc Registry...");
- // remove the current Ephemeral node, if can connect to jdbc
- try {
- jdbcRegistryServer.close();
- jdbcRegistryClient.close();
- } catch (Exception e) {
- log.error("Close Jdbc Registry error", e);
- }
- log.info("Closed Jdbc Registry...");
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistryAutoConfiguration.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistryAutoConfiguration.java
deleted file mode 100644
index bb6799055..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistryAutoConfiguration.java
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.apache.streampark.registry.core.mapper.JdbcRegistryClientHeartbeatMapper;
-import org.apache.streampark.registry.core.mapper.JdbcRegistryDataChanceEventMapper;
-import org.apache.streampark.registry.core.mapper.JdbcRegistryDataMapper;
-import org.apache.streampark.registry.core.mapper.JdbcRegistryLockMapper;
-import org.apache.streampark.registry.core.repository.JdbcRegistryClientRepository;
-import org.apache.streampark.registry.core.repository.JdbcRegistryDataChanceEventRepository;
-import org.apache.streampark.registry.core.repository.JdbcRegistryDataRepository;
-import org.apache.streampark.registry.core.repository.JdbcRegistryLockRepository;
-import org.apache.streampark.registry.core.server.IJdbcRegistryServer;
-import org.apache.streampark.registry.core.server.JdbcRegistryServer;
-
-import org.apache.ibatis.session.SqlSessionFactory;
-
-import com.baomidou.mybatisplus.autoconfigure.MybatisPlusAutoConfiguration;
-import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
-import com.zaxxer.hikari.HikariDataSource;
-import lombok.extern.slf4j.Slf4j;
-import org.mybatis.spring.SqlSessionTemplate;
-import org.mybatis.spring.annotation.MapperScan;
-import org.springframework.boot.autoconfigure.AutoConfigureAfter;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.ComponentScan;
-import org.springframework.context.annotation.Configuration;
-
-@Slf4j
-@ComponentScan
-@Configuration(proxyBeanMethods = false)
-@MapperScan("org.apache.streampark.registry.core.mapper")
-@ConditionalOnProperty(prefix = "registry", name = "type", havingValue = "jdbc")
-@AutoConfigureAfter(MybatisPlusAutoConfiguration.class)
-public class JdbcRegistryAutoConfiguration {
-
- public JdbcRegistryAutoConfiguration() {
- log.info("Load JdbcRegistryAutoConfiguration");
- }
-
- @Bean
- public IJdbcRegistryServer jdbcRegistryServer(JdbcRegistryDataRepository jdbcRegistryDataRepository,
- JdbcRegistryLockRepository jdbcRegistryLockRepository,
- JdbcRegistryClientRepository jdbcRegistryClientRepository,
- JdbcRegistryDataChanceEventRepository jdbcRegistryDataChanceEventRepository,
- JdbcRegistryProperties jdbcRegistryProperties) {
- return new JdbcRegistryServer(
- jdbcRegistryDataRepository,
- jdbcRegistryLockRepository,
- jdbcRegistryClientRepository,
- jdbcRegistryDataChanceEventRepository,
- jdbcRegistryProperties);
- }
-
- @Bean
- public JdbcRegistry jdbcRegistry(JdbcRegistryProperties jdbcRegistryProperties,
- IJdbcRegistryServer jdbcRegistryServer) {
- JdbcRegistry jdbcRegistry = new JdbcRegistry(jdbcRegistryProperties, jdbcRegistryServer);
- jdbcRegistry.start();
- return jdbcRegistry;
- }
-
- @Bean
- @ConditionalOnMissingBean
- public SqlSessionFactory sqlSessionFactory(JdbcRegistryProperties jdbcRegistryProperties) throws Exception {
- log.info("Initialize jdbcRegistrySqlSessionFactory");
- MybatisSqlSessionFactoryBean sqlSessionFactoryBean = new MybatisSqlSessionFactoryBean();
- sqlSessionFactoryBean.setDataSource(new HikariDataSource(jdbcRegistryProperties.getHikariConfig()));
- return sqlSessionFactoryBean.getObject();
- }
-
- @Bean
- @ConditionalOnMissingBean
- public SqlSessionTemplate sqlSessionTemplate(SqlSessionFactory jdbcRegistrySqlSessionFactory) {
- log.info("Initialize jdbcRegistrySqlSessionTemplate");
- return new SqlSessionTemplate(jdbcRegistrySqlSessionFactory);
- }
-
- @Bean
- public JdbcRegistryDataMapper jdbcRegistryDataMapper(SqlSessionTemplate jdbcRegistrySqlSessionTemplate) {
- jdbcRegistrySqlSessionTemplate.getConfiguration().addMapper(JdbcRegistryDataMapper.class);
- return jdbcRegistrySqlSessionTemplate.getMapper(JdbcRegistryDataMapper.class);
- }
-
- @Bean
- public JdbcRegistryLockMapper jdbcRegistryLockMapper(SqlSessionTemplate jdbcRegistrySqlSessionTemplate) {
- jdbcRegistrySqlSessionTemplate.getConfiguration().addMapper(JdbcRegistryLockMapper.class);
- return jdbcRegistrySqlSessionTemplate.getMapper(JdbcRegistryLockMapper.class);
- }
-
- @Bean
- public JdbcRegistryDataChanceEventMapper jdbcRegistryDataChanceEventMapper(SqlSessionTemplate jdbcRegistrySqlSessionTemplate) {
- jdbcRegistrySqlSessionTemplate.getConfiguration().addMapper(JdbcRegistryDataChanceEventMapper.class);
- return jdbcRegistrySqlSessionTemplate.getMapper(JdbcRegistryDataChanceEventMapper.class);
- }
-
- @Bean
- public JdbcRegistryClientHeartbeatMapper jdbcRegistryClientHeartbeatMapper(SqlSessionTemplate jdbcRegistrySqlSessionTemplate) {
- jdbcRegistrySqlSessionTemplate.getConfiguration().addMapper(JdbcRegistryClientHeartbeatMapper.class);
- return jdbcRegistrySqlSessionTemplate.getMapper(JdbcRegistryClientHeartbeatMapper.class);
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistryProperties.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistryProperties.java
deleted file mode 100644
index 16bd4de88..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistryProperties.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.apache.streampark.common.utils.NetworkUtils;
-
-import org.apache.commons.lang3.StringUtils;
-
-import com.zaxxer.hikari.HikariConfig;
-import lombok.Getter;
-import lombok.Setter;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.validation.Errors;
-import org.springframework.validation.Validator;
-import org.springframework.validation.annotation.Validated;
-
-import java.time.Duration;
-
-@Getter
-@Setter
-@Slf4j
-@Validated
-@Configuration
-@ConditionalOnProperty(prefix = "registry", name = "type", havingValue = "jdbc")
-@ConfigurationProperties(prefix = "registry")
-public class JdbcRegistryProperties implements Validator {
-
- @Autowired
- private NetworkUtils networkUtils;
-
- private static final Duration MIN_HEARTBEAT_REFRESH_INTERVAL = Duration.ofSeconds(1);
-
- @Value("${server.port:10000}")
- private int serverPort;
-
- private String jdbcRegistryClientName;
-
- private Duration heartbeatRefreshInterval = Duration.ofSeconds(3);
- private Duration sessionTimeout = Duration.ofSeconds(60);
- private HikariConfig hikariConfig;
-
- @Override
- public boolean supports(Class<?> clazz) {
- return JdbcRegistryProperties.class.isAssignableFrom(clazz);
- }
-
- @Override
- public void validate(Object target, Errors errors) {
- JdbcRegistryProperties jdbcRegistryProperties = (JdbcRegistryProperties) target;
- if (jdbcRegistryProperties.getHeartbeatRefreshInterval().compareTo(MIN_HEARTBEAT_REFRESH_INTERVAL) < 0) {
- errors.rejectValue("heartbeatRefreshInterval", "heartbeatRefreshInterval",
- "heartbeatRefreshInterval must be greater than 1s");
- }
-
- if (jdbcRegistryProperties.getSessionTimeout().toMillis() < 3
- * jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis()) {
- errors.rejectValue("sessionTimeout", "sessionTimeout",
- "sessionTimeout must be greater than 3 * heartbeatRefreshInterval");
- }
- if (StringUtils.isEmpty(jdbcRegistryClientName)) {
- jdbcRegistryClientName = NetworkUtils.getHost() + ":" + serverPort;
- }
-
- print();
- }
-
- private void print() {
- String config =
- "\n****************************JdbcRegistryProperties**************************************" +
- "\n jdbcRegistryClientName -> " + jdbcRegistryClientName +
- "\n heartbeatRefreshInterval -> " + heartbeatRefreshInterval +
- "\n sessionTimeout -> " + sessionTimeout +
- "\n hikariConfig -> " + hikariConfig +
- "\n****************************JdbcRegistryProperties**************************************";
- log.info(config);
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistryThreadFactory.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistryThreadFactory.java
deleted file mode 100644
index 8b004c640..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/JdbcRegistryThreadFactory.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.apache.streampark.registry.api.thread.ThreadUtils;
-
-import java.util.concurrent.ScheduledExecutorService;
-
-public class JdbcRegistryThreadFactory {
-
- public static ScheduledExecutorService getDefaultSchedulerThreadExecutor() {
- return ThreadUtils.newDaemonScheduledExecutorService("jdbc-registry-default-scheduler-thread-pool",
- Runtime.getRuntime().availableProcessors());
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/LockUtils.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/LockUtils.java
deleted file mode 100644
index 0f4f86c18..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/LockUtils.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.apache.streampark.common.utils.NetworkUtils;
-import org.apache.streampark.common.utils.OSUtils;
-
-import lombok.experimental.UtilityClass;
-
-@UtilityClass
-public class LockUtils {
-
- private static final String LOCK_OWNER_PREFIX = NetworkUtils.getHost() + "_" + OSUtils.getProcessID() + "_";
-
- public static String getLockOwner() {
- return LOCK_OWNER_PREFIX + Thread.currentThread().getName();
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/client/IJdbcRegistryClient.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/client/IJdbcRegistryClient.java
deleted file mode 100644
index b6f487090..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/client/IJdbcRegistryClient.java
+++ /dev/null
@@ -1,106 +0,0 @@
-/*
- * 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.registry.core.client;
-
-import org.apache.streampark.registry.core.model.DTO.DataType;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataDTO;
-import org.apache.streampark.registry.core.server.ConnectionStateListener;
-import org.apache.streampark.registry.core.server.JdbcRegistryDataChangeListener;
-
-import java.util.List;
-import java.util.Optional;
-
-public interface IJdbcRegistryClient extends AutoCloseable {
-
- /**
- * Start the jdbc registry client, once started, the client will connect to the jdbc registry server, and then it can be used.
- */
- void start();
-
- /**
- * Get identify of the client.
- */
- JdbcRegistryClientIdentify getJdbcRegistryClientIdentify();
-
- /**
- * Check the connectivity of the client.
- */
- boolean isConnectivity();
-
- /**
- * Subscribe the jdbc registry connection state change event.
- */
- void subscribeConnectionStateChange(ConnectionStateListener connectionStateListener);
-
- /**
- * Subscribe the {@link JdbcRegistryDataDTO} change event.
- */
- void subscribeJdbcRegistryDataChange(JdbcRegistryDataChangeListener jdbcRegistryDataChangeListener);
-
- /**
- * Get the {@link JdbcRegistryDataDTO} by key.
- */
- Optional<JdbcRegistryDataDTO> getJdbcRegistryDataByKey(String key);
-
- /**
- * Put the {@link JdbcRegistryDataDTO} to the jdbc registry server.
- * <p>
- * If the key is already exist, then update the {@link JdbcRegistryDataDTO}. If the key is not exist, then insert a new {@link JdbcRegistryDataDTO}.
- */
- void putJdbcRegistryData(String key, String value, DataType dataType);
-
- /**
- * Delete the {@link JdbcRegistryDataDTO} by key.
- */
- void deleteJdbcRegistryDataByKey(String key);
-
- /**
- * List all the {@link JdbcRegistryDataDTO} children by key.
- * <p>
- * e.g. key = "/streampark/master", and data exist in db is "/streampark/master/master1", "/streampark/master/master2"
- * <p>
- * then the return value will be ["master1", "master2"]
- */
- List<JdbcRegistryDataDTO> listJdbcRegistryDataChildren(String key);
-
- /**
- * Check the key exist in the jdbc registry server.
- */
- boolean existJdbcRegistryDataKey(String key);
-
- /**
- * Acquire the jdbc registry lock by key. this is a blocking method. if you want to stop the blocking, you can use interrupt the thread.
- */
- void acquireJdbcRegistryLock(String lockKey) throws IllegalArgumentException;
-
- /**
- * Acquire the jdbc registry lock by key until timeout.
- */
- boolean acquireJdbcRegistryLock(String lockKey, long timeout);
-
- /**
- * Release the jdbc registry lock by key, if the lockKey is not exist will do nothing.
- */
- void releaseJdbcRegistryLock(String lockKey);
-
- /**
- * Close the jdbc registry client, once the client been closed, it cannot work anymore.
- */
- @Override
- void close();
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/client/JdbcRegistryClient.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/client/JdbcRegistryClient.java
deleted file mode 100644
index 2775953e7..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/client/JdbcRegistryClient.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * 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.registry.core.client;
-
-import org.apache.streampark.common.utils.NetworkUtils;
-import org.apache.streampark.common.utils.OSUtils;
-import org.apache.streampark.common.utils.UUIDUtils;
-import org.apache.streampark.registry.core.JdbcRegistryProperties;
-import org.apache.streampark.registry.core.model.DTO.DataType;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataDTO;
-import org.apache.streampark.registry.core.server.ConnectionStateListener;
-import org.apache.streampark.registry.core.server.IJdbcRegistryServer;
-import org.apache.streampark.registry.core.server.JdbcRegistryDataChangeListener;
-import org.apache.streampark.registry.core.server.JdbcRegistryServer;
-import org.apache.streampark.registry.core.server.JdbcRegistryServerState;
-
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.List;
-import java.util.Optional;
-
-/**
- * The client of jdbc registry, used to interact with the {@link JdbcRegistryServer}.
- */
-@Slf4j
-public class JdbcRegistryClient implements IJdbcRegistryClient {
-
- private static final String DEFAULT_CLIENT_NAME = NetworkUtils.getHost() + "_" + OSUtils.getProcessID();
-
- private final JdbcRegistryProperties jdbcRegistryProperties;
-
- private final JdbcRegistryClientIdentify jdbcRegistryClientIdentify;
-
- private final IJdbcRegistryServer jdbcRegistryServer;
-
- public JdbcRegistryClient(JdbcRegistryProperties jdbcRegistryProperties, IJdbcRegistryServer jdbcRegistryServer) {
- this.jdbcRegistryProperties = jdbcRegistryProperties;
- this.jdbcRegistryServer = jdbcRegistryServer;
- this.jdbcRegistryClientIdentify =
- new JdbcRegistryClientIdentify(UUIDUtils.generateUUID(), DEFAULT_CLIENT_NAME);
- }
-
- @Override
- public void start() {
- jdbcRegistryServer.registerClient(this);
- }
-
- @Override
- public JdbcRegistryClientIdentify getJdbcRegistryClientIdentify() {
- return jdbcRegistryClientIdentify;
- }
-
- @Override
- public void subscribeConnectionStateChange(ConnectionStateListener connectionStateListener) {
- jdbcRegistryServer.subscribeConnectionStateChange(connectionStateListener);
- }
-
- @Override
- public void subscribeJdbcRegistryDataChange(JdbcRegistryDataChangeListener jdbcRegistryDataChangeListener) {
- jdbcRegistryServer.subscribeJdbcRegistryDataChange(jdbcRegistryDataChangeListener);
- }
-
- @Override
- public Optional<JdbcRegistryDataDTO> getJdbcRegistryDataByKey(String key) {
- return jdbcRegistryServer.getJdbcRegistryDataByKey(key);
- }
-
- @Override
- public void putJdbcRegistryData(String key, String value, DataType dataType) {
- jdbcRegistryServer.putJdbcRegistryData(jdbcRegistryClientIdentify.getClientId(), key, value, dataType);
- }
-
- @Override
- public void deleteJdbcRegistryDataByKey(String key) {
- jdbcRegistryServer.deleteJdbcRegistryDataByKey(key);
- }
-
- @Override
- public List<JdbcRegistryDataDTO> listJdbcRegistryDataChildren(String key) {
- return jdbcRegistryServer.listJdbcRegistryDataChildren(key);
- }
-
- @Override
- public boolean existJdbcRegistryDataKey(String key) {
- return jdbcRegistryServer.existJdbcRegistryDataKey(key);
- }
-
- @Override
- public void acquireJdbcRegistryLock(String key) {
- jdbcRegistryServer.acquireJdbcRegistryLock(jdbcRegistryClientIdentify.getClientId(), key);
- }
-
- @Override
- public boolean acquireJdbcRegistryLock(String key, long timeout) {
- return jdbcRegistryServer.acquireJdbcRegistryLock(jdbcRegistryClientIdentify.getClientId(), key, timeout);
- }
-
- @Override
- public void releaseJdbcRegistryLock(String key) {
- jdbcRegistryServer.releaseJdbcRegistryLock(jdbcRegistryClientIdentify.getClientId(), key);
- }
-
- @Override
- public void close() {
- jdbcRegistryServer.deregisterClient(this);
- log.info("Closed JdbcRegistryClient: {}", jdbcRegistryClientIdentify);
- }
-
- @Override
- public boolean isConnectivity() {
- return jdbcRegistryServer.getServerState() == JdbcRegistryServerState.STARTED;
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/client/JdbcRegistryClientIdentify.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/client/JdbcRegistryClientIdentify.java
deleted file mode 100644
index 7481c3a30..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/client/JdbcRegistryClientIdentify.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.registry.core.client;
-
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-import lombok.ToString;
-
-@ToString
-@Getter
-@AllArgsConstructor
-public class JdbcRegistryClientIdentify {
-
- private final Long clientId;
-
- private final String clientName;
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryClientHeartbeatMapper.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryClientHeartbeatMapper.java
deleted file mode 100644
index f1a27c299..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryClientHeartbeatMapper.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.registry.core.mapper;
-
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryClientHeartbeat;
-
-import org.apache.ibatis.annotations.Select;
-
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-
-import java.util.List;
-
-public interface JdbcRegistryClientHeartbeatMapper extends BaseMapper<JdbcRegistryClientHeartbeat> {
-
- @Select("select * from t_jdbc_registry_client_heartbeat")
- List<JdbcRegistryClientHeartbeat> selectAll();
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryDataChanceEventMapper.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryDataChanceEventMapper.java
deleted file mode 100644
index b66982ed9..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryDataChanceEventMapper.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.registry.core.mapper;
-
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryDataChanceEvent;
-
-import org.apache.ibatis.annotations.Delete;
-import org.apache.ibatis.annotations.Param;
-import org.apache.ibatis.annotations.Select;
-
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-
-import java.util.Date;
-import java.util.List;
-
-public interface JdbcRegistryDataChanceEventMapper extends BaseMapper<JdbcRegistryDataChanceEvent> {
-
- @Select("select max(id) from t_jdbc_registry_data_change_event")
- Long getMaxId();
-
- @Select("select * from t_jdbc_registry_data_change_event where id > #{id} order by id asc limit 1000")
- List<JdbcRegistryDataChanceEvent> selectJdbcRegistryDataChangeEventWhereIdAfter(@Param("id") long id);
-
- @Delete("delete from t_jdbc_registry_data_change_event where create_time > #{createTime}")
- void deleteJdbcRegistryDataChangeEventBeforeCreateTime(@Param("createTime") Date createTime);
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryDataMapper.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryDataMapper.java
deleted file mode 100644
index c0eb45dff..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryDataMapper.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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.registry.core.mapper;
-
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryData;
-
-import org.apache.ibatis.annotations.Delete;
-import org.apache.ibatis.annotations.Param;
-import org.apache.ibatis.annotations.Select;
-
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-
-import java.util.List;
-
-public interface JdbcRegistryDataMapper extends BaseMapper<JdbcRegistryData> {
-
- @Select("select * from t_jdbc_registry_data")
- List<JdbcRegistryData> selectAll();
-
- @Select("select * from t_jdbc_registry_data where data_key = #{key}")
- JdbcRegistryData selectByKey(@Param("key") String key);
-
- @Delete("delete from t_jdbc_registry_data where data_key = #{key}")
- void deleteByKey(@Param("key") String key);
-
- @Delete({"<script>",
- "delete from t_jdbc_registry_data",
- "where client_id IN ",
- "<foreach item='clientId' index='index' collection='clientIds' open='(' separator=',' close=')'>",
- " #{clientId}",
- "</foreach>",
- "and data_type = #{dataType}",
- "</script>"})
- void deleteByClientIds(@Param("clientIds") List<Long> clientIds, @Param("dataType") String dataType);
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryLockMapper.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryLockMapper.java
deleted file mode 100644
index 84f8d7e45..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/mapper/JdbcRegistryLockMapper.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.registry.core.mapper;
-
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryLock;
-
-import org.apache.ibatis.annotations.Delete;
-import org.apache.ibatis.annotations.Param;
-
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-
-import java.util.List;
-
-public interface JdbcRegistryLockMapper extends BaseMapper<JdbcRegistryLock> {
-
- @Delete({"<script>",
- "delete from t_jdbc_registry_lock",
- "where client_id IN ",
- "<foreach item='clientId' index='index' collection='clientIds' open='(' separator=',' close=')'>",
- " #{clientId}",
- "</foreach>",
- "</script>"})
- void deleteByClientIds(@Param("clientIds") List<Long> clientIds);
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryClientHeartbeat.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryClientHeartbeat.java
deleted file mode 100644
index 3875acab8..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryClientHeartbeat.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.registry.core.model.DO;
-
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.util.Date;
-
-@TableName(value = "t_jdbc_registry_client_heartbeat")
-@Getter
-@Setter
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class JdbcRegistryClientHeartbeat {
-
- @TableId(value = "id", type = IdType.INPUT)
- private Long id;
- private String clientName;
- private Long lastHeartbeatTime;
- private String connectionConfig;
- private Date createTime;
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryData.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryData.java
deleted file mode 100644
index 29e3c8810..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryData.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.registry.core.model.DO;
-
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.util.Date;
-
-@TableName(value = "t_jdbc_registry_data")
-@Getter
-@Setter
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class JdbcRegistryData {
-
- @TableId(value = "id", type = IdType.AUTO)
- private Long id;
- private String dataKey;
- private String dataValue;
- private String dataType;
- private long clientId;
- private Date createTime;
- private Date lastUpdateTime;
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryDataChanceEvent.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryDataChanceEvent.java
deleted file mode 100644
index 8088a3bb4..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryDataChanceEvent.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.registry.core.model.DO;
-
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.util.Date;
-
-@TableName(value = "t_jdbc_registry_data_change_event")
-@Getter
-@Setter
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class JdbcRegistryDataChanceEvent {
-
- @TableId(value = "id", type = IdType.AUTO)
- private Long id;
-
- private String eventType;
-
- private String jdbcRegistryData;
-
- private Date createTime;
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryLock.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryLock.java
deleted file mode 100644
index f77783c01..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DO/JdbcRegistryLock.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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.registry.core.model.DO;
-
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.util.Date;
-
-@TableName(value = "t_jdbc_registry_lock")
-@Getter
-@Setter
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class JdbcRegistryLock {
-
- @TableId(value = "id", type = IdType.AUTO)
- private Long id;
- private String lockKey;
- private String lockOwner;
- private Long clientId;
- private Date createTime;
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/DataType.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/DataType.java
deleted file mode 100644
index cd33dc186..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/DataType.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.registry.core.model.DTO;
-
-public enum DataType {
-
- EPHEMERAL(),
- PERSISTENT(),
- ;
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryClientHeartbeatDTO.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryClientHeartbeatDTO.java
deleted file mode 100644
index 462ba92f9..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryClientHeartbeatDTO.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.registry.core.model.DTO;
-
-import org.apache.streampark.common.utils.JSONUtils;
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryClientHeartbeat;
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-import lombok.SneakyThrows;
-
-import java.util.Date;
-
-@Getter
-@Setter
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class JdbcRegistryClientHeartbeatDTO {
-
- private Long id;
-
- // clientName
- private String clientName;
-
- private Long lastHeartbeatTime;
-
- private ClientConfig clientConfig;
-
- private Date createTime;
-
- public static JdbcRegistryClientHeartbeatDTO fromJdbcRegistryClientHeartbeat(JdbcRegistryClientHeartbeat jdbcRegistryClientHeartbeat) {
- return JdbcRegistryClientHeartbeatDTO.builder()
- .id(jdbcRegistryClientHeartbeat.getId())
- .clientName(jdbcRegistryClientHeartbeat.getClientName())
- .lastHeartbeatTime(jdbcRegistryClientHeartbeat.getLastHeartbeatTime())
- .clientConfig(
- JSONUtils.parseObject(jdbcRegistryClientHeartbeat.getConnectionConfig(), ClientConfig.class))
- .createTime(jdbcRegistryClientHeartbeat.getCreateTime())
- .build();
- }
-
- public static JdbcRegistryClientHeartbeat toJdbcRegistryClientHeartbeat(JdbcRegistryClientHeartbeatDTO jdbcRegistryClientHeartbeatDTO) {
- return JdbcRegistryClientHeartbeat.builder()
- .id(jdbcRegistryClientHeartbeatDTO.getId())
- .clientName(jdbcRegistryClientHeartbeatDTO.getClientName())
- .lastHeartbeatTime(jdbcRegistryClientHeartbeatDTO.getLastHeartbeatTime())
- .connectionConfig(JSONUtils.toJsonString(jdbcRegistryClientHeartbeatDTO.getClientConfig()))
- .createTime(jdbcRegistryClientHeartbeatDTO.getCreateTime())
- .build();
- }
-
- public boolean isDead() {
- // check if the client connection is expired.
- return System.currentTimeMillis() - lastHeartbeatTime > clientConfig.getSessionTimeout();
- }
-
- @SneakyThrows
- @Override
- public JdbcRegistryClientHeartbeatDTO clone() {
- return JdbcRegistryClientHeartbeatDTO.builder()
- .id(id)
- .clientName(clientName)
- .lastHeartbeatTime(lastHeartbeatTime)
- .clientConfig(clientConfig)
- .createTime(createTime)
- .build();
- }
-
- @Getter
- @Setter
- @Builder
- @NoArgsConstructor
- @AllArgsConstructor
- public static class ClientConfig {
-
- @Builder.Default
- private long sessionTimeout = 60 * 1000L;
-
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryDataChanceEventDTO.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryDataChanceEventDTO.java
deleted file mode 100644
index 8803e83fc..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryDataChanceEventDTO.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.registry.core.model.DTO;
-
-import org.apache.streampark.common.utils.JSONUtils;
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryData;
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryDataChanceEvent;
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.util.Date;
-
-@Getter
-@Setter
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class JdbcRegistryDataChanceEventDTO {
-
- private Long id;
-
- private EventType eventType;
-
- private JdbcRegistryDataDTO jdbcRegistryData;
-
- private Date createTime;
-
- public enum EventType {
- ADD,
- UPDATE,
- DELETE;
-
- }
-
- public static JdbcRegistryDataChanceEventDTO fromJdbcRegistryDataChanceEvent(JdbcRegistryDataChanceEvent jdbcRegistryDataChanceEvent) {
- JdbcRegistryData jdbcRegistryData =
- JSONUtils.parseObject(jdbcRegistryDataChanceEvent.getJdbcRegistryData(), JdbcRegistryData.class);
- if (jdbcRegistryData == null) {
- throw new IllegalArgumentException(
- "jdbcRegistryData: " + jdbcRegistryDataChanceEvent.getJdbcRegistryData() + " is invalidated");
- }
- return JdbcRegistryDataChanceEventDTO.builder()
- .id(jdbcRegistryDataChanceEvent.getId())
- .jdbcRegistryData(JdbcRegistryDataDTO.fromJdbcRegistryData(jdbcRegistryData))
- .eventType(EventType.valueOf(jdbcRegistryDataChanceEvent.getEventType()))
- .createTime(jdbcRegistryDataChanceEvent.getCreateTime())
- .build();
- }
-
- public static JdbcRegistryDataChanceEvent toJdbcRegistryDataChanceEvent(JdbcRegistryDataChanceEventDTO jdbcRegistryDataChanceEvent) {
- return JdbcRegistryDataChanceEvent.builder()
- .id(jdbcRegistryDataChanceEvent.getId())
- .jdbcRegistryData(JSONUtils.toJsonString(jdbcRegistryDataChanceEvent.getJdbcRegistryData()))
- .eventType(jdbcRegistryDataChanceEvent.getEventType().name())
- .createTime(jdbcRegistryDataChanceEvent.getCreateTime())
- .build();
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryDataDTO.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryDataDTO.java
deleted file mode 100644
index 06df011a7..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryDataDTO.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.registry.core.model.DTO;
-
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryData;
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.util.Date;
-
-@Getter
-@Setter
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class JdbcRegistryDataDTO {
-
- private Long id;
- private String dataKey;
- private String dataValue;
- private String dataType;
- private Long clientId;
- private Date createTime;
- private Date lastUpdateTime;
-
- public static JdbcRegistryDataDTO fromJdbcRegistryData(JdbcRegistryData jdbcRegistryData) {
- return JdbcRegistryDataDTO.builder()
- .id(jdbcRegistryData.getId())
- .dataKey(jdbcRegistryData.getDataKey())
- .dataValue(jdbcRegistryData.getDataValue())
- .dataType(jdbcRegistryData.getDataType())
- .clientId(jdbcRegistryData.getClientId())
- .createTime(jdbcRegistryData.getCreateTime())
- .lastUpdateTime(jdbcRegistryData.getLastUpdateTime())
- .build();
- }
-
- public static JdbcRegistryData toJdbcRegistryData(JdbcRegistryDataDTO jdbcRegistryData) {
- return JdbcRegistryData.builder()
- .id(jdbcRegistryData.getId())
- .dataKey(jdbcRegistryData.getDataKey())
- .dataValue(jdbcRegistryData.getDataValue())
- .dataType(jdbcRegistryData.getDataType())
- .clientId(jdbcRegistryData.getClientId())
- .createTime(jdbcRegistryData.getCreateTime())
- .lastUpdateTime(jdbcRegistryData.getLastUpdateTime())
- .build();
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryLockDTO.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryLockDTO.java
deleted file mode 100644
index e274f52ec..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/model/DTO/JdbcRegistryLockDTO.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.registry.core.model.DTO;
-
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryLock;
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-
-import java.util.Date;
-
-@Getter
-@Setter
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class JdbcRegistryLockDTO {
-
- private Long id;
- private String lockKey;
- private String lockOwner;
- private Long clientId;
- private Date createTime;
-
- public static JdbcRegistryLockDTO fromJdbcRegistryLock(JdbcRegistryLock jdbcRegistryLock) {
- return JdbcRegistryLockDTO.builder()
- .id(jdbcRegistryLock.getId())
- .lockKey(jdbcRegistryLock.getLockKey())
- .lockOwner(jdbcRegistryLock.getLockOwner())
- .clientId(jdbcRegistryLock.getClientId())
- .createTime(jdbcRegistryLock.getCreateTime())
- .build();
- }
-
- public static JdbcRegistryLock toJdbcRegistryLock(JdbcRegistryLockDTO jdbcRegistryLock) {
- return JdbcRegistryLock.builder()
- .id(jdbcRegistryLock.getId())
- .lockKey(jdbcRegistryLock.getLockKey())
- .lockOwner(jdbcRegistryLock.getLockOwner())
- .clientId(jdbcRegistryLock.getClientId())
- .createTime(jdbcRegistryLock.getCreateTime())
- .build();
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryClientRepository.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryClientRepository.java
deleted file mode 100644
index c3ae1797b..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryClientRepository.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * 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.registry.core.repository;
-
-import org.apache.streampark.registry.core.mapper.JdbcRegistryClientHeartbeatMapper;
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryClientHeartbeat;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryClientHeartbeatDTO;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Repository;
-import org.springframework.util.CollectionUtils;
-
-import java.util.List;
-import java.util.stream.Collectors;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-@Repository
-public class JdbcRegistryClientRepository {
-
- @Autowired
- private JdbcRegistryClientHeartbeatMapper jdbcRegistryClientHeartbeatMapper;
-
- public List<JdbcRegistryClientHeartbeatDTO> queryAll() {
- return jdbcRegistryClientHeartbeatMapper.selectAll()
- .stream()
- .map(JdbcRegistryClientHeartbeatDTO::fromJdbcRegistryClientHeartbeat)
- .collect(Collectors.toList());
- }
-
- public void deleteByIds(List<Long> clientIds) {
- if (CollectionUtils.isEmpty(clientIds)) {
- return;
- }
- jdbcRegistryClientHeartbeatMapper.deleteBatchIds(clientIds);
- }
-
- public boolean updateById(JdbcRegistryClientHeartbeatDTO jdbcRegistryClientHeartbeatDTO) {
- JdbcRegistryClientHeartbeat jdbcRegistryClientHeartbeat =
- JdbcRegistryClientHeartbeatDTO.toJdbcRegistryClientHeartbeat(jdbcRegistryClientHeartbeatDTO);
- return jdbcRegistryClientHeartbeatMapper.updateById(jdbcRegistryClientHeartbeat) == 1;
- }
-
- public void insert(JdbcRegistryClientHeartbeatDTO jdbcRegistryClient) {
- checkNotNull(jdbcRegistryClient.getId());
- JdbcRegistryClientHeartbeat jdbcRegistryClientHeartbeat =
- JdbcRegistryClientHeartbeatDTO.toJdbcRegistryClientHeartbeat(jdbcRegistryClient);
- jdbcRegistryClientHeartbeatMapper.insert(jdbcRegistryClientHeartbeat);
-
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryDataChanceEventRepository.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryDataChanceEventRepository.java
deleted file mode 100644
index 489443827..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryDataChanceEventRepository.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.registry.core.repository;
-
-import org.apache.streampark.registry.core.mapper.JdbcRegistryDataChanceEventMapper;
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryDataChanceEvent;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataChanceEventDTO;
-
-import org.springframework.stereotype.Repository;
-
-import java.util.Date;
-import java.util.List;
-import java.util.stream.Collectors;
-
-@Repository
-public class JdbcRegistryDataChanceEventRepository {
-
- private final JdbcRegistryDataChanceEventMapper jdbcRegistryDataChanceEventMapper;
-
- public JdbcRegistryDataChanceEventRepository(JdbcRegistryDataChanceEventMapper jdbcRegistryDataChanceEventMapper) {
- this.jdbcRegistryDataChanceEventMapper = jdbcRegistryDataChanceEventMapper;
- }
-
- public long getMaxJdbcRegistryDataChanceEventId() {
- Long maxId = jdbcRegistryDataChanceEventMapper.getMaxId();
- if (maxId == null) {
- return -1;
- } else {
- return maxId;
- }
- }
-
- public List<JdbcRegistryDataChanceEventDTO> selectJdbcRegistryDataChangeEventWhereIdAfter(long id) {
- return jdbcRegistryDataChanceEventMapper.selectJdbcRegistryDataChangeEventWhereIdAfter(id)
- .stream()
- .map(JdbcRegistryDataChanceEventDTO::fromJdbcRegistryDataChanceEvent)
- .collect(Collectors.toList());
- }
-
- public void insert(JdbcRegistryDataChanceEventDTO registryDataChanceEvent) {
- JdbcRegistryDataChanceEvent jdbcRegistryDataChanceEvent =
- JdbcRegistryDataChanceEventDTO.toJdbcRegistryDataChanceEvent(registryDataChanceEvent);
- jdbcRegistryDataChanceEventMapper.insert(jdbcRegistryDataChanceEvent);
- registryDataChanceEvent.setId(jdbcRegistryDataChanceEvent.getId());
- }
-
- public void deleteJdbcRegistryDataChangeEventBeforeCreateTime(Date createTime) {
- jdbcRegistryDataChanceEventMapper.deleteJdbcRegistryDataChangeEventBeforeCreateTime(createTime);
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryDataRepository.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryDataRepository.java
deleted file mode 100644
index 0ceedafb1..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryDataRepository.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.registry.core.repository;
-
-import org.apache.streampark.registry.core.mapper.JdbcRegistryDataMapper;
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryData;
-import org.apache.streampark.registry.core.model.DTO.DataType;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataDTO;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Repository;
-import org.springframework.util.CollectionUtils;
-
-import java.util.List;
-import java.util.Optional;
-import java.util.stream.Collectors;
-
-@Repository
-public class JdbcRegistryDataRepository {
-
- @Autowired
- private JdbcRegistryDataMapper jdbcRegistryDataMapper;
-
- public List<JdbcRegistryDataDTO> selectAll() {
- return jdbcRegistryDataMapper
- .selectAll()
- .stream()
- .map(JdbcRegistryDataDTO::fromJdbcRegistryData)
- .collect(Collectors.toList());
- }
-
- public Optional<JdbcRegistryDataDTO> selectByKey(String key) {
- return Optional.ofNullable(jdbcRegistryDataMapper.selectByKey(key))
- .map(JdbcRegistryDataDTO::fromJdbcRegistryData);
- }
-
- public void deleteEphemeralDateByClientIds(List<Long> clientIds) {
- if (CollectionUtils.isEmpty(clientIds)) {
- return;
- }
- jdbcRegistryDataMapper.deleteByClientIds(clientIds, DataType.EPHEMERAL.name());
- }
-
- public void deleteByKey(String key) {
- jdbcRegistryDataMapper.deleteByKey(key);
- }
-
- public void insert(JdbcRegistryDataDTO jdbcRegistryData) {
- JdbcRegistryData jdbcRegistryDataDO = JdbcRegistryDataDTO.toJdbcRegistryData(jdbcRegistryData);
- jdbcRegistryDataMapper.insert(jdbcRegistryDataDO);
- jdbcRegistryData.setId(jdbcRegistryDataDO.getId());
- }
-
- public void updateById(JdbcRegistryDataDTO jdbcRegistryDataDTO) {
- jdbcRegistryDataMapper.updateById(JdbcRegistryDataDTO.toJdbcRegistryData(jdbcRegistryDataDTO));
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryLockRepository.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryLockRepository.java
deleted file mode 100644
index b9771bc44..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/repository/JdbcRegistryLockRepository.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * 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.registry.core.repository;
-
-import org.apache.streampark.registry.core.mapper.JdbcRegistryLockMapper;
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryLock;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryLockDTO;
-
-import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Repository;
-
-import java.util.List;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-@Repository
-public class JdbcRegistryLockRepository {
-
- @Autowired
- private JdbcRegistryLockMapper jdbcRegistryLockMapper;
-
- public void deleteByClientIds(List<Long> clientIds) {
- if (CollectionUtils.isEmpty(clientIds)) {
- return;
- }
- jdbcRegistryLockMapper.deleteByClientIds(clientIds);
- }
-
- public void insert(JdbcRegistryLockDTO jdbcRegistryLock) {
- checkNotNull(jdbcRegistryLock);
- JdbcRegistryLock jdbcRegistryLockDO = JdbcRegistryLockDTO.toJdbcRegistryLock(jdbcRegistryLock);
- jdbcRegistryLockMapper.insert(jdbcRegistryLockDO);
- jdbcRegistryLock.setId(jdbcRegistryLockDO.getId());
- }
-
- public void deleteById(Long id) {
- jdbcRegistryLockMapper.deleteById(id);
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/ConnectionStateListener.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/ConnectionStateListener.java
deleted file mode 100644
index e0053c797..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/ConnectionStateListener.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * 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.registry.core.server;
-
-public interface ConnectionStateListener {
-
- void onConnected();
-
- void onDisConnected();
-
- void onReconnected();
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IJdbcRegistryDataManager.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IJdbcRegistryDataManager.java
deleted file mode 100644
index 4ecca0738..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IJdbcRegistryDataManager.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * 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.registry.core.server;
-
-import org.apache.streampark.registry.core.model.DTO.DataType;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataDTO;
-
-import java.util.List;
-import java.util.Optional;
-
-public interface IJdbcRegistryDataManager {
-
- boolean existKey(String key);
-
- /**
- * Get the {@link JdbcRegistryDataDTO} by key.
- */
- Optional<JdbcRegistryDataDTO> getRegistryDataByKey(String key);
-
- /**
- * List all the {@link JdbcRegistryDataDTO} children by key.
- * <p>
- * e.g. key = "/streampark/master", and data exist in db is "/streampark/master/master1", "/streampark/master/master2"
- * <p>
- * then the return value will be ["master1", "master2"]
- */
- List<JdbcRegistryDataDTO> listJdbcRegistryDataChildren(String key);
-
- /**
- * Put the {@link JdbcRegistryDataDTO} to the jdbc registry server.
- * <p>
- * If the key is already exist, then update the {@link JdbcRegistryDataDTO}. If the key is not exist, then insert a new {@link JdbcRegistryDataDTO}.
- */
- void putJdbcRegistryData(Long clientId, String key, String value, DataType dataType);
-
- /**
- * Delete the {@link JdbcRegistryDataDTO} by key.
- */
- void deleteJdbcRegistryDataByKey(String key);
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IJdbcRegistryLockManager.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IJdbcRegistryLockManager.java
deleted file mode 100644
index 7d8f9d291..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IJdbcRegistryLockManager.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.registry.core.server;
-
-public interface IJdbcRegistryLockManager {
-
- /**
- * Acquire the jdbc registry lock by key. this is a blocking method. if you want to stop the blocking, you can use interrupt the thread.
- */
- void acquireJdbcRegistryLock(Long clientId, String lockKey) throws InterruptedException;
-
- /**
- * Acquire the jdbc registry lock by key until timeout.
- */
- boolean acquireJdbcRegistryLock(Long clientId, String lockKey, long timeout);
-
- /**
- * Release the jdbc registry lock by key, if the lockKey is not exist will do nothing.
- */
- void releaseJdbcRegistryLock(Long clientId, String lockKey);
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IJdbcRegistryServer.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IJdbcRegistryServer.java
deleted file mode 100644
index 3f03e7460..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IJdbcRegistryServer.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * 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.registry.core.server;
-
-import org.apache.streampark.registry.core.client.IJdbcRegistryClient;
-import org.apache.streampark.registry.core.model.DO.JdbcRegistryData;
-import org.apache.streampark.registry.core.model.DTO.DataType;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataDTO;
-
-import java.util.List;
-import java.util.Optional;
-
-/**
- * The JdbcRegistryServer is represent the server side of the jdbc registry, it can be thought as db server.
- */
-public interface IJdbcRegistryServer extends AutoCloseable {
-
- void start();
-
- /**
- * Register a client to the server, once the client connect to the server then the server will refresh the client's term interval.
- */
- void registerClient(IJdbcRegistryClient jdbcRegistryClient);
-
- /**
- * Deregister a client to the server, once the client id deregister, then the server will deleted the data related to the client and stop refresh the client's term.
- */
- void deregisterClient(IJdbcRegistryClient jdbcRegistryClient);
-
- /**
- * Get the {@link JdbcRegistryServerState}
- */
- JdbcRegistryServerState getServerState();
-
- /**
- * Subscribe the jdbc registry connection state change
- */
- void subscribeConnectionStateChange(ConnectionStateListener connectionStateListener);
-
- /**
- * Subscribe the {@link JdbcRegistryData} change.
- */
- void subscribeJdbcRegistryDataChange(JdbcRegistryDataChangeListener jdbcRegistryDataChangeListener);
-
- /**
- * Check the jdbc registry data key is exist or not.
- */
- boolean existJdbcRegistryDataKey(String key);
-
- /**
- * Get the {@link JdbcRegistryDataDTO} by key.
- */
- Optional<JdbcRegistryDataDTO> getJdbcRegistryDataByKey(String key);
-
- /**
- * List all the {@link JdbcRegistryDataDTO} children by key.
- * <p>
- * e.g. key = "/streampark/master", and data exist in db is "/streampark/master/master1", "/streampark/master/master2"
- * <p>
- * then the return value will be ["master1", "master2"]
- */
- List<JdbcRegistryDataDTO> listJdbcRegistryDataChildren(String key);
-
- /**
- * Put the {@link JdbcRegistryDataDTO} to the jdbc registry server.
- * <p>
- * If the key is already exist, then update the {@link JdbcRegistryDataDTO}. If the key is not exist, then insert a new {@link JdbcRegistryDataDTO}.
- */
- void putJdbcRegistryData(Long clientId, String key, String value, DataType dataType);
-
- /**
- * Delete the {@link JdbcRegistryDataDTO} by key.
- */
- void deleteJdbcRegistryDataByKey(String key);
-
- /**
- * Acquire the jdbc registry lock by key. this is a blocking method. if you want to stop the blocking, you can use interrupt the thread.
- */
- void acquireJdbcRegistryLock(Long clientId, String key);
-
- /**
- * Acquire the jdbc registry lock by key until timeout.
- */
- boolean acquireJdbcRegistryLock(Long clientId, String key, long timeout);
-
- /**
- * Release the jdbc registry lock by key, if the lockKey is not exist will do nothing.
- */
- void releaseJdbcRegistryLock(Long clientId, String key);
-
- /**
- * Close the server, once the server been closed, it cannot work anymore.
- */
- @Override
- void close();
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IRegistryRowChangeNotifier.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IRegistryRowChangeNotifier.java
deleted file mode 100644
index 723f63103..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/IRegistryRowChangeNotifier.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.registry.core.server;
-
-public interface IRegistryRowChangeNotifier<T> {
-
- void start();
-
- void subscribeRegistryRowChange(RegistryRowChangeListener<T> registryRowChangeListener);
-
- interface RegistryRowChangeListener<T> {
-
- void onRegistryRowUpdated(T data);
-
- void onRegistryRowAdded(T data);
-
- void onRegistryRowDeleted(T data);
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryDataChangeListener.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryDataChangeListener.java
deleted file mode 100644
index 015219ea8..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryDataChangeListener.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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.registry.core.server;
-
-public interface JdbcRegistryDataChangeListener {
-
- void onJdbcRegistryDataChanged(String key, String value);
-
- void onJdbcRegistryDataDeleted(String key);
-
- void onJdbcRegistryDataAdded(String key, String value);
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryDataManager.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryDataManager.java
deleted file mode 100644
index 1b2ef8c49..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryDataManager.java
+++ /dev/null
@@ -1,264 +0,0 @@
-/*
- * 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.registry.core.server;
-
-import org.apache.streampark.registry.core.JdbcRegistryProperties;
-import org.apache.streampark.registry.core.JdbcRegistryThreadFactory;
-import org.apache.streampark.registry.core.model.DTO.DataType;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataChanceEventDTO;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataDTO;
-import org.apache.streampark.registry.core.repository.JdbcRegistryDataChanceEventRepository;
-import org.apache.streampark.registry.core.repository.JdbcRegistryDataRepository;
-
-import org.apache.commons.lang3.time.DateUtils;
-
-import com.google.common.collect.Lists;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.util.CollectionUtils;
-
-import java.time.Duration;
-import java.util.Date;
-import java.util.List;
-import java.util.Optional;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-@Slf4j
-public class JdbcRegistryDataManager
- implements
- IRegistryRowChangeNotifier<JdbcRegistryDataDTO>,
- IJdbcRegistryDataManager {
-
- private final Integer keepJdbcRegistryDataChanceEventHours = 2;
-
- private final JdbcRegistryProperties registryProperties;
-
- private final JdbcRegistryDataRepository jdbcRegistryDataRepository;
-
- private final JdbcRegistryDataChanceEventRepository jdbcRegistryDataChanceEventRepository;
-
- private final List<RegistryRowChangeListener<JdbcRegistryDataDTO>> registryRowChangeListeners;
-
- private long lastDetectedJdbcRegistryDataChangeEventId = -1;
-
- public JdbcRegistryDataManager(JdbcRegistryProperties registryProperties,
- JdbcRegistryDataRepository jdbcRegistryDataRepository,
- JdbcRegistryDataChanceEventRepository jdbcRegistryDataChanceEventRepository) {
- this.registryProperties = registryProperties;
- this.jdbcRegistryDataChanceEventRepository = jdbcRegistryDataChanceEventRepository;
- this.jdbcRegistryDataRepository = jdbcRegistryDataRepository;
- this.registryRowChangeListeners = new CopyOnWriteArrayList<>();
- this.lastDetectedJdbcRegistryDataChangeEventId =
- jdbcRegistryDataChanceEventRepository.getMaxJdbcRegistryDataChanceEventId();
- }
-
- @Override
- public void start() {
- JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().scheduleWithFixedDelay(
- this::detectJdbcRegistryDataChangeEvent,
- registryProperties.getHeartbeatRefreshInterval().toMillis(),
- registryProperties.getHeartbeatRefreshInterval().toMillis(),
- TimeUnit.MILLISECONDS);
-
- JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().scheduleWithFixedDelay(
- this::purgeHistoryJdbcRegistryDataChangeEvent,
- 0,
- Duration.ofHours(keepJdbcRegistryDataChanceEventHours).toHours(),
- TimeUnit.HOURS);
- }
-
- private void detectJdbcRegistryDataChangeEvent() {
- final List<JdbcRegistryDataChanceEventDTO> jdbcRegistryDataChanceEvents = jdbcRegistryDataChanceEventRepository
- .selectJdbcRegistryDataChangeEventWhereIdAfter(lastDetectedJdbcRegistryDataChangeEventId);
- if (CollectionUtils.isEmpty(jdbcRegistryDataChanceEvents)) {
- return;
- }
- for (JdbcRegistryDataChanceEventDTO jdbcRegistryDataChanceEvent : jdbcRegistryDataChanceEvents) {
- log.debug("Detect JdbcRegistryDataChangeEvent: {}", jdbcRegistryDataChanceEvent);
- switch (jdbcRegistryDataChanceEvent.getEventType()) {
- case ADD:
- doTriggerJdbcRegistryDataAddedListener(
- Lists.newArrayList(jdbcRegistryDataChanceEvent.getJdbcRegistryData()));
- break;
- case UPDATE:
- doTriggerJdbcRegistryDataUpdatedListener(
- Lists.newArrayList(jdbcRegistryDataChanceEvent.getJdbcRegistryData()));
- break;
- case DELETE:
- doTriggerJdbcRegistryDataRemovedListener(
- Lists.newArrayList(jdbcRegistryDataChanceEvent.getJdbcRegistryData()));
- break;
- default:
- log.error("Unknown event type: {}", jdbcRegistryDataChanceEvent.getEventType());
- break;
- }
- if (jdbcRegistryDataChanceEvent.getId() > lastDetectedJdbcRegistryDataChangeEventId) {
- lastDetectedJdbcRegistryDataChangeEventId = jdbcRegistryDataChanceEvent.getId();
- }
- }
- }
-
- private void purgeHistoryJdbcRegistryDataChangeEvent() {
- log.info("Purge JdbcRegistryDataChanceEvent which createTime is before: {} hours",
- keepJdbcRegistryDataChanceEventHours);
- jdbcRegistryDataChanceEventRepository.deleteJdbcRegistryDataChangeEventBeforeCreateTime(
- DateUtils.addHours(new Date(), -keepJdbcRegistryDataChanceEventHours));
- }
-
- @Override
- public void subscribeRegistryRowChange(RegistryRowChangeListener<JdbcRegistryDataDTO> registryRowChangeListener) {
- registryRowChangeListeners.add(checkNotNull(registryRowChangeListener));
- }
-
- @Override
- public boolean existKey(String key) {
- checkNotNull(key);
- return jdbcRegistryDataRepository.selectByKey(key).isPresent();
- }
-
- @Override
- public Optional<JdbcRegistryDataDTO> getRegistryDataByKey(String key) {
- checkNotNull(key);
- return jdbcRegistryDataRepository.selectByKey(key);
- }
-
- @Override
- public List<JdbcRegistryDataDTO> listJdbcRegistryDataChildren(String key) {
- checkNotNull(key);
- return jdbcRegistryDataRepository.selectAll()
- .stream()
- .filter(jdbcRegistryDataDTO -> jdbcRegistryDataDTO.getDataKey().startsWith(key)
- && !jdbcRegistryDataDTO.getDataKey().equals(key))
- .collect(Collectors.toList());
- }
-
- @Override
- public void putJdbcRegistryData(Long clientId, String key, String value, DataType dataType) {
- checkNotNull(clientId);
- checkNotNull(key);
- checkNotNull(dataType);
-
- Optional<JdbcRegistryDataDTO> jdbcRegistryDataOptional = jdbcRegistryDataRepository.selectByKey(key);
- if (jdbcRegistryDataOptional.isPresent()) {
- JdbcRegistryDataDTO jdbcRegistryData = jdbcRegistryDataOptional.get();
- if (!dataType.name().equals(jdbcRegistryData.getDataType())) {
- throw new UnsupportedOperationException("The data type: " + jdbcRegistryData.getDataType()
- + " of the key: " + key + " cannot be updated");
- }
-
- if (DataType.EPHEMERAL.name().equals(jdbcRegistryData.getDataType())) {
- if (!jdbcRegistryData.getClientId().equals(clientId)) {
- throw new UnsupportedOperationException(
- "The EPHEMERAL data: " + key + " can only be updated by its owner: "
- + jdbcRegistryData.getClientId() + " but not: " + clientId);
- }
- }
-
- jdbcRegistryData.setDataValue(value);
- jdbcRegistryData.setLastUpdateTime(new Date());
- jdbcRegistryDataRepository.updateById(jdbcRegistryData);
-
- JdbcRegistryDataChanceEventDTO jdbcRegistryDataChanceEvent = JdbcRegistryDataChanceEventDTO.builder()
- .jdbcRegistryData(jdbcRegistryData)
- .eventType(JdbcRegistryDataChanceEventDTO.EventType.UPDATE)
- .createTime(new Date())
- .build();
- jdbcRegistryDataChanceEventRepository.insert(jdbcRegistryDataChanceEvent);
- } else {
- JdbcRegistryDataDTO jdbcRegistryDataDTO = JdbcRegistryDataDTO.builder()
- .clientId(clientId)
- .dataKey(key)
- .dataValue(value)
- .dataType(dataType.name())
- .createTime(new Date())
- .lastUpdateTime(new Date())
- .build();
- jdbcRegistryDataRepository.insert(jdbcRegistryDataDTO);
- JdbcRegistryDataChanceEventDTO registryDataChanceEvent = JdbcRegistryDataChanceEventDTO.builder()
- .jdbcRegistryData(jdbcRegistryDataDTO)
- .eventType(JdbcRegistryDataChanceEventDTO.EventType.ADD)
- .createTime(new Date())
- .build();
- jdbcRegistryDataChanceEventRepository.insert(registryDataChanceEvent);
- }
-
- }
-
- @Override
- public void deleteJdbcRegistryDataByKey(String key) {
- checkNotNull(key);
- // todo: this is not atomic, need to be improved
- Optional<JdbcRegistryDataDTO> jdbcRegistryDataOptional = jdbcRegistryDataRepository.selectByKey(key);
- if (!jdbcRegistryDataOptional.isPresent()) {
- return;
- }
- jdbcRegistryDataRepository.deleteByKey(key);
- JdbcRegistryDataChanceEventDTO registryDataChanceEvent = JdbcRegistryDataChanceEventDTO.builder()
- .jdbcRegistryData(jdbcRegistryDataOptional.get())
- .eventType(JdbcRegistryDataChanceEventDTO.EventType.DELETE)
- .createTime(new Date())
- .build();
- jdbcRegistryDataChanceEventRepository.insert(registryDataChanceEvent);
- }
-
- private void doTriggerJdbcRegistryDataAddedListener(List<JdbcRegistryDataDTO> valuesToAdd) {
- if (CollectionUtils.isEmpty(valuesToAdd)) {
- return;
- }
- log.debug("Trigger:onJdbcRegistryDataAdded: {}", valuesToAdd);
- valuesToAdd.forEach(jdbcRegistryData -> {
- try {
- registryRowChangeListeners.forEach(listener -> listener.onRegistryRowAdded(jdbcRegistryData));
- } catch (Exception ex) {
- log.error("Trigger:onRegistryRowAdded: {} failed", jdbcRegistryData, ex);
- }
- });
- }
-
- private void doTriggerJdbcRegistryDataRemovedListener(List<JdbcRegistryDataDTO> valuesToRemoved) {
- if (CollectionUtils.isEmpty(valuesToRemoved)) {
- return;
- }
- log.debug("Trigger:onJdbcRegistryDataDeleted: {}", valuesToRemoved);
- valuesToRemoved.forEach(jdbcRegistryData -> {
- try {
- registryRowChangeListeners.forEach(listener -> listener.onRegistryRowDeleted(jdbcRegistryData));
- } catch (Exception ex) {
- log.error("Trigger:onRegistryRowAdded: {} failed", jdbcRegistryData, ex);
- }
- });
- }
-
- private void doTriggerJdbcRegistryDataUpdatedListener(List<JdbcRegistryDataDTO> valuesToUpdated) {
- if (CollectionUtils.isEmpty(valuesToUpdated)) {
- return;
- }
- log.debug("Trigger:onJdbcRegistryDataUpdated: {}", valuesToUpdated);
- valuesToUpdated.forEach(jdbcRegistryData -> {
- try {
- registryRowChangeListeners.forEach(listener -> listener.onRegistryRowUpdated(jdbcRegistryData));
- } catch (Exception ex) {
- log.error("Trigger:onRegistryRowAdded: {} failed", jdbcRegistryData, ex);
- }
- });
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryLockManager.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryLockManager.java
deleted file mode 100644
index ec0392f3a..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryLockManager.java
+++ /dev/null
@@ -1,150 +0,0 @@
-/*
- * 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.registry.core.server;
-
-import org.apache.streampark.registry.api.thread.ThreadUtils;
-import org.apache.streampark.registry.core.JdbcRegistryProperties;
-import org.apache.streampark.registry.core.LockUtils;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryLockDTO;
-import org.apache.streampark.registry.core.repository.JdbcRegistryLockRepository;
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.dao.DuplicateKeyException;
-
-import java.util.Date;
-import java.util.HashMap;
-import java.util.Map;
-
-@Slf4j
-public class JdbcRegistryLockManager implements IJdbcRegistryLockManager {
-
- private final JdbcRegistryProperties jdbcRegistryProperties;
- private final JdbcRegistryLockRepository jdbcRegistryLockRepository;
-
- // lockKey -> LockEntry
- private final Map<String, LockEntry> jdbcRegistryLockHolderMap = new HashMap<>();
-
- public JdbcRegistryLockManager(JdbcRegistryProperties jdbcRegistryProperties,
- JdbcRegistryLockRepository jdbcRegistryLockRepository) {
- this.jdbcRegistryProperties = jdbcRegistryProperties;
- this.jdbcRegistryLockRepository = jdbcRegistryLockRepository;
- }
-
- @Override
- public void acquireJdbcRegistryLock(Long clientId, String lockKey) {
- String lockOwner = LockUtils.getLockOwner();
- while (true) {
- LockEntry lockEntry = jdbcRegistryLockHolderMap.get(lockKey);
- if (lockEntry != null && lockOwner.equals(lockEntry.getLockOwner())) {
- return;
- }
- JdbcRegistryLockDTO jdbcRegistryLock = JdbcRegistryLockDTO.builder()
- .lockKey(lockKey)
- .clientId(clientId)
- .lockOwner(lockOwner)
- .createTime(new Date())
- .build();
- try {
- jdbcRegistryLockRepository.insert(jdbcRegistryLock);
- if (jdbcRegistryLock != null) {
- jdbcRegistryLockHolderMap.put(lockKey, LockEntry.builder()
- .lockKey(lockKey)
- .lockOwner(lockOwner)
- .jdbcRegistryLock(jdbcRegistryLock)
- .build());
- return;
- }
- log.debug("{} acquire the lock {} success", lockOwner, lockKey);
- } catch (DuplicateKeyException duplicateKeyException) {
- // The lock is already exist, wait it release.
- continue;
- }
- log.debug("Acquire the lock {} failed try again", lockKey);
- // acquire failed, wait and try again
- ThreadUtils.sleep(jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis());
- }
- }
-
- @Override
- public boolean acquireJdbcRegistryLock(Long clientId, String lockKey, long timeout) {
- String lockOwner = LockUtils.getLockOwner();
- long start = System.currentTimeMillis();
- while (System.currentTimeMillis() - start <= timeout) {
- LockEntry lockEntry = jdbcRegistryLockHolderMap.get(lockKey);
- if (lockEntry != null && lockOwner.equals(lockEntry.getLockOwner())) {
- return true;
- }
- JdbcRegistryLockDTO jdbcRegistryLock = JdbcRegistryLockDTO.builder()
- .lockKey(lockKey)
- .clientId(clientId)
- .lockOwner(lockOwner)
- .createTime(new Date())
- .build();
- try {
- jdbcRegistryLockRepository.insert(jdbcRegistryLock);
- if (jdbcRegistryLock != null) {
- jdbcRegistryLockHolderMap.put(lockKey, LockEntry.builder()
- .lockKey(lockKey)
- .lockOwner(lockOwner)
- .jdbcRegistryLock(jdbcRegistryLock)
- .build());
- return true;
- }
- log.debug("{} acquire the lock {} success", lockOwner, lockKey);
- } catch (DuplicateKeyException duplicateKeyException) {
- // The lock is already exist, wait it release.
- continue;
- }
- log.debug("Acquire the lock {} failed try again", lockKey);
- // acquire failed, wait and try again
- ThreadUtils.sleep(jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis());
- }
- return false;
- }
-
- @Override
- public void releaseJdbcRegistryLock(Long clientId, String lockKey) {
- LockEntry lockEntry = jdbcRegistryLockHolderMap.get(lockKey);
- if (lockEntry == null) {
- return;
- }
- if (!clientId.equals(lockEntry.getJdbcRegistryLock().getClientId())) {
- throw new UnsupportedOperationException(
- "The client " + clientId + " is not the lock owner of the lock: " + lockKey);
- }
- jdbcRegistryLockRepository.deleteById(lockEntry.getJdbcRegistryLock().getId());
- jdbcRegistryLockHolderMap.remove(lockKey);
- }
-
- @Getter
- @Setter
- @Builder
- @NoArgsConstructor
- @AllArgsConstructor
- public static class LockEntry {
-
- private String lockKey;
- private String lockOwner;
- private JdbcRegistryLockDTO jdbcRegistryLock;
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryServer.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryServer.java
deleted file mode 100644
index 596cbe58e..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryServer.java
+++ /dev/null
@@ -1,386 +0,0 @@
-/*
- * 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.registry.core.server;
-
-import org.apache.streampark.registry.api.RegistryException;
-import org.apache.streampark.registry.core.JdbcRegistryProperties;
-import org.apache.streampark.registry.core.JdbcRegistryThreadFactory;
-import org.apache.streampark.registry.core.client.IJdbcRegistryClient;
-import org.apache.streampark.registry.core.client.JdbcRegistryClientIdentify;
-import org.apache.streampark.registry.core.model.DTO.DataType;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryClientHeartbeatDTO;
-import org.apache.streampark.registry.core.model.DTO.JdbcRegistryDataDTO;
-import org.apache.streampark.registry.core.repository.JdbcRegistryClientRepository;
-import org.apache.streampark.registry.core.repository.JdbcRegistryDataChanceEventRepository;
-import org.apache.streampark.registry.core.repository.JdbcRegistryDataRepository;
-import org.apache.streampark.registry.core.repository.JdbcRegistryLockRepository;
-
-import org.apache.commons.collections4.CollectionUtils;
-
-import com.google.common.collect.Lists;
-import lombok.SneakyThrows;
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-/**
- * The JdbcRegistryServer will manage the client, once a client is disconnected, the server will remove the client from the registry, and remove it's related data and lock.
- */
-@Slf4j
-public class JdbcRegistryServer implements IJdbcRegistryServer {
-
- private final JdbcRegistryProperties jdbcRegistryProperties;
-
- private final JdbcRegistryDataRepository jdbcRegistryDataRepository;
-
- private final JdbcRegistryLockRepository jdbcRegistryLockRepository;
-
- private final JdbcRegistryClientRepository jdbcRegistryClientRepository;
-
- private final JdbcRegistryDataManager jdbcRegistryDataManager;
-
- private final JdbcRegistryLockManager jdbcRegistryLockManager;
-
- private JdbcRegistryServerState jdbcRegistryServerState;
-
- private final List<IJdbcRegistryClient> jdbcRegistryClients = new CopyOnWriteArrayList<>();
-
- private final List<ConnectionStateListener> connectionStateListeners = new CopyOnWriteArrayList<>();
-
- private final Map<JdbcRegistryClientIdentify, JdbcRegistryClientHeartbeatDTO> jdbcRegistryClientDTOMap =
- new ConcurrentHashMap<>();
-
- private Long lastSuccessHeartbeat;
-
- public JdbcRegistryServer(JdbcRegistryDataRepository jdbcRegistryDataRepository,
- JdbcRegistryLockRepository jdbcRegistryLockRepository,
- JdbcRegistryClientRepository jdbcRegistryClientRepository,
- JdbcRegistryDataChanceEventRepository jdbcRegistryDataChanceEventRepository,
- JdbcRegistryProperties jdbcRegistryProperties) {
- this.jdbcRegistryDataRepository = checkNotNull(jdbcRegistryDataRepository);
- this.jdbcRegistryLockRepository = checkNotNull(jdbcRegistryLockRepository);
- this.jdbcRegistryClientRepository = checkNotNull(jdbcRegistryClientRepository);
- this.jdbcRegistryProperties = checkNotNull(jdbcRegistryProperties);
- this.jdbcRegistryDataManager = new JdbcRegistryDataManager(
- jdbcRegistryProperties, jdbcRegistryDataRepository, jdbcRegistryDataChanceEventRepository);
- this.jdbcRegistryLockManager = new JdbcRegistryLockManager(
- jdbcRegistryProperties, jdbcRegistryLockRepository);
- this.jdbcRegistryServerState = JdbcRegistryServerState.INIT;
- lastSuccessHeartbeat = System.currentTimeMillis();
- }
-
- @Override
- public void start() {
- if (jdbcRegistryServerState != JdbcRegistryServerState.INIT) {
- // The server is already started or stopped, will not start again.
- return;
- }
- // Purge the previous client to avoid the client is still in the registry.
- purgePreviousJdbcRegistryClient();
- // Start the Purge thread
- // The Purge thread will remove the client from the registry, and remove it's related data and lock.
- // Connect to the database, load the data and lock.
- purgeDeadJdbcRegistryClient();
- JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor()
- .scheduleWithFixedDelay(this::purgeDeadJdbcRegistryClient,
- jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis(),
- jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis(),
- TimeUnit.MILLISECONDS);
- jdbcRegistryDataManager.start();
- jdbcRegistryServerState = JdbcRegistryServerState.STARTED;
- doTriggerOnConnectedListener();
- JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor()
- .scheduleWithFixedDelay(this::refreshClientsHeartbeat,
- 0,
- jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis(),
- TimeUnit.MILLISECONDS);
- }
-
- @SneakyThrows
- @Override
- public void registerClient(IJdbcRegistryClient jdbcRegistryClient) {
- checkNotNull(jdbcRegistryClient);
-
- JdbcRegistryClientIdentify jdbcRegistryClientIdentify = jdbcRegistryClient.getJdbcRegistryClientIdentify();
- checkNotNull(jdbcRegistryClientIdentify);
-
- JdbcRegistryClientHeartbeatDTO registryClientDTO = JdbcRegistryClientHeartbeatDTO.builder()
- .id(jdbcRegistryClientIdentify.getClientId())
- .clientName(jdbcRegistryClientIdentify.getClientName())
- .clientConfig(
- new JdbcRegistryClientHeartbeatDTO.ClientConfig(
- jdbcRegistryProperties.getSessionTimeout().toMillis()))
- .createTime(new Date())
- .lastHeartbeatTime(System.currentTimeMillis())
- .build();
-
- while (jdbcRegistryClientDTOMap.containsKey(jdbcRegistryClientIdentify)) {
- log.warn("The client {} is already exist the registry.", jdbcRegistryClientIdentify.getClientId());
- Thread.sleep(jdbcRegistryProperties.getHeartbeatRefreshInterval().toMillis());
- }
- jdbcRegistryClientRepository.insert(registryClientDTO);
- jdbcRegistryClients.add(jdbcRegistryClient);
- jdbcRegistryClientDTOMap.put(jdbcRegistryClientIdentify, registryClientDTO);
- }
-
- @Override
- public void deregisterClient(IJdbcRegistryClient jdbcRegistryClient) {
- checkNotNull(jdbcRegistryClient);
- jdbcRegistryClients.remove(jdbcRegistryClient);
- jdbcRegistryClientDTOMap.remove(jdbcRegistryClient.getJdbcRegistryClientIdentify());
-
- JdbcRegistryClientIdentify jdbcRegistryClientIdentify = jdbcRegistryClient.getJdbcRegistryClientIdentify();
- checkNotNull(jdbcRegistryClientIdentify);
-
- doPurgeJdbcRegistryClientInDB(Lists.newArrayList(jdbcRegistryClientIdentify.getClientId()));
- }
-
- @Override
- public JdbcRegistryServerState getServerState() {
- return jdbcRegistryServerState;
- }
-
- @Override
- public void subscribeConnectionStateChange(ConnectionStateListener connectionStateListener) {
- checkNotNull(connectionStateListener);
- connectionStateListeners.add(connectionStateListener);
- }
-
- @Override
- public void subscribeJdbcRegistryDataChange(JdbcRegistryDataChangeListener jdbcRegistryDataChangeListener) {
- checkNotNull(jdbcRegistryDataChangeListener);
- jdbcRegistryDataManager.subscribeRegistryRowChange(
- new IRegistryRowChangeNotifier.RegistryRowChangeListener<JdbcRegistryDataDTO>() {
-
- @Override
- public void onRegistryRowUpdated(JdbcRegistryDataDTO data) {
- jdbcRegistryDataChangeListener.onJdbcRegistryDataChanged(data.getDataKey(),
- data.getDataValue());
- }
-
- @Override
- public void onRegistryRowAdded(JdbcRegistryDataDTO data) {
- jdbcRegistryDataChangeListener.onJdbcRegistryDataAdded(data.getDataKey(), data.getDataValue());
- }
-
- @Override
- public void onRegistryRowDeleted(JdbcRegistryDataDTO data) {
- jdbcRegistryDataChangeListener.onJdbcRegistryDataDeleted(data.getDataKey());
- }
- });
- }
-
- @Override
- public boolean existJdbcRegistryDataKey(String key) {
- return jdbcRegistryDataManager.existKey(key);
- }
-
- @Override
- public Optional<JdbcRegistryDataDTO> getJdbcRegistryDataByKey(String key) {
- return jdbcRegistryDataManager.getRegistryDataByKey(key);
- }
-
- @Override
- public List<JdbcRegistryDataDTO> listJdbcRegistryDataChildren(String key) {
- return jdbcRegistryDataManager.listJdbcRegistryDataChildren(key);
- }
-
- @Override
- public void putJdbcRegistryData(Long clientId, String key, String value, DataType dataType) {
- jdbcRegistryDataManager.putJdbcRegistryData(clientId, key, value, dataType);
- }
-
- @Override
- public void deleteJdbcRegistryDataByKey(String key) {
- jdbcRegistryDataManager.deleteJdbcRegistryDataByKey(key);
- }
-
- @Override
- public void acquireJdbcRegistryLock(Long clientId, String lockKey) {
- try {
- jdbcRegistryLockManager.acquireJdbcRegistryLock(clientId, lockKey);
- } catch (Exception ex) {
- throw new RegistryException("Acquire the lock: " + lockKey + " error", ex);
- }
- }
-
- @Override
- public boolean acquireJdbcRegistryLock(Long clientId, String lockKey, long timeout) {
- try {
- return jdbcRegistryLockManager.acquireJdbcRegistryLock(clientId, lockKey, timeout);
- } catch (Exception ex) {
- throw new RegistryException("Acquire the lock: " + lockKey + " error", ex);
- }
- }
-
- @Override
- public void releaseJdbcRegistryLock(Long clientId, String lockKey) {
- try {
- jdbcRegistryLockManager.releaseJdbcRegistryLock(clientId, lockKey);
- } catch (Exception ex) {
- throw new RegistryException("Release the lock: " + lockKey + " error", ex);
- }
- }
-
- @Override
- public void close() {
- jdbcRegistryServerState = JdbcRegistryServerState.STOPPED;
- JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor().shutdown();
- List<Long> clientIds = jdbcRegistryClients.stream()
- .map(IJdbcRegistryClient::getJdbcRegistryClientIdentify)
- .map(JdbcRegistryClientIdentify::getClientId)
- .collect(Collectors.toList());
- doPurgeJdbcRegistryClientInDB(clientIds);
- jdbcRegistryClients.clear();
- jdbcRegistryClientDTOMap.clear();
- }
-
- private void purgePreviousJdbcRegistryClient() {
- if (jdbcRegistryServerState == JdbcRegistryServerState.STOPPED) {
- return;
- }
- List<Long> previousJdbcRegistryClientIds = jdbcRegistryClientRepository.queryAll()
- .stream()
- .filter(jdbcRegistryClientHeartbeat -> jdbcRegistryClientHeartbeat.getClientName()
- .equals(jdbcRegistryProperties.getJdbcRegistryClientName()))
- .map(JdbcRegistryClientHeartbeatDTO::getId)
- .collect(Collectors.toList());
- doPurgeJdbcRegistryClientInDB(previousJdbcRegistryClientIds);
-
- }
-
- private void purgeDeadJdbcRegistryClient() {
- if (jdbcRegistryServerState == JdbcRegistryServerState.STOPPED) {
- return;
- }
- List<Long> deadJdbcRegistryClientIds = jdbcRegistryClientRepository.queryAll()
- .stream()
- .filter(JdbcRegistryClientHeartbeatDTO::isDead)
- .map(JdbcRegistryClientHeartbeatDTO::getId)
- .collect(Collectors.toList());
- doPurgeJdbcRegistryClientInDB(deadJdbcRegistryClientIds);
-
- }
-
- private void doPurgeJdbcRegistryClientInDB(List<Long> jdbcRegistryClientIds) {
- if (CollectionUtils.isEmpty(jdbcRegistryClientIds)) {
- return;
- }
- log.info("Begin to delete dead jdbcRegistryClient: {}", jdbcRegistryClientIds);
- jdbcRegistryDataRepository.deleteEphemeralDateByClientIds(jdbcRegistryClientIds);
- jdbcRegistryLockRepository.deleteByClientIds(jdbcRegistryClientIds);
- jdbcRegistryClientRepository.deleteByIds(jdbcRegistryClientIds);
- log.info("Success delete dead jdbcRegistryClient: {}", jdbcRegistryClientIds);
- }
-
- private void refreshClientsHeartbeat() {
- if (CollectionUtils.isEmpty(jdbcRegistryClients)) {
- return;
- }
- if (jdbcRegistryServerState == JdbcRegistryServerState.STOPPED) {
- log.warn("The JdbcRegistryServer is STOPPED, will not refresh clients: {} heartbeat.",
- CollectionUtils.collect(jdbcRegistryClients, IJdbcRegistryClient::getJdbcRegistryClientIdentify));
- return;
- }
- // Refresh the client's term
- try {
- long now = System.currentTimeMillis();
- for (IJdbcRegistryClient jdbcRegistryClient : jdbcRegistryClients) {
- JdbcRegistryClientHeartbeatDTO jdbcRegistryClientHeartbeatDTO =
- jdbcRegistryClientDTOMap.get(jdbcRegistryClient.getJdbcRegistryClientIdentify());
- if (jdbcRegistryClientHeartbeatDTO == null) {
- // This may occur when the data in db has been deleted, but the client is still alive.
- log.error(
- "The client {} is not found in the registry, will not refresh it's term. (This may happen when the client is removed from the db)",
- jdbcRegistryClient.getJdbcRegistryClientIdentify().getClientId());
- continue;
- }
- JdbcRegistryClientHeartbeatDTO clone = jdbcRegistryClientHeartbeatDTO.clone();
- clone.setLastHeartbeatTime(now);
- jdbcRegistryClientRepository.updateById(jdbcRegistryClientHeartbeatDTO);
- jdbcRegistryClientHeartbeatDTO.setLastHeartbeatTime(clone.getLastHeartbeatTime());
- }
- if (jdbcRegistryServerState == JdbcRegistryServerState.SUSPENDED) {
- jdbcRegistryServerState = JdbcRegistryServerState.STARTED;
- doTriggerReconnectedListener();
- }
- lastSuccessHeartbeat = now;
- log.debug("Success refresh clients: {} heartbeat.",
- CollectionUtils.collect(jdbcRegistryClients, IJdbcRegistryClient::getJdbcRegistryClientIdentify));
- } catch (Exception ex) {
- log.error("Failed to refresh the client's term", ex);
- switch (jdbcRegistryServerState) {
- case STARTED:
- jdbcRegistryServerState = JdbcRegistryServerState.SUSPENDED;
- break;
- case SUSPENDED:
- if (System.currentTimeMillis() - lastSuccessHeartbeat > jdbcRegistryProperties.getSessionTimeout()
- .toMillis()) {
- jdbcRegistryServerState = JdbcRegistryServerState.DISCONNECTED;
- doTriggerOnDisConnectedListener();
- }
- break;
- default:
- break;
- }
- }
- }
-
- private void doTriggerReconnectedListener() {
- log.info("Trigger:onReconnected listener.");
- connectionStateListeners.forEach(listener -> {
- try {
- listener.onReconnected();
- } catch (Exception ex) {
- log.error("Trigger:onReconnected failed", ex);
- }
- });
- }
-
- private void doTriggerOnConnectedListener() {
- log.info("Trigger:onConnected listener.");
- connectionStateListeners.forEach(listener -> {
- try {
- listener.onConnected();
- } catch (Exception ex) {
- log.error("Trigger:onConnected failed", ex);
- }
- });
- }
-
- private void doTriggerOnDisConnectedListener() {
- log.info("Trigger:onDisConnected listener.");
- connectionStateListeners.forEach(listener -> {
- try {
- listener.onDisConnected();
- } catch (Exception ex) {
- log.error("Trigger:onDisConnected failed", ex);
- }
- });
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryServerState.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryServerState.java
deleted file mode 100644
index e568e7d8a..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/java/org/apache/streampark/registry/core/server/JdbcRegistryServerState.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.registry.core.server;
-
-import org.apache.streampark.registry.core.JdbcRegistryProperties;
-
-public enum JdbcRegistryServerState {
- /**
- * Once the {@link JdbcRegistryServer} created, it will be in the INIT state.
- */
- INIT,
- /**
- * After the {@link JdbcRegistryServer} started, it will be in the STARTED state.
- */
- STARTED,
- /**
- * Once the {@link JdbcRegistryServer} cannot connect to DB, it will be in the SUSPENDED state, and if it can reconnect to DB in {@link JdbcRegistryProperties#getSessionTimeout()} ()},
- * it will be changed to the STARTED state again.
- */
- SUSPENDED,
- /**
- * If the {@link JdbcRegistryServer} cannot connected to DB in {@link JdbcRegistryProperties#getSessionTimeout()}, it will be in the DISCONNECTED state.
- */
- DISCONNECTED,
- /**
- * If the {@link JdbcRegistryServer} closed, it will be in the STOPPED state.
- */
- STOPPED,
- ;
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/resources/META-INF/spring.factories b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/resources/META-INF/spring.factories
deleted file mode 100644
index a4f56a18a..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/resources/META-INF/spring.factories
+++ /dev/null
@@ -1,19 +0,0 @@
-#
-# 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.
-#
-
-org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
- org.apache.streampark.registry.core.JdbcRegistryAutoConfiguration
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/resources/mysql_registry_init.sql b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/resources/mysql_registry_init.sql
deleted file mode 100644
index 4ee6c47a5..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/resources/mysql_registry_init.sql
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.
-*/
-
-
-DROP TABLE IF EXISTS `t_jdbc_registry_data`;
-CREATE TABLE `t_jdbc_registry_data`
-(
- `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT 'primary key',
- `data_key` varchar(256) NOT NULL COMMENT 'key, like zookeeper node path',
- `data_value` text NOT NULL COMMENT 'data, like zookeeper node value',
- `data_type` varchar(64) NOT NULL COMMENT 'EPHEMERAL, PERSISTENT',
- `client_id` bigint(11) NOT NULL COMMENT 'client id',
- `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'create time',
- `last_update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'last update time',
- PRIMARY KEY (`id`),
- unique Key `uk_t_jdbc_registry_dataKey` (`data_key`)
-) ENGINE = InnoDB
- DEFAULT CHARSET = utf8;
-
-
-DROP TABLE IF EXISTS `t_jdbc_registry_lock`;
-CREATE TABLE `t_jdbc_registry_lock`
-(
- `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT 'primary key',
- `lock_key` varchar(256) NOT NULL COMMENT 'lock path',
- `lock_owner` varchar(256) NOT NULL COMMENT 'the lock owner, ip_processId',
- `client_id` bigint(11) NOT NULL COMMENT 'client id',
- `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'create time',
- PRIMARY KEY (`id`),
- unique Key `uk_t_jdbc_registry_lockKey` (`lock_key`)
-) ENGINE = InnoDB
- DEFAULT CHARSET = utf8;
-
-DROP TABLE IF EXISTS `t_jdbc_registry_client_heartbeat`;
-CREATE TABLE `t_jdbc_registry_client_heartbeat`
-(
- `id` bigint(11) NOT NULL COMMENT 'primary key',
- `client_name` varchar(256) NOT NULL COMMENT 'client name, ip_processId',
- `last_heartbeat_time` bigint(11) NOT NULL COMMENT 'last heartbeat timestamp',
- `connection_config` text NOT NULL COMMENT 'connection config',
- `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'create time',
- PRIMARY KEY (`id`)
-) ENGINE = InnoDB
- DEFAULT CHARSET = utf8;
-
-DROP TABLE IF EXISTS `t_jdbc_registry_data_change_event`;
-CREATE TABLE `t_jdbc_registry_data_change_event`
-(
- `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT 'primary key',
- `event_type` varchar(64) NOT NULL COMMENT 'ADD, UPDATE, DELETE',
- `jdbc_registry_data` text NOT NULL COMMENT 'jdbc registry data',
- `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'create time',
- PRIMARY KEY (`id`)
-) ENGINE = InnoDB
- DEFAULT CHARSET = utf8;
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/resources/postgresql_registry_init.sql b/streampark-console/streampark-console-registry/streampark-registry-core/src/main/resources/postgresql_registry_init.sql
deleted file mode 100644
index dcfd82074..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/main/resources/postgresql_registry_init.sql
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.
-*/
-
-DROP TABLE IF EXISTS t_jdbc_registry_data;
-create table t_jdbc_registry_data
-(
- id bigserial not null,
- data_key varchar not null,
- data_value text not null,
- data_type varchar not null,
- client_id bigint not null,
- create_time timestamp not null default current_timestamp,
- last_update_time timestamp not null default current_timestamp,
- primary key (id)
-);
-create unique index uk_t_jdbc_registry_dataKey on t_jdbc_registry_data (data_key);
-
-
-DROP TABLE IF EXISTS t_jdbc_registry_lock;
-create table t_jdbc_registry_lock
-(
- id bigserial not null,
- lock_key varchar not null,
- lock_owner varchar not null,
- client_id bigint not null,
- create_time timestamp not null default current_timestamp,
- primary key (id)
-);
-create unique index uk_t_jdbc_registry_lockKey on t_jdbc_registry_lock (lock_key);
-
-
-DROP TABLE IF EXISTS t_jdbc_registry_client_heartbeat;
-create table t_jdbc_registry_client_heartbeat
-(
- id bigint not null,
- client_name varchar not null,
- last_heartbeat_time bigint not null,
- connection_config text not null,
- create_time timestamp not null default current_timestamp,
- primary key (id)
-);
-
-DROP TABLE IF EXISTS t_jdbc_registry_data_change_event;
-create table t_jdbc_registry_data_change_event
-(
- id bigserial not null,
- event_type varchar not null,
- jdbc_registry_data text not null,
- create_time timestamp not null default current_timestamp,
- primary key (id)
-);
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/JdbcRegistryTestCase.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/JdbcRegistryTestCase.java
deleted file mode 100644
index 75b876d38..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/JdbcRegistryTestCase.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.apache.streampark.common.CommonConfiguration;
-import org.apache.streampark.registry.RegistryTestCase;
-import org.apache.streampark.registry.api.ConnectionState;
-import org.apache.streampark.registry.core.server.IJdbcRegistryServer;
-
-import lombok.SneakyThrows;
-import org.junit.jupiter.api.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.annotation.DirtiesContext;
-
-import java.util.concurrent.atomic.AtomicReference;
-
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-
-@SpringBootTest(classes = {CommonConfiguration.class, JdbcRegistryProperties.class})
-@SpringBootApplication(scanBasePackageClasses = {CommonConfiguration.class, JdbcRegistryProperties.class})
-@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD)
-public abstract class JdbcRegistryTestCase extends RegistryTestCase<JdbcRegistry> {
-
- @Autowired
- private JdbcRegistryProperties jdbcRegistryProperties;
-
- @Autowired
- private IJdbcRegistryServer jdbcRegistryServer;
-
- @SneakyThrows
- @Test
- public void testAddConnectionStateListener() {
-
- AtomicReference<ConnectionState> connectionState = new AtomicReference<>();
- registry.addConnectionStateListener(connectionState::set);
-
- // todo: since the jdbc registry is started at the auto configuration, the stateListener is added after the
- // registry is started.
- assertThat(connectionState.get()).isEqualTo(null);
- }
-
- @Override
- public JdbcRegistry createRegistry() {
- return new JdbcRegistry(jdbcRegistryProperties, jdbcRegistryServer);
- }
-
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/JdbcRegistryThreadFactoryTest.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/JdbcRegistryThreadFactoryTest.java
deleted file mode 100644
index 6686d3987..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/JdbcRegistryThreadFactoryTest.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
-
-class JdbcRegistryThreadFactoryTest {
-
- @Test
- void getDefaultSchedulerThreadExecutor() {
- ScheduledExecutorService schedulerThreadExecutor =
- JdbcRegistryThreadFactory.getDefaultSchedulerThreadExecutor();
- AtomicInteger atomicInteger = new AtomicInteger(0);
- for (int i = 0; i < 100; i++) {
- schedulerThreadExecutor.scheduleWithFixedDelay(atomicInteger::incrementAndGet, 0, 1, TimeUnit.SECONDS);
- }
- await()
- .atMost(10, TimeUnit.SECONDS)
- .untilAsserted(() -> assertThat(atomicInteger.get()).isEqualTo(100));
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/LockUtilsTest.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/LockUtilsTest.java
deleted file mode 100644
index 7d2b98e46..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/LockUtilsTest.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.apache.streampark.common.utils.NetworkUtils;
-
-import org.junit.jupiter.api.Test;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.TestPropertySource;
-
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-
-@SpringBootTest
-@ContextConfiguration(classes = {NetworkUtils.class})
-@TestPropertySource(locations = "classpath:application.yaml")
-class LockUtilsTest {
-
- @Test
- void getLockOwner() {
- assertThat(LockUtils.getLockOwner()).isNotNull();
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/MysqlJdbcRegistryTestCase.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/MysqlJdbcRegistryTestCase.java
deleted file mode 100644
index 262df31d4..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/MysqlJdbcRegistryTestCase.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.apache.streampark.registry.api.sql.SqlScriptRunner;
-
-import com.zaxxer.hikari.HikariConfig;
-import com.zaxxer.hikari.HikariDataSource;
-import lombok.SneakyThrows;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.springframework.test.context.ActiveProfiles;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.MySQLContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.wait.strategy.Wait;
-import org.testcontainers.lifecycle.Startables;
-import org.testcontainers.utility.DockerImageName;
-
-import java.time.Duration;
-import java.util.stream.Stream;
-
-@ActiveProfiles("mysql")
-class MysqlJdbcRegistryTestCase extends JdbcRegistryTestCase {
-
- private static GenericContainer<?> mysqlContainer;
-
- @SneakyThrows
- @BeforeAll
- public static void setUpTestingServer() {
- mysqlContainer = new MySQLContainer(DockerImageName.parse("mysql:8.0"))
- .withUsername("root")
- .withPassword("root")
- .withDatabaseName("streampark")
- .withNetwork(Network.newNetwork())
- .withExposedPorts(3306)
- .waitingFor(Wait.forHealthcheck().withStartupTimeout(Duration.ofSeconds(300)));
-
- Startables.deepStart(Stream.of(mysqlContainer)).join();
-
- String jdbcUrl = "jdbc:mysql://localhost:" + mysqlContainer.getMappedPort(3306)
- + "/streampark?useSSL=false&serverTimezone=UTC";
- System.clearProperty("spring.datasource.url");
- System.setProperty("spring.datasource.url", jdbcUrl);
-
- HikariConfig config = new HikariConfig();
- config.setJdbcUrl(jdbcUrl);
- config.setUsername("root");
- config.setPassword("root");
-
- try (HikariDataSource dataSource = new HikariDataSource(config)) {
- SqlScriptRunner sqlScriptRunner = new SqlScriptRunner(dataSource, "mysql_registry_init.sql");
- sqlScriptRunner.execute();
- }
- }
-
- @SneakyThrows
- @AfterAll
- public static void tearDownTestingServer() {
- mysqlContainer.close();
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/PostgresqlJdbcRegistryTestCase.java b/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/PostgresqlJdbcRegistryTestCase.java
deleted file mode 100644
index 1e87d284c..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/java/org/apache/streampark/registry/core/PostgresqlJdbcRegistryTestCase.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.registry.core;
-
-import org.apache.streampark.registry.api.sql.SqlScriptRunner;
-
-import com.zaxxer.hikari.HikariConfig;
-import com.zaxxer.hikari.HikariDataSource;
-import lombok.SneakyThrows;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.springframework.test.context.ActiveProfiles;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.PostgreSQLContainer;
-import org.testcontainers.lifecycle.Startables;
-import org.testcontainers.utility.DockerImageName;
-
-import java.util.stream.Stream;
-
-@ActiveProfiles("postgresql")
-public class PostgresqlJdbcRegistryTestCase extends JdbcRegistryTestCase {
-
- private static GenericContainer<?> postgresqlContainer;
-
- @SneakyThrows
- @BeforeAll
- public static void setUpTestingServer() {
- postgresqlContainer = new PostgreSQLContainer(DockerImageName.parse("postgres:16.0"))
- .withUsername("root")
- .withPassword("root")
- .withDatabaseName("streampark")
- .withNetwork(Network.newNetwork())
- .withExposedPorts(5432);
-
- Startables.deepStart(Stream.of(postgresqlContainer)).join();
-
- String jdbcUrl = "jdbc:postgresql://localhost:" + postgresqlContainer.getMappedPort(5432) + "/streampark";
- System.clearProperty("spring.datasource.url");
- System.setProperty("spring.datasource.url", jdbcUrl);
-
- HikariConfig config = new HikariConfig();
- config.setJdbcUrl(jdbcUrl);
- config.setUsername("root");
- config.setPassword("root");
-
- try (HikariDataSource dataSource = new HikariDataSource(config)) {
- SqlScriptRunner sqlScriptRunner = new SqlScriptRunner(dataSource, "postgresql_registry_init.sql");
- sqlScriptRunner.execute();
- }
- }
-
- @SneakyThrows
- @AfterAll
- public static void tearDownTestingServer() {
- postgresqlContainer.close();
- }
-}
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/application-mysql.yaml b/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/application-mysql.yaml
deleted file mode 100644
index 779b43a2c..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/application-mysql.yaml
+++ /dev/null
@@ -1,38 +0,0 @@
-#
-# 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.
-#
-spring:
- sql:
- init:
- schema-locations: classpath:mysql_registry_init.sql
- datasource:
- driver-class-name: com.mysql.cj.jdbc.Driver
- url: jdbc:mysql://localhost:3306/streampark?useUnicode=true&characterEncoding=UTF-8
- username: root
- password: root
-
-registry:
- type: jdbc
- heartbeat-refresh-interval: 1s
- session-timeout: 3s
-
-network:
- # network interface preferred like eth0, default: empty
- preferred-interface: ""
- # network interface restricted like docker0, default: empty
- restrict-interface: docker0
- # network IP gets priority, default inner outer
- priority-strategy: default
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/application-postgresql.yaml b/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/application-postgresql.yaml
deleted file mode 100644
index e5fbb5a56..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/application-postgresql.yaml
+++ /dev/null
@@ -1,36 +0,0 @@
-#
-# 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.
-#
-spring:
- datasource:
- driver-class-name: org.postgresql.Driver
- url: jdbc:postgresql://localhost:5432/streampark
- username: root
- password: root
-
-registry:
- type: jdbc
- heartbeat-refresh-interval: 1s
- session-timeout: 3s
-
-network:
- # network interface preferred like eth0, default: empty
- preferred-interface: ""
- # network interface restricted like docker0, default: empty
- restrict-interface: docker0
- # network IP gets priority, default inner outer
- priority-strategy: default
-
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/application.yaml b/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/application.yaml
deleted file mode 100644
index 7f828bf58..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/application.yaml
+++ /dev/null
@@ -1,24 +0,0 @@
-#
-# 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.
-#
-
-network:
- # network interface preferred like eth0, default: empty
- preferred-interface: ""
- # network interface restricted like docker0, default: empty
- restrict-interface: docker0
- # network IP gets priority, default inner outer
- priority-strategy: default
diff --git a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/logback.xml b/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/logback.xml
deleted file mode 100644
index 6f211959c..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-core/src/test/resources/logback.xml
+++ /dev/null
@@ -1,21 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ 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.
- -->
-
-<configuration>
- <logger name="*" level="ERROR"/>
-</configuration>
diff --git a/streampark-console/streampark-console-registry/streampark-registry-it/pom.xml b/streampark-console/streampark-console-registry/streampark-registry-it/pom.xml
deleted file mode 100644
index 6fce49e99..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-it/pom.xml
+++ /dev/null
@@ -1,62 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ 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.
- -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-console-registry</artifactId>
- <version>2.2.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>streampark-registry-it</artifactId>
- <name>StreamPark : Console Registry Testcase</name>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-registry-api</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-test</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <configuration>
- <skip>false</skip>
- </configuration>
- <executions>
- <execution>
- <goals>
- <goal>test-jar</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-
-</project>
diff --git a/streampark-console/streampark-console-registry/streampark-registry-it/src/test/java/org/apache/streampark/plugin/registry/RegistryTestCase.java b/streampark-console/streampark-console-registry/streampark-registry-it/src/test/java/org/apache/streampark/plugin/registry/RegistryTestCase.java
deleted file mode 100644
index 63ec42936..000000000
--- a/streampark-console/streampark-console-registry/streampark-registry-it/src/test/java/org/apache/streampark/plugin/registry/RegistryTestCase.java
+++ /dev/null
@@ -1,260 +0,0 @@
-/*
- * 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.registry;
-
-import org.apache.streampark.common.utils.NetworkUtils;
-import org.apache.streampark.registry.api.ConnectionState;
-import org.apache.streampark.registry.api.Event;
-import org.apache.streampark.registry.api.Registry;
-import org.apache.streampark.registry.api.RegistryException;
-import org.apache.streampark.registry.api.SubscribeListener;
-
-import lombok.SneakyThrows;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.springframework.test.util.ReflectionTestUtils;
-
-import java.time.Duration;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
-
-public abstract class RegistryTestCase<R extends Registry> {
-
- protected R registry;
-
- @BeforeAll
- public static void init() {
- ReflectionTestUtils.setField(NetworkUtils.class, "priorityStrategy", "default");
- }
-
- @BeforeEach
- public void setupRegistry() {
- registry = createRegistry();
- }
-
- @SneakyThrows
- @AfterEach
- public void tearDownRegistry() {
- this.registry.close();
- }
-
- @Test
- public void testIsConnected() {
- registry.start();
- assertThat(registry.isConnected()).isTrue();
- }
-
- @Test
- public void testConnectUntilTimeout() {
- registry.start();
- await().atMost(Duration.ofSeconds(10))
- .untilAsserted(() -> registry.connectUntilTimeout(Duration.ofSeconds(3)));
-
- }
-
- @SneakyThrows
- @Test
- public void testSubscribe() {
- registry.start();
-
- final AtomicBoolean subscribeAdded = new AtomicBoolean(false);
- final AtomicBoolean subscribeRemoved = new AtomicBoolean(false);
- final AtomicBoolean subscribeUpdated = new AtomicBoolean(false);
-
- SubscribeListener subscribeListener = event -> {
- System.out.println("Receive event: " + event);
- if (event.type() == Event.Type.ADD) {
- subscribeAdded.compareAndSet(false, true);
- }
- if (event.type() == Event.Type.REMOVE) {
- subscribeRemoved.compareAndSet(false, true);
- }
- if (event.type() == Event.Type.UPDATE) {
- subscribeUpdated.compareAndSet(false, true);
- }
- };
- String key = "/nodes/console-server" + System.nanoTime();
- registry.subscribe(key, subscribeListener);
- registry.put(key, String.valueOf(System.nanoTime()), true);
- // Sleep 3 seconds here since in mysql jdbc registry
- // If multiple event occurs in a refresh time, only the last event will be triggered
- Thread.sleep(3000);
- registry.put(key, String.valueOf(System.nanoTime()), true);
- Thread.sleep(3000);
- registry.delete(key);
-
- await().atMost(Duration.ofSeconds(10))
- .untilAsserted(() -> {
- Assertions.assertTrue(subscribeAdded.get());
- Assertions.assertTrue(subscribeUpdated.get());
- Assertions.assertTrue(subscribeRemoved.get());
- });
- }
-
- @SneakyThrows
- @Test
- public void testAddConnectionStateListener() {
-
- AtomicReference<ConnectionState> connectionState = new AtomicReference<>();
- registry.addConnectionStateListener(connectionState::set);
-
- assertThat(connectionState.get()).isNull();
- registry.start();
-
- await().atMost(Duration.ofSeconds(2))
- .until(() -> ConnectionState.CONNECTED == connectionState.get());
-
- }
-
- @Test
- public void testGet() {
- registry.start();
- String key = "/nodes/console-server" + System.nanoTime();
- String value = "127.0.0.1:8080";
- assertThrows(RegistryException.class, () -> registry.get(key));
- registry.put(key, value, true);
- assertThat(registry.get(key)).isEqualTo(value);
- }
-
- @Test
- public void testPut() {
- registry.start();
- String key = "/nodes/console-server" + System.nanoTime();
- String value = "127.0.0.1:8080";
- registry.put(key, value, true);
- assertThat(registry.get(key)).isEqualTo(value);
-
- // Update the value
- registry.put(key, "123", true);
- assertThat(registry.get(key)).isEqualTo("123");
- }
-
- @Test
- public void testDelete() {
- registry.start();
- String key = "/nodes/console-server" + System.nanoTime();
- String value = "127.0.0.1:8080";
- // Delete a non-existent key
- registry.delete(key);
-
- registry.put(key, value, true);
- assertThat(registry.get(key)).isEqualTo(value);
- registry.delete(key);
- assertThat(registry.exists(key)).isFalse();
-
- }
-
- @Test
- public void testChildren() {
- registry.start();
- String master1 = "/nodes/children/childGroup1/127.0.0.1:8080";
- String master2 = "/nodes/children/childGroup1/127.0.0.2:8080";
- String value = "123";
- registry.put(master1, value, true);
- registry.put(master2, value, true);
- assertThat(registry.children("/nodes/children")).contains("childGroup1");
- assertThat(registry.children("/nodes/children/childGroup1")).contains("127.0.0.1:8080",
- "127.0.0.2:8080");
- }
-
- @Test
- public void testExists() {
- registry.start();
- String key = "/nodes/console-server" + System.nanoTime();
- String value = "123";
- assertThat(registry.exists(key)).isFalse();
- registry.put(key, value, true);
- assertThat(registry.exists(key)).isTrue();
-
- }
-
- @SneakyThrows
- @Test
- public void testAcquireLock() {
- registry.start();
- String lockKey = "/lock" + System.nanoTime();
-
- // 1. Acquire the lock at the main thread
- assertThat(registry.acquireLock(lockKey)).isTrue();
- // Acquire the lock at the main thread again
- // It should acquire success
- assertThat(registry.acquireLock(lockKey)).isTrue();
-
- // Acquire the lock at another thread
- // It should acquire failed
- CompletableFuture<Boolean> acquireResult = CompletableFuture.supplyAsync(() -> registry.acquireLock(lockKey));
- assertThrows(TimeoutException.class, () -> acquireResult.get(3000, TimeUnit.MILLISECONDS));
-
- }
-
- @SneakyThrows
- @Test
- public void testAcquireLock_withTimeout() {
- registry.start();
- String lockKey = "/lock" + System.nanoTime();
- // 1. Acquire the lock in the main thread
- assertThat(registry.acquireLock(lockKey, 3000)).isTrue();
-
- // Acquire the lock in the main thread
- // It should acquire success
- assertThat(registry.acquireLock(lockKey, 3000)).isTrue();
-
- // Acquire the lock at another thread
- // It should acquire failed
- CompletableFuture<Boolean> acquireResult =
- CompletableFuture.supplyAsync(() -> registry.acquireLock(lockKey, 3000));
- assertThat(acquireResult.get()).isFalse();
-
- }
-
- @SneakyThrows
- @Test
- public void testReleaseLock() {
- registry.start();
- String lockKey = "/lock" + System.nanoTime();
- // 1. Acquire the lock in the main thread
- assertThat(registry.acquireLock(lockKey, 3000)).isTrue();
-
- // Acquire the lock at another thread
- // It should acquire failed
- CompletableFuture<Boolean> acquireResult =
- CompletableFuture.supplyAsync(() -> registry.acquireLock(lockKey, 3000));
- assertThat(acquireResult.get()).isFalse();
-
- // 2. Release the lock in the main thread
- assertThat(registry.releaseLock(lockKey)).isTrue();
-
- // Acquire the lock at another thread
- // It should acquire success
- acquireResult = CompletableFuture.supplyAsync(() -> registry.acquireLock(lockKey, 3000));
- assertThat(acquireResult.get()).isTrue();
- }
-
- public abstract R createRegistry();
-
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/StreamParkConsoleBootstrap.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/StreamParkConsoleBootstrap.java
index 3dfea1344..30c9317ef 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/StreamParkConsoleBootstrap.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/StreamParkConsoleBootstrap.java
@@ -19,13 +19,8 @@ package org.apache.streampark.console;
import org.apache.streampark.common.CommonConfiguration;
import org.apache.streampark.common.IStoppable;
-import org.apache.streampark.common.constants.Constants;
import org.apache.streampark.console.base.config.SpringProperties;
import org.apache.streampark.console.base.util.SpringContextUtils;
-import org.apache.streampark.console.core.registry.ConsoleRegistryClient;
-import org.apache.streampark.registry.api.RegistryConfiguration;
-import org.apache.streampark.registry.api.lifecycle.ServerLifeCycleManager;
-import org.apache.streampark.registry.api.thread.ThreadUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@@ -34,8 +29,6 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.annotation.Import;
import org.springframework.scheduling.annotation.EnableScheduling;
-import javax.annotation.PostConstruct;
-
/**
*
*
@@ -58,13 +51,9 @@ import javax.annotation.PostConstruct;
@Slf4j
@SpringBootApplication
@EnableScheduling
-@Import({CommonConfiguration.class,
- RegistryConfiguration.class})
+@Import({CommonConfiguration.class})
public class StreamParkConsoleBootstrap implements IStoppable {
- @Autowired
- private ConsoleRegistryClient consoleRegistryClient;
-
@Autowired
private SpringContextUtils springContextUtils;
@@ -75,47 +64,8 @@ public class StreamParkConsoleBootstrap implements IStoppable {
.run(args);
}
- @PostConstruct
- public void run() {
- consoleRegistryClient.start();
- consoleRegistryClient.setRegistryStoppable(this);
-
- Runtime.getRuntime().addShutdownHook(new Thread(() -> {
- if (!ServerLifeCycleManager.isStopped()) {
- close("ConsoleServer shutdownHook");
- }
- }));
- }
-
- /**
- * gracefully close console server
- *
- * @param cause close cause
- */
- public void close(String cause) {
- // set stop signal is true
- // execute only once
- if (!ServerLifeCycleManager.toStopped()) {
- log.warn("MasterServer is already stopped, current cause: {}", cause);
- return;
- }
- // thread sleep 3 seconds for thread quietly stop
- ThreadUtils.sleep(Constants.SERVER_CLOSE_WAIT_TIME.toMillis());
- try (ConsoleRegistryClient closedMasterRegistryClient = consoleRegistryClient) {
- // todo: close other resources
- springContextUtils.close();
- log.info("Master server is stopping, current cause : {}", cause);
- } catch (Exception e) {
- log.error("MasterServer stop failed, current cause: {}", cause, e);
- return;
- }
- log.info("MasterServer stopped, current cause: {}", cause);
- }
-
@Override
public void stop(String cause) {
- close(cause);
-
// make sure exit after server closed, don't call System.exit in close logic, will cause deadlock if close
// multiple times at the same time
System.exit(1);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/config/ConsoleConfig.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/config/ConsoleConfig.java
index 6f17c038c..3d181cd4e 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/config/ConsoleConfig.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/config/ConsoleConfig.java
@@ -18,8 +18,6 @@
package org.apache.streampark.console.core.config;
import org.apache.streampark.common.utils.NetworkUtils;
-import org.apache.streampark.registry.api.ConnectStrategyProperties;
-import org.apache.streampark.registry.api.enums.RegistryNodeType;
import org.apache.commons.lang3.StringUtils;
@@ -55,8 +53,6 @@ public class ConsoleConfig implements Validator {
@Value("${streampark.max-heartbeat-interval:10}")
private Duration maxHeartbeatInterval = Duration.ofSeconds(10);
- private ConnectStrategyProperties registryDisconnectStrategy = new ConnectStrategyProperties();
-
/**
* The IP address and listening port of the console server in the format 'ip:listenPort'.
*/
@@ -85,9 +81,6 @@ public class ConsoleConfig implements Validator {
consoleConfig.setConsoleAddress(NetworkUtils.getAddr(consoleConfig.getListenPort()));
}
- consoleConfig.setConsoleRegistryPath(
- RegistryNodeType.CONSOLE_SERVER.getRegistryPath() + "/" + consoleConfig.getConsoleAddress());
-
printConfig();
}
@@ -95,10 +88,9 @@ public class ConsoleConfig implements Validator {
String config =
"\n****************************Console Configuration**************************************" +
"\n listen-port -> " + listenPort +
- "\n registry-disconnect-strategy -> " + registryDisconnectStrategy +
"\n console-address -> " + consoleAddress +
"\n console-registry-path: " + consoleRegistryPath +
- "\n****************************Master Configuration**************************************";
+ "\n**************************** Master Configuration**************************************";
log.info(config);
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleConnectStrategy.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleConnectStrategy.java
deleted file mode 100644
index 97bb3aa6b..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleConnectStrategy.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.registry;
-
-import org.apache.streampark.registry.api.ConnectStrategy;
-
-public interface ConsoleConnectStrategy extends ConnectStrategy {
-
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleConnectionStateListener.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleConnectionStateListener.java
deleted file mode 100644
index 0e44e1c50..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleConnectionStateListener.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.registry;
-
-import org.apache.streampark.registry.api.ConnectionListener;
-import org.apache.streampark.registry.api.ConnectionState;
-import org.apache.streampark.registry.api.lifecycle.ServerLifeCycleManager;
-
-import lombok.NonNull;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class ConsoleConnectionStateListener implements ConnectionListener {
-
- private final ConsoleConnectStrategy consoleConnectStrategy;
-
- public ConsoleConnectionStateListener(@NonNull ConsoleConnectStrategy consoleConnectStrategy) {
- this.consoleConnectStrategy = consoleConnectStrategy;
- }
-
- @Override
- public void onUpdate(ConnectionState state) {
- log.info("Master received a {} event from registry, the current server state is {}", state,
- ServerLifeCycleManager.getServerLifeCycle());
- switch (state) {
- case CONNECTED:
- break;
- case SUSPENDED:
- break;
- case RECONNECTED:
- consoleConnectStrategy.reconnect();
- break;
- case DISCONNECTED:
- consoleConnectStrategy.disconnect();
- break;
- default:
- }
- }
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java
deleted file mode 100644
index b9c70be51..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java
+++ /dev/null
@@ -1,203 +0,0 @@
-/*
- * 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.registry;
-
-import org.apache.streampark.common.IStoppable;
-import org.apache.streampark.common.utils.JSONUtils;
-import org.apache.streampark.common.utils.NetworkUtils;
-import org.apache.streampark.console.core.config.ConsoleConfig;
-import org.apache.streampark.console.core.service.DistributedTaskService;
-import org.apache.streampark.console.core.task.ConsoleHeartBeatTask;
-import org.apache.streampark.registry.api.RegistryClient;
-import org.apache.streampark.registry.api.RegistryException;
-import org.apache.streampark.registry.api.enums.RegistryNodeType;
-import org.apache.streampark.registry.api.enums.ServerStatusEnum;
-import org.apache.streampark.registry.api.model.ConsoleHeartBeat;
-import org.apache.streampark.registry.api.thread.ThreadUtils;
-
-import org.apache.commons.lang3.StringUtils;
-
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-import java.util.Set;
-
-import static org.apache.streampark.common.constants.Constants.SLEEP_TIME_MILLIS;
-
-/**
- * <p>streampark console register client, used to connect to registry and hand the registry events.
- * <p>When the console node startup, it will register in registry center. And start a {@link ConsoleHeartBeatTask} to update its metadata in registry.
- */
-@Component
-@Slf4j
-public class ConsoleRegistryClient implements AutoCloseable {
-
- @Autowired
- private RegistryClient registryClient;
-
- @Autowired
- private ConsoleConfig consoleConfig;
-
- @Autowired
- private ConsoleConnectStrategy consoleConnectStrategy;
-
- @Autowired
- private DistributedTaskService distributedTaskService;
-
- private ConsoleHeartBeatTask consoleHeartBeatTask;
-
- public void start() {
- try {
- log.info("consoleConfig: {}", consoleConfig);
- this.consoleHeartBeatTask = new ConsoleHeartBeatTask(consoleConfig, registryClient);
- // console registry
- registry();
- registryClient.addConnectionStateListener(new ConsoleConnectionStateListener(consoleConnectStrategy));
- registryClient.subscribe(RegistryNodeType.ALL_SERVERS.getRegistryPath(), new ConsoleRegistryDataListener());
- } catch (Exception e) {
- throw new RegistryException("Console registry client start up error", e);
- }
- }
-
- public void setRegistryStoppable(IStoppable stoppable) {
- registryClient.setStoppable(stoppable);
- }
-
- @Override
- public void close() {
- // TODO unsubscribe ConsoleRegistryDataListener
- deregister();
- }
-
- /**
- * add console node path
- *
- * @param path node path
- * @param nodeType node type
- */
- public void addConsoleNodePath(String path, RegistryNodeType nodeType) {
- log.info("{} node added : {}", nodeType, path);
-
- if (StringUtils.isEmpty(path)) {
- log.error("server start error: empty path: {}, nodeType:{}", path, nodeType);
- return;
- }
-
- String serverHost = registryClient.getHostByEventDataPath(path);
- if (StringUtils.isEmpty(serverHost)) {
- log.error("server start error: unknown path: {}, nodeType:{}", path, nodeType);
- return;
- }
-
- try {
- if (!registryClient.exists(path)) {
- log.info("path: {} not exists", path);
- }
- distributedTaskService.addServer(serverHost);
- } catch (Exception e) {
- log.error("{} server failover failed, host:{}", nodeType, serverHost, e);
- }
- }
-
- /**
- * remove console node path
- *
- * @param path node path
- * @param nodeType node type
- * @param failover is failover
- */
- public void removeConsoleNodePath(String path, RegistryNodeType nodeType, boolean failover) {
- log.info("{} node deleted : {}", nodeType, path);
-
- if (StringUtils.isEmpty(path)) {
- log.error("server down error: empty path: {}, nodeType:{}", path, nodeType);
- return;
- }
-
- String serverHost = registryClient.getHostByEventDataPath(path);
- if (StringUtils.isEmpty(serverHost)) {
- log.error("server down error: unknown path: {}, nodeType:{}", path, nodeType);
- return;
- }
-
- try {
- if (!registryClient.exists(path)) {
- log.info("path: {} not exists", path);
- }
- // todo: add failover logic here
- } catch (Exception e) {
- log.error("{} server failover failed, host:{}", nodeType, serverHost, e);
- }
- }
-
- /**
- * Registry the current console server itself to registry.
- */
- void registry() {
- log.info("Console node : {} registering to registry center", consoleConfig.getConsoleAddress());
- String consoleRegistryPath = consoleConfig.getConsoleRegistryPath();
-
- ConsoleHeartBeat heartBeat = consoleHeartBeatTask.getHeartBeat();
- while (ServerStatusEnum.BUSY.equals(heartBeat.getServerStatusEnum())) {
- log.warn("Console node is BUSY: {}", heartBeat);
- ThreadUtils.sleep(SLEEP_TIME_MILLIS);
- heartBeat = consoleHeartBeatTask.getHeartBeat();
- }
-
- // remove before persist
- registryClient.remove(consoleRegistryPath);
- registryClient.persistEphemeral(consoleRegistryPath,
- JSONUtils.toJsonString(consoleHeartBeatTask.getHeartBeat()));
-
- while (!registryClient.checkNodeExists(NetworkUtils.getHost(), RegistryNodeType.CONSOLE_SERVER)) {
- log.warn("The current console server node:{} cannot find in registry", NetworkUtils.getHost());
- ThreadUtils.sleep(SLEEP_TIME_MILLIS);
- }
-
- // sleep 1s, waiting console failover remove
- ThreadUtils.sleep(SLEEP_TIME_MILLIS);
-
- consoleHeartBeatTask.start();
- Set<String> allServers = getServerNodes(RegistryNodeType.CONSOLE_SERVER);
- distributedTaskService.init(allServers, consoleConfig.getConsoleAddress());
- log.info("Console node : {} registered to registry center successfully", consoleConfig.getConsoleAddress());
-
- }
-
- public void deregister() {
- try {
- registryClient.remove(consoleConfig.getConsoleRegistryPath());
- log.info("Console node : {} unRegistry to register center.", consoleConfig.getConsoleAddress());
- if (consoleHeartBeatTask != null) {
- consoleHeartBeatTask.shutdown();
- }
- registryClient.close();
- } catch (Exception e) {
- log.error("ConsoleServer remove registry path exception ", e);
- }
- }
-
- public boolean isAvailable() {
- return registryClient.isConnected();
- }
-
- public Set<String> getServerNodes(RegistryNodeType nodeType) {
- return registryClient.getServerNodeSet(nodeType);
- }
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryDataListener.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryDataListener.java
deleted file mode 100644
index cb40aa072..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryDataListener.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * 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.registry;
-
-import org.apache.streampark.common.constants.Constants;
-import org.apache.streampark.console.base.util.SpringContextUtils;
-import org.apache.streampark.registry.api.Event;
-import org.apache.streampark.registry.api.SubscribeListener;
-import org.apache.streampark.registry.api.enums.RegistryNodeType;
-
-import com.google.common.base.Strings;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class ConsoleRegistryDataListener implements SubscribeListener {
-
- private final ConsoleRegistryClient consoleRegistryClient;
-
- public ConsoleRegistryDataListener() {
- consoleRegistryClient = SpringContextUtils.getBean(ConsoleRegistryClient.class);
- }
-
- @Override
- public void notify(Event event) {
- final String path = event.path();
- if (Strings.isNullOrEmpty(path)) {
- return;
- }
- // monitor console
- if (path.startsWith(RegistryNodeType.CONSOLE_SERVER.getRegistryPath() + Constants.SINGLE_SLASH)) {
- handleConsoleEvent(event);
- }
- }
-
- private void handleConsoleEvent(Event event) {
- final String path = event.path();
- switch (event.type()) {
- case ADD:
- log.info("console node added : {}", path);
- consoleRegistryClient.addConsoleNodePath(path, RegistryNodeType.CONSOLE_SERVER);
- break;
- case REMOVE:
- log.info("console node deleted : {}", path);
- consoleRegistryClient.removeConsoleNodePath(path, RegistryNodeType.CONSOLE_SERVER, true);
- break;
- default:
- break;
- }
- }
-
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleStopStrategy.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleStopStrategy.java
deleted file mode 100644
index 60034dd43..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleStopStrategy.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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.registry;
-
-import org.apache.streampark.registry.api.RegistryClient;
-import org.apache.streampark.registry.api.StrategyType;
-
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-/**
- * This strategy will stop the console server, when disconnected from {@link org.apache.streampark.registry.api.Registry}.
- */
-@Service
-@Slf4j
-public class ConsoleStopStrategy implements ConsoleConnectStrategy {
-
- @Autowired
- private RegistryClient registryClient;
-
- @Override
- public void disconnect() {
- registryClient.getStoppable()
- .stop("Console disconnected from registry, will stop myself due to the stop strategy");
- }
-
- @Override
- public void reconnect() {
- log.warn("The current connect strategy is stop, so the console will not reconnect to registry");
- }
-
- @Override
- public StrategyType getStrategyType() {
- return StrategyType.STOP;
- }
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/ConsoleHeartBeatTask.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/ConsoleHeartBeatTask.java
deleted file mode 100644
index 1fc25a18a..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/ConsoleHeartBeatTask.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.task;
-
-import org.apache.streampark.common.utils.JSONUtils;
-import org.apache.streampark.common.utils.NetworkUtils;
-import org.apache.streampark.common.utils.OSUtils;
-import org.apache.streampark.console.core.config.ConsoleConfig;
-import org.apache.streampark.registry.api.RegistryClient;
-import org.apache.streampark.registry.api.enums.ServerStatusEnum;
-import org.apache.streampark.registry.api.lifecycle.ServerLifeCycleManager;
-import org.apache.streampark.registry.api.model.BaseHeartBeatTask;
-import org.apache.streampark.registry.api.model.ConsoleHeartBeat;
-
-import lombok.NonNull;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class ConsoleHeartBeatTask extends BaseHeartBeatTask<ConsoleHeartBeat> {
-
- private final ConsoleConfig consoleConfig;
-
- private final RegistryClient registryClient;
-
- private final String heartBeatPath;
-
- private final int processId;
-
- public ConsoleHeartBeatTask(@NonNull ConsoleConfig consoleConfig,
- @NonNull RegistryClient registryClient) {
- super("ConsoleHeartBeatTask", consoleConfig.getMaxHeartbeatInterval().toMillis());
- this.consoleConfig = consoleConfig;
- this.registryClient = registryClient;
- this.heartBeatPath = consoleConfig.getConsoleRegistryPath();
- this.processId = OSUtils.getProcessID();
- }
-
- @Override
- public ConsoleHeartBeat getHeartBeat() {
- return ConsoleHeartBeat.builder()
- .startupTime(ServerLifeCycleManager.getServerStartupTime())
- .reportTime(System.currentTimeMillis())
- .processId(processId)
- .serverStatusEnum(ServerStatusEnum.NORMAL)
- .host(NetworkUtils.getHost())
- .port(consoleConfig.getListenPort())
- .build();
- }
-
- @Override
- public void writeHeartBeat(ConsoleHeartBeat consoleHeartBeat) {
- String masterHeartBeatJson = JSONUtils.toJsonString(consoleHeartBeat);
- registryClient.persistEphemeral(heartBeatPath, masterHeartBeatJson);
- log.debug("Success write master heartBeatInfo into registry, masterRegistryPath: {}, heartBeatInfo: {}",
- heartBeatPath, masterHeartBeatJson);
- }
-}