You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by GitBox <gi...@apache.org> on 2022/06/11 02:55:40 UTC
[GitHub] [dolphinscheduler] kezhenxu94 commented on a diff in pull request #10406: Add mysql registry plugin
kezhenxu94 commented on code in PR #10406:
URL: https://github.com/apache/dolphinscheduler/pull/10406#discussion_r894669035
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-mysql/README.md:
##########
@@ -0,0 +1,23 @@
+# Introduction
+
+This module is the mysql registry plugin module, this plugin will use mysql as the registry center.
+
+# How to use
+
+If you want to set the registry center as mysql, you need to do the below two steps:
+
+1. Initialize the mysql table
+
+You can directly execute the sql script `src/main/resources/mysql_registry_init.sql`.
+
+2. Open the config
+
+You need to set the registry type to mysql in master/worker/api's appplication.yml
+
+```yaml
+registry:
+ type: zookeeper
Review Comment:
```suggestion
type: mysql
```
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-mysql/pom.xml:
##########
@@ -0,0 +1,56 @@
+<?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">
+ <parent>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-registry-plugins</artifactId>
+ <version>dev-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>dolphinscheduler-registry-mysql</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-registry-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-dao</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.projectlombok</groupId>
Review Comment:
If you want to use this. Add this to the root module (not in `<dependencyManagement>`, but in `<dependencies>`). We don't want to add it again and again in every module
##########
pom.xml:
##########
@@ -850,6 +856,13 @@
<version>${aws.sdk.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.projectlombok</groupId>
+ <artifactId>lombok</artifactId>
+ <version>${lombok.version}</version>
+ <scope>provided</scope>
+ </dependency>
Review Comment:
Move this out from `<dependencyManagement` and move it into `<dependencies>`
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-mysql/src/main/java/org/apache/dolphinscheduler/plugin/registry/mysql/MysqlRegistryConstant.java:
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.dolphinscheduler.plugin.registry.mysql;
+
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor
Review Comment:
I think you need `@UtilityClass`. `@NoArgsConstructor` means it will generate a constructor with no argument.
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-mysql/src/main/resources/mysql_registry_init.sql:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.
+*/
+
+SET FOREIGN_KEY_CHECKS = 0;
+
+DROP TABLE IF EXISTS `t_ds_mysql_registry_data`;
+CREATE TABLE `t_ds_mysql_registry_data`
+(
+ `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT 'primary key',
+ `key` varchar(200) NOT NULL COMMENT 'key, like zookeeper node path',
+ `data` varchar(200) NOT NULL COMMENT 'data, like zookeeper node value',
+ `type` tinyint(4) NOT NULL COMMENT '1: ephemeral node, 2: persistent node',
+ `lastUpdateTime` timestamp NULL COMMENT 'last update time',
+ `createTime` timestamp NULL COMMENT 'create time',
Review Comment:
column names are usually snake_case
##########
dolphinscheduler-api/pom.xml:
##########
@@ -64,6 +64,10 @@
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-registry-zookeeper</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.dolphinscheduler</groupId>
+ <artifactId>dolphinscheduler-registry-mysql</artifactId>
+ </dependency>
Review Comment:
Can you do in the same way as how we did in the task plugins? Create a new module `dolphinscheduler-registry-all` to contain all these registry plugin, and we can only add `dolphinscheduler-registry-all` to these starter module, in the future when we add new registry module we can only add them in `dolphinscheduler-registry-all` once
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-mysql/src/main/java/org/apache/dolphinscheduler/plugin/registry/mysql/task/RegistryLockManager.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.dolphinscheduler.plugin.registry.mysql.task;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.dolphinscheduler.common.thread.ThreadUtils;
+import org.apache.dolphinscheduler.plugin.registry.mysql.MysqlOperator;
+import org.apache.dolphinscheduler.plugin.registry.mysql.MysqlRegistryConstant;
+import org.apache.dolphinscheduler.plugin.registry.mysql.model.MysqlRegistryLock;
+import org.apache.dolphinscheduler.registry.api.RegistryException;
+
+import java.sql.SQLException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+public class RegistryLockManager implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(RegistryLockManager.class);
+
+ private final MysqlOperator mysqlOperator;
+ private final Map<String, MysqlRegistryLock> lockHoldMap;
+ private final ScheduledExecutorService lockTermUpdateThreadPool;
+
+ public RegistryLockManager(MysqlOperator mysqlOperator) {
+ this.mysqlOperator = mysqlOperator;
+ mysqlOperator.clearExpireLock();
+ this.lockHoldMap = new ConcurrentHashMap<>();
+ this.lockTermUpdateThreadPool = Executors.newScheduledThreadPool(
+ 1,
+ new ThreadFactoryBuilder().setNameFormat("MysqlRegistryLockTermRefreshThread").setDaemon(true).build());
+ }
+
+ public void start() {
+ lockTermUpdateThreadPool.scheduleWithFixedDelay(
+ new LockTermRefreshTask(lockHoldMap, mysqlOperator),
+ MysqlRegistryConstant.TERM_REFRESH_INTERVAL,
+ MysqlRegistryConstant.TERM_REFRESH_INTERVAL,
+ TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Acquire the lock, if cannot get the lock will await.
+ */
+ public void acquireLock(String lockKey) throws RegistryException {
+ // maybe we can use the computeIf absent
+ lockHoldMap.computeIfAbsent(lockKey, key -> {
+ MysqlRegistryLock mysqlRegistryLock;
+ try {
+ while ((mysqlRegistryLock = mysqlOperator.tryToAcquireLock(lockKey)) == null) {
+ logger.debug("Acquire the lock {} failed try again", key);
+ // acquire failed, wait and try again
+ ThreadUtils.sleep(1_000L);
+ }
+ } catch (SQLException e) {
+ throw new RegistryException("Acquire the lock error", e);
+ }
+ return mysqlRegistryLock;
+ });
+ }
+
+ public void releaseLock(String lockKey) {
+ MysqlRegistryLock mysqlRegistryLock = lockHoldMap.get(lockKey);
+ if (mysqlRegistryLock != null) {
+ try {
+ // the lock is unExit
+ mysqlOperator.releaseLock(mysqlRegistryLock.getId());
+ lockHoldMap.remove(lockKey);
+ } catch (SQLException e) {
+ throw new RegistryException(String.format("Release lock: %s error", lockKey), e);
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ lockTermUpdateThreadPool.shutdownNow();
+ for (Map.Entry<String, MysqlRegistryLock> lockEntry : lockHoldMap.entrySet()) {
+ releaseLock(lockEntry.getKey());
+ }
+ }
+
+ /**
+ * This task is used to refresh the lock held by the current server.
+ */
+ static class LockTermRefreshTask implements Runnable {
+ private final Map<String, MysqlRegistryLock> lockHoldMap;
+ private final MysqlOperator mysqlOperator;
+
+ private LockTermRefreshTask(Map<String, MysqlRegistryLock> lockHoldMap, MysqlOperator mysqlOperator) {
+ this.lockHoldMap = checkNotNull(lockHoldMap);
+ this.mysqlOperator = checkNotNull(mysqlOperator);
+ }
+
Review Comment:
Replace these with `@RequiredArgsConstructor(access = AccessLevel.PRIVATE)` in the class
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-mysql/src/main/java/org/apache/dolphinscheduler/plugin/registry/mysql/task/SubscribeDataManager.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.dolphinscheduler.plugin.registry.mysql.task;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.dolphinscheduler.plugin.registry.mysql.MysqlOperator;
+import org.apache.dolphinscheduler.plugin.registry.mysql.MysqlRegistryConstant;
+import org.apache.dolphinscheduler.plugin.registry.mysql.model.MysqlRegistryData;
+import org.apache.dolphinscheduler.registry.api.Event;
+import org.apache.dolphinscheduler.registry.api.SubscribeListener;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+/**
+ * Used to refresh if the subscribe path has been changed.
+ */
+public class SubscribeDataManager implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SubscribeDataManager.class);
+
+ private final MysqlOperator mysqlOperator;
+ private final Map<String, List<SubscribeListener>> dataSubScribeMap = new ConcurrentHashMap<>();
+ private final ScheduledExecutorService dataSubscribeCheckThreadPool;
+ private final Map<String, MysqlRegistryData> mysqlRegistryDataMap = new ConcurrentHashMap<>();
+
+ public SubscribeDataManager(MysqlOperator mysqlOperator) {
+ this.mysqlOperator = mysqlOperator;
+ this.dataSubscribeCheckThreadPool = Executors.newScheduledThreadPool(
+ 1,
+ new ThreadFactoryBuilder().setNameFormat("MysqlRegistrySubscribeDataCheckThread").setDaemon(true).build());
+ }
+
+ public void start() {
+ dataSubscribeCheckThreadPool.scheduleWithFixedDelay(
+ new RegistrySubscribeDataCheckTask(mysqlRegistryDataMap, dataSubScribeMap, mysqlOperator),
+ MysqlRegistryConstant.TERM_REFRESH_INTERVAL,
+ MysqlRegistryConstant.TERM_REFRESH_INTERVAL,
+ TimeUnit.MILLISECONDS);
+ }
+
+ public void addListener(String path, SubscribeListener subscribeListener) {
+ dataSubScribeMap.computeIfAbsent(path, k -> new ArrayList<>()).add(subscribeListener);
+ }
+
+ public void removeListener(String path) {
+ dataSubScribeMap.remove(path);
+ }
+
+ public String getData(String path) throws SQLException {
+ MysqlRegistryData mysqlRegistryData = mysqlRegistryDataMap.get(path);
+ if (mysqlRegistryData == null) {
+ return null;
+ }
+ return mysqlRegistryData.getData();
+ }
+
+ @Override
+ public void close() {
+ dataSubscribeCheckThreadPool.shutdownNow();
+ dataSubScribeMap.clear();
+ }
+
+ static class RegistrySubscribeDataCheckTask implements Runnable {
+
+ private final Map<String, List<SubscribeListener>> dataSubScribeMap;
+ private final MysqlOperator mysqlOperator;
+ private final Map<String, MysqlRegistryData> mysqlRegistryDataMap;
+
+ public RegistrySubscribeDataCheckTask(
+ Map<String, MysqlRegistryData> mysqlRegistryDataMap,
+ Map<String, List<SubscribeListener>> dataSubScribeMap,
+ MysqlOperator mysqlOperator) {
+ this.mysqlRegistryDataMap = checkNotNull(mysqlRegistryDataMap);
+ this.dataSubScribeMap = checkNotNull(dataSubScribeMap);
+ this.mysqlOperator = checkNotNull(mysqlOperator);
+ }
Review Comment:
Replace this with `@RequiredArgsConstructor`
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-mysql/src/main/java/org/apache/dolphinscheduler/plugin/registry/mysql/MysqlOperator.java:
##########
@@ -0,0 +1,294 @@
+/*
+ * 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.dolphinscheduler.plugin.registry.mysql;
+
+import org.apache.dolphinscheduler.common.utils.NetUtils;
+import org.apache.dolphinscheduler.plugin.registry.mysql.model.DataType;
+import org.apache.dolphinscheduler.plugin.registry.mysql.model.MysqlRegistryData;
+import org.apache.dolphinscheduler.plugin.registry.mysql.model.MysqlRegistryLock;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLIntegrityConstraintViolationException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.sql.DataSource;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Used to CRUD from mysql
+ */
+public class MysqlOperator {
+
+ private static final Logger logger = LoggerFactory.getLogger(MysqlOperator.class);
+
+ private final DataSource dataSource;
+
+ public MysqlOperator(DataSource dataSource) {
+ this.dataSource = dataSource;
+ }
+
+ public void healthCheck() throws SQLException {
+ String sql = "select 1 from t_ds_mysql_registry_data";
+ try (Connection connection = dataSource.getConnection();
+ PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+ // if no exception, the healthCheck success
+ preparedStatement.executeQuery();
+ }
+ }
+
+ public List<MysqlRegistryData> queryAllMysqlRegistryData() throws SQLException {
+ String sql = "select id, `key`, data, type, createTime, lastUpdateTime from t_ds_mysql_registry_data";
+ try (Connection connection = dataSource.getConnection();
+ PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+ ResultSet resultSet = preparedStatement.executeQuery();
+ List<MysqlRegistryData> result = new ArrayList<>(resultSet.getFetchSize());
+ while (resultSet.next()) {
+ MysqlRegistryData mysqlRegistryData = MysqlRegistryData.builder()
+ .id(resultSet.getLong(1))
+ .key(resultSet.getString(2))
+ .data(resultSet.getString(3))
+ .type(resultSet.getInt(4))
+ .createTime(resultSet.getTimestamp(5))
+ .lastUpdateTime(resultSet.getTimestamp(6))
Review Comment:
What about getting by column label instead of the column index?
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-zookeeper/src/main/java/org/apache/dolphinscheduler/plugin/registry/zookeeper/ZookeeperRegistryProperties.java:
##########
@@ -1,33 +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
+ * 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.
+ * 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.dolphinscheduler.registry.api;
+package org.apache.dolphinscheduler.plugin.registry.zookeeper;
import java.time.Duration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
@Configuration
+@ConditionalOnProperty(prefix = "registry", name = "type", havingValue = "zookeeper")
Review Comment:
I think you can remove the `type` field in this class then
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-mysql/README.md:
##########
@@ -0,0 +1,23 @@
+# Introduction
+
+This module is the mysql registry plugin module, this plugin will use mysql as the registry center.
+
+# How to use
+
+If you want to set the registry center as mysql, you need to do the below two steps:
+
+1. Initialize the mysql table
Review Comment:
I think users need to add MySQL driver manually, configure the connection string, etc., if they are not already using MySQL as database.
--
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: commits-unsubscribe@dolphinscheduler.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org