You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@streampark.apache.org by "SbloodyS (via GitHub)" <gi...@apache.org> on 2024/08/08 03:22:41 UTC
[PR] [Feature-3934] Add registry center [incubator-streampark]
SbloodyS opened a new pull request, #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948
<!--
Thank you for contributing to StreamPark! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
## Contribution Checklist
- If this is your first time, please read our contributor guidelines: [Submit Code](https://streampark.apache.org/community/submit_guide/submit_code).
- Make sure that the pull request corresponds to a [GITHUB issue](https://github.com/apache/incubator-streampark/issues).
- Name the pull request in the form "[Feature] Title of the pull request", where *Feature* can be replaced by `Hotfix`, `Bug`, etc.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- If the PR is unfinished, add `[WIP]` in your PR title, e.g., `[WIP][Feature] Title of the pull request`.
-->
## What changes were proposed in this pull request
Issue Number: close #3934
<!--(For example: This pull request proposed to add checkstyle plugin).-->
## Brief change log
<!--*(for example:)*
- *Add maven-checkstyle-plugin to root pom.xml*
-->
## Verifying this change
<!--*(Please pick either of the following options)*-->
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
<!--*(example:)*
- *Added integration tests for end-to-end.*
- *Added *Test to verify the change.*
- *Manually verified the change by testing locally.* -->
## Does this pull request potentially affect one of the following parts
- Dependencies (does it add or upgrade a dependency): (yes / no)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1714575428
##########
streampark-common/src/main/java/org/apache/streampark/common/sql/ClasspathSqlScriptParser.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.common.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 {
Review Comment:
Because the registry is a separate function, it should be separate from the core function to ensure that the core function is not affected. So it will use a separate database connection pool and a separate thread pool just like `org.apache.streampark.common.thread.BaseDaemonThread`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2275149273
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS closed pull request #3948: [Feature-3934] Add registy module for streampark
URL: https://github.com/apache/incubator-streampark/pull/3948
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1711169894
##########
streampark-common/src/main/java/org/apache/streampark/common/utils/UUIDUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.common.utils;
+
+import lombok.experimental.UtilityClass;
+
+import java.util.Arrays;
+import java.util.UUID;
+
+@UtilityClass
+public class UUIDUtils {
+
+ public static long generateUUID() {
+ String uuidString = UUID.randomUUID().toString().replaceAll("[^0-9]]", "");
Review Comment:
Good point. Will fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1711457292
##########
streampark-common/src/main/java/org/apache/streampark/common/utils/UUIDUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.common.utils;
+
+import lombok.experimental.UtilityClass;
+
+import java.util.Arrays;
+import java.util.UUID;
+
+@UtilityClass
+public class UUIDUtils {
+
+ public static long generateUUID() {
+ String uuidString = UUID.randomUUID().toString().replaceAll("[^0-9]]", "");
Review Comment:
Since the maximum length is too long, I reduced the length to 14 bits. I also added multithreaded unit tests to ensure there are no conflicts. PTAL @HxpSerein
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.enums.ServerStatus;
+import org.apache.streampark.common.model.ConsoleHeartBeat;
+import org.apache.streampark.common.thread.ThreadUtils;
+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.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.commons.lang3.StringUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+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;
+
+ 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();
+ }
+
+ /**
+ * 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 (ServerStatus.BUSY.equals(heartBeat.getServerStatus())) {
+ log.warn("Console node is BUSY: {}", heartBeat);
+ heartBeat = consoleHeartBeatTask.getHeartBeat();
+ ThreadUtils.sleep(SLEEP_TIME_MILLIS);
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registry center [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2274882089
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2290727248
As no consensus can be reached at the moment, temporarily closing this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "wolfboys (via GitHub)" <gi...@apache.org>.
wolfboys commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1714572812
##########
streampark-common/src/main/java/org/apache/streampark/common/sql/ClasspathSqlScriptParser.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.common.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 {
Review Comment:
I don't know why this class is added, Do we need to execute sql manually?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2274927508
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1711171064
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.enums.ServerStatus;
+import org.apache.streampark.common.model.ConsoleHeartBeat;
+import org.apache.streampark.common.thread.ThreadUtils;
+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.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.commons.lang3.StringUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+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;
+
+ 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();
+ }
+
+ /**
+ * 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 (ServerStatus.BUSY.equals(heartBeat.getServerStatus())) {
+ log.warn("Console node is BUSY: {}", heartBeat);
+ heartBeat = consoleHeartBeatTask.getHeartBeat();
+ ThreadUtils.sleep(SLEEP_TIME_MILLIS);
+ }
+
+ // remove before persist
+ registryClient.remove(consoleRegistryPath);
+ registryClient.persistEphemeral(consoleRegistryPath,
+ JSONUtils.toJsonString(consoleHeartBeatTask.getHeartBeat()));
Review Comment:
This is for the future if new node types are to be added. eg: `alertServerHeartBeat`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2274885280
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2275005323
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2275011538
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2275003165
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "HxpSerein (via GitHub)" <gi...@apache.org>.
HxpSerein commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1710878700
##########
streampark-common/src/main/java/org/apache/streampark/common/utils/UUIDUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.common.utils;
+
+import lombok.experimental.UtilityClass;
+
+import java.util.Arrays;
+import java.util.UUID;
+
+@UtilityClass
+public class UUIDUtils {
+
+ public static long generateUUID() {
+ String uuidString = UUID.randomUUID().toString().replaceAll("[^0-9]]", "");
Review Comment:
According to [stackoverflow](https://stackoverflow.com/questions/15184820/how-to-generate-unique-positive-long-using-uuid), I perfer `UUID.randomUUID().getMostSignificantBits() & Long.MAX_VALUE`.
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.enums.ServerStatus;
+import org.apache.streampark.common.model.ConsoleHeartBeat;
+import org.apache.streampark.common.thread.ThreadUtils;
+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.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.commons.lang3.StringUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+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;
+
+ 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();
+ }
+
+ /**
+ * 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 (ServerStatus.BUSY.equals(heartBeat.getServerStatus())) {
+ log.warn("Console node is BUSY: {}", heartBeat);
+ heartBeat = consoleHeartBeatTask.getHeartBeat();
+ ThreadUtils.sleep(SLEEP_TIME_MILLIS);
+ }
+
+ // remove before persist
+ registryClient.remove(consoleRegistryPath);
+ registryClient.persistEphemeral(consoleRegistryPath,
+ JSONUtils.toJsonString(consoleHeartBeatTask.getHeartBeat()));
Review Comment:
why not just use `heartBeat` instead of creating a new one?
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.enums.ServerStatus;
+import org.apache.streampark.common.model.ConsoleHeartBeat;
+import org.apache.streampark.common.thread.ThreadUtils;
+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.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.commons.lang3.StringUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+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;
+
+ 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();
+ }
+
+ /**
+ * 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 (ServerStatus.BUSY.equals(heartBeat.getServerStatus())) {
+ log.warn("Console node is BUSY: {}", heartBeat);
+ heartBeat = consoleHeartBeatTask.getHeartBeat();
+ ThreadUtils.sleep(SLEEP_TIME_MILLIS);
Review Comment:
move sleep before `getHeartBeat()`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2277862602
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2274923398
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1714572802
##########
pom.xml:
##########
@@ -146,6 +147,14 @@
<mockito.version>3.4.6</mockito.version>
<mockito-scala.version>1.17.14</mockito-scala.version>
<assertj.version>3.23.1</assertj.version>
+ <testcontainer.version>1.19.3</testcontainer.version>
+ <zookeeper.version>3.8.0</zookeeper.version>
Review Comment:
These two dependency only using in test scope. So it doesn't package into dist just like mysql.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "wolfboys (via GitHub)" <gi...@apache.org>.
wolfboys commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1714564753
##########
pom.xml:
##########
@@ -146,6 +147,14 @@
<mockito.version>3.4.6</mockito.version>
<mockito-scala.version>1.17.14</mockito-scala.version>
<assertj.version>3.23.1</assertj.version>
+ <testcontainer.version>1.19.3</testcontainer.version>
+ <zookeeper.version>3.8.0</zookeeper.version>
Review Comment:
Please make sure to declare these dependencies in the [LICENSE](https://github.com/apache/incubator-streampark/blob/dev/dist-material/release-docs/LICENSE)
##########
pom.xml:
##########
@@ -146,6 +147,14 @@
<mockito.version>3.4.6</mockito.version>
<mockito-scala.version>1.17.14</mockito-scala.version>
<assertj.version>3.23.1</assertj.version>
+ <testcontainer.version>1.19.3</testcontainer.version>
+ <zookeeper.version>3.8.0</zookeeper.version>
Review Comment:
Import ZooKeeper may result in additional dependencies, potentially causing dependency conflicts. Can you confirm this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1714577240
##########
streampark-registry/streampark-registry-plugins/streampark-registry-it/pom.xml:
##########
@@ -0,0 +1,61 @@
+<?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-registry-plugins</artifactId>
Review Comment:
Usually in java, provider stands for interface, which is provided to other implementations. But `streampark-registry-plugins` module is not an interface but an implementations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "wolfboys (via GitHub)" <gi...@apache.org>.
wolfboys commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2292544505
> > Thanks for your contribution. There are some minor issues:
> >
> > 1. streampark needs to support HA, which is a unique feature of the streampark-console module. As per our architectural design, this feature will depend on the registry module. The registry module is only used by the console. So, The registry module should not be placed under the root module; instead, it should be a submodule of the console.
> > 2. I don't think it's necessary to manually execute JDBC in the registry module because the console platform already has an ORM framework. Also, the registry module will be packaged into the console module, so there's no need to introduce a complex design to make the project look clearer.
> >
> > <img alt="image" width="371" src="https://private-user-images.githubusercontent.com/13284744/358124068-4084f6a3-2784-4eac-a7f4-877c286cad5e.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MjM3MDIyNTMsIm5iZiI6MTcyMzcwMTk1MywicGF0aCI6Ii8xMzI4NDc0NC8zNTgxMjQwNjgtNDA4NGY2YTMtMjc4NC00ZWFjLWE3ZjQtODc3YzI4NmNhZDVlLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNDA4MTUlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjQwODE1VDA2MDU1M1omWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPWU2OTFlMjBkMmYyY2M1NjQ5NGVkMjYwMWU2NjE3NjQyNTczYzgwNmZkY2I0NGFmYWRiZjhiM2E1MzRmYzYyNDcmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JmFjdG9yX2lkPTAma2V5X2lkPTAmcmVwb19pZD0wIn0.lCwJCrwDatTc6BrY0CPeg3VchX1gXsFj_vLMIooOx6w">
>
> I have a different opinion. We take this opportunity to reconstruct the console module. At present, the coupling of the console module is very exaggerated in my opinion, which is not conducive to the subsequent development of the project.
I have never been against splitting the console module or even refactoring the console module to make the project better. Still, these are all internal to the console module, and new modules should not be placed under the root module. You can look at the streampark-flink module, which has many submodules.
<img width="537" alt="image" src="https://github.com/user-attachments/assets/d3f23efa-ce1c-4932-9f34-88107e332211">
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2274931688
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registry center [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2274879552
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "HxpSerein (via GitHub)" <gi...@apache.org>.
HxpSerein commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1711545396
##########
streampark-common/src/main/java/org/apache/streampark/common/utils/UUIDUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.common.utils;
+
+import lombok.experimental.UtilityClass;
+
+import java.util.Arrays;
+import java.util.UUID;
+
+@UtilityClass
+public class UUIDUtils {
+
+ public static long generateUUID() {
+ String uuidString = UUID.randomUUID().toString().replaceAll("[^0-9]]", "");
Review Comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2277934922
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "HxpSerein (via GitHub)" <gi...@apache.org>.
HxpSerein commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1711536960
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.enums.ServerStatus;
+import org.apache.streampark.common.model.ConsoleHeartBeat;
+import org.apache.streampark.common.thread.ThreadUtils;
+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.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.commons.lang3.StringUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+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;
+
+ 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();
+ }
+
+ /**
+ * 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 (ServerStatus.BUSY.equals(heartBeat.getServerStatus())) {
+ log.warn("Console node is BUSY: {}", heartBeat);
+ heartBeat = consoleHeartBeatTask.getHeartBeat();
+ ThreadUtils.sleep(SLEEP_TIME_MILLIS);
+ }
+
+ // remove before persist
+ registryClient.remove(consoleRegistryPath);
+ registryClient.persistEphemeral(consoleRegistryPath,
+ JSONUtils.toJsonString(consoleHeartBeatTask.getHeartBeat()));
Review Comment:
ok
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.enums.ServerStatus;
+import org.apache.streampark.common.model.ConsoleHeartBeat;
+import org.apache.streampark.common.thread.ThreadUtils;
+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.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.commons.lang3.StringUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+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;
+
+ 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();
+ }
+
+ /**
+ * 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 (ServerStatus.BUSY.equals(heartBeat.getServerStatus())) {
+ log.warn("Console node is BUSY: {}", heartBeat);
+ heartBeat = consoleHeartBeatTask.getHeartBeat();
+ ThreadUtils.sleep(SLEEP_TIME_MILLIS);
Review Comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1714569530
##########
pom.xml:
##########
@@ -146,6 +147,14 @@
<mockito.version>3.4.6</mockito.version>
<mockito-scala.version>1.17.14</mockito-scala.version>
<assertj.version>3.23.1</assertj.version>
+ <testcontainer.version>1.19.3</testcontainer.version>
+ <zookeeper.version>3.8.0</zookeeper.version>
Review Comment:
These two dependency only using in test scope. So there will be no dependency conflicts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2278923956
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "caicancai (via GitHub)" <gi...@apache.org>.
caicancai commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2290734717
> Thanks for your contribution. There are some minor issues:
>
> 1. streampark needs to support HA, which is a unique feature of the streampark-console module. As per our architectural design, this feature will depend on the registry module. The registry module is only used by the console. So, The registry module should not be placed under the root module; instead, it should be a submodule of the console.
> 2. I don't think it's necessary to manually execute JDBC in the registry module because the console platform already has an ORM framework. Also, the registry module will be packaged into the console module, so there's no need to introduce a complex design to make the project look clearer.
>
> <img alt="image" width="371" src="https://private-user-images.githubusercontent.com/13284744/358124068-4084f6a3-2784-4eac-a7f4-877c286cad5e.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MjM3MDIyNTMsIm5iZiI6MTcyMzcwMTk1MywicGF0aCI6Ii8xMzI4NDc0NC8zNTgxMjQwNjgtNDA4NGY2YTMtMjc4NC00ZWFjLWE3ZjQtODc3YzI4NmNhZDVlLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNDA4MTUlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjQwODE1VDA2MDU1M1omWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPWU2OTFlMjBkMmYyY2M1NjQ5NGVkMjYwMWU2NjE3NjQyNTczYzgwNmZkY2I0NGFmYWRiZjhiM2E1MzRmYzYyNDcmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JmFjdG9yX2lkPTAma2V5X2lkPTAmcmVwb19pZD0wIn0.lCwJCrwDatTc6BrY0CPeg3VchX1gXsFj_vLMIooOx6w">
I have a different opinion. We take this opportunity to reconstruct the console module. At present, the coupling of the console module is very exaggerated in my opinion, which is not conducive to the subsequent development of the project.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "wolfboys (via GitHub)" <gi...@apache.org>.
wolfboys commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2290622684
1. streampark needs to support HA, which is a unique feature of the streampark-console module. As per our architectural design, this feature will depend on the registry module. The registry module is only used by the console. So, The registry module should not be placed under the root module; instead, it should be a submodule of the console.
2. I don't think it's necessary to manually execute JDBC in the registry module because the console platform already has an ORM framework. Also, the registry module will be packaged into the console module, so there's no need to introduce a complex design to make the project look clearer.
<img width="371" alt="image" src="https://github.com/user-attachments/assets/4084f6a3-2784-4eac-a7f4-877c286cad5e">
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "wolfboys (via GitHub)" <gi...@apache.org>.
wolfboys commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1717928774
##########
streampark-registry/streampark-registry-plugins/streampark-registry-it/pom.xml:
##########
@@ -0,0 +1,61 @@
+<?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-registry-plugins</artifactId>
Review Comment:
> Usually in java, provider stands for interface, which is provided to other implementations. But `streampark-registry-plugins` module is not an interface but an implementations.
Plugins may create ambiguity, and plugin modules will be introduced later. To avoid confusion, it is not recommended to use xx-plugin.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1711172960
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/registry/ConsoleRegistryClient.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.enums.ServerStatus;
+import org.apache.streampark.common.model.ConsoleHeartBeat;
+import org.apache.streampark.common.thread.ThreadUtils;
+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.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.commons.lang3.StringUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+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;
+
+ 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();
+ }
+
+ /**
+ * 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 (ServerStatus.BUSY.equals(heartBeat.getServerStatus())) {
+ log.warn("Console node is BUSY: {}", heartBeat);
+ heartBeat = consoleHeartBeatTask.getHeartBeat();
+ ThreadUtils.sleep(SLEEP_TIME_MILLIS);
Review Comment:
This code is not currently valid. But it is a good optimization point. Will fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "SbloodyS (via GitHub)" <gi...@apache.org>.
SbloodyS commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1711457292
##########
streampark-common/src/main/java/org/apache/streampark/common/utils/UUIDUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.common.utils;
+
+import lombok.experimental.UtilityClass;
+
+import java.util.Arrays;
+import java.util.UUID;
+
+@UtilityClass
+public class UUIDUtils {
+
+ public static long generateUUID() {
+ String uuidString = UUID.randomUUID().toString().replaceAll("[^0-9]]", "");
Review Comment:
Since the maximum length is too long for database column, I reduced the length to 14 bits. I also added multithreaded unit tests to ensure there are no conflicts. PTAL @HxpSerein
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#issuecomment-2277947782
## [](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) **Quality Gate failed**
Failed conditions
 [3.6% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_incubator-streampark&pullRequest=3948&metric=new_duplicated_lines_density&view=list) (required ≤ 3%)
 [E Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_incubator-streampark&pullRequest=3948)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Feature-3934] Add registy module for streampark [incubator-streampark]
Posted by "wolfboys (via GitHub)" <gi...@apache.org>.
wolfboys commented on code in PR #3948:
URL: https://github.com/apache/incubator-streampark/pull/3948#discussion_r1714575077
##########
streampark-registry/streampark-registry-plugins/streampark-registry-it/pom.xml:
##########
@@ -0,0 +1,61 @@
+<?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-registry-plugins</artifactId>
Review Comment:
How about streampark-registry-provider? I don't the name like xxx-plugins
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org