Merge branch 'apache:dev' into dev

Author: det101

.asf.yaml

@@ -52,7 +52,7 @@ github:
           - "Title Validator"
       required_pull_request_reviews:
         dismiss_stale_reviews: true
-        required_approving_review_count: 2
+        required_approving_review_count: 1
   features:
     # Enable issues management
     issues: true

dolphinscheduler-alert/dolphinscheduler-alert-plugins/dolphinscheduler-alert-webexteams/pom.xml

@@ -37,12 +37,6 @@
             <scope>provided</scope>
         </dependency>
 
-        <dependency>
-            <groupId>org.apache.dolphinscheduler</groupId>
-            <artifactId>dolphinscheduler-spi</artifactId>
-            <scope>provided</scope>
-        </dependency>
-
         <dependency>
             <groupId>org.apache.dolphinscheduler</groupId>
             <artifactId>dolphinscheduler-alert-api</artifactId>

dolphinscheduler-alert/dolphinscheduler-alert-plugins/dolphinscheduler-alert-wechat/pom.xml

@@ -37,12 +37,6 @@
             <scope>provided</scope>
         </dependency>
 
-        <dependency>
-            <groupId>org.apache.dolphinscheduler</groupId>
-            <artifactId>dolphinscheduler-spi</artifactId>
-            <scope>provided</scope>
-        </dependency>
-
         <dependency>
             <groupId>org.apache.dolphinscheduler</groupId>
             <artifactId>dolphinscheduler-alert-api</artifactId>

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/ApiApplicationServer.java

@@ -22,7 +22,7 @@
 import org.apache.dolphinscheduler.common.lifecycle.ServerLifeCycleManager;
 import org.apache.dolphinscheduler.common.thread.DefaultUncaughtExceptionHandler;
 import org.apache.dolphinscheduler.dao.DaoConfiguration;
-import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider;
+import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourcePluginManager;
 import org.apache.dolphinscheduler.plugin.storage.api.StorageConfiguration;
 import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
 import org.apache.dolphinscheduler.registry.api.RegistryConfiguration;
@@ -57,7 +57,7 @@ public static void main(String[] args) {
     public void run(ApplicationReadyEvent readyEvent) {
         ServerLifeCycleManager.toRunning();
         log.info("Received spring application context ready event will load taskPlugin and write to DB");
-        DataSourceProcessorProvider.initialize();
+        DataSourcePluginManager.loadDataSourcePlugin();
         TaskPluginManager.loadTaskPlugin();
     }
 }

dolphinscheduler-bom/pom.xml

@@ -86,7 +86,6 @@
         <jsr305.version>3.0.0</jsr305.version>
         <commons-compress.version>1.27.1</commons-compress.version>
         <commons-math3.version>3.6.1</commons-math3.version>
-        <error_prone_annotations.version>2.5.1</error_prone_annotations.version>
         <hibernate-validator.version>6.2.2.Final</hibernate-validator.version>
         <aws-sdk.version>1.12.300</aws-sdk.version>
         <aliyun-sdk-oss.version>3.15.1</aliyun-sdk-oss.version>
@@ -152,16 +151,6 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
-            <dependency>
-                <groupId>io.grpc</groupId>
-                <artifactId>grpc-api</artifactId>
-                <version>${io.grpc.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>io.grpc</groupId>
-                <artifactId>grpc-core</artifactId>
-                <version>${io.grpc.version}</version>
-            </dependency>
             <dependency>
                 <groupId>io.grpc</groupId>
                 <artifactId>grpc-services</artifactId>
@@ -740,11 +729,6 @@
                 <version>${py4j.version}</version>
             </dependency>
 
-            <dependency>
-                <groupId>com.google.code.findbugs</groupId>
-                <artifactId>jsr305</artifactId>
-                <version>${jsr305.version}</version>
-            </dependency>
             <dependency>
                 <groupId>org.apache.commons</groupId>
                 <artifactId>commons-compress</artifactId>
@@ -755,11 +739,6 @@
                 <artifactId>commons-math3</artifactId>
                 <version>${commons-math3.version}</version>
             </dependency>
-            <dependency>
-                <groupId>com.google.errorprone</groupId>
-                <artifactId>error_prone_annotations</artifactId>
-                <version>${error_prone_annotations.version}</version>
-            </dependency>
 
             <!-- https://mvnrepository.com/artifact/org.hibernate.validator/hibernate-validator -->
             <dependency>
@@ -936,12 +915,6 @@
                 <version>${zeppelin-client.version}</version>
             </dependency>
 
-            <dependency>
-                <groupId>com.google.protobuf</groupId>
-                <artifactId>protobuf-java</artifactId>
-                <version>${protobuf.version}</version>
-            </dependency>
-
             <dependency>
                 <groupId>com.huaweicloud</groupId>
                 <artifactId>esdk-obs-java-bundle</artifactId>
@@ -1048,7 +1021,7 @@
                 <activeByDefault>true</activeByDefault>
             </activation>
             <properties>
-                <zookeeper.version>3.8.0</zookeeper.version>
+                <zookeeper.version>3.8.3</zookeeper.version>
                 <curator.version>5.5.0</curator.version>
             </properties>
         </profile>

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/main/java/org/apache/dolphinscheduler/plugin/datasource/api/client/BaseAdHocDataSourceClient.java

@@ -17,7 +17,7 @@
 
 package org.apache.dolphinscheduler.plugin.datasource.api.client;
 
-import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider;
+import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourcePluginManager;
 import org.apache.dolphinscheduler.spi.datasource.AdHocDataSourceClient;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.enums.DbType;
@@ -38,7 +38,7 @@ protected BaseAdHocDataSourceClient(BaseConnectionParam baseConnectionParam, DbT
     @Override
     public Connection getConnection() throws SQLException {
         try {
-            return DataSourceProcessorProvider.getDataSourceProcessor(dbType).getConnection(baseConnectionParam);
+            return DataSourcePluginManager.getDataSourceProcessor(dbType).getConnection(baseConnectionParam);
         } catch (Exception e) {
             throw new SQLException("Create adhoc connection error", e);
         }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/main/java/org/apache/dolphinscheduler/plugin/datasource/api/plugin/DataSourceClientProvider.java

@@ -57,18 +57,13 @@ public class DataSourceClientProvider {
                     })
                     .maximumSize(100)
                     .build();
-    private static final DataSourcePluginManager dataSourcePluginManager = new DataSourcePluginManager();
-
-    static {
-        dataSourcePluginManager.installPlugin();
-    }
 
     public static DataSourceClient getPooledDataSourceClient(DbType dbType,
                                                              ConnectionParam connectionParam) throws ExecutionException {
         BaseConnectionParam baseConnectionParam = (BaseConnectionParam) connectionParam;
         String datasourceUniqueId = DataSourceUtils.getDatasourceUniqueId(baseConnectionParam, dbType);
         return POOLED_DATASOURCE_CLIENT_CACHE.get(datasourceUniqueId, () -> {
-            DataSourceChannel dataSourceChannel = dataSourcePluginManager.getDataSourceChannel(dbType);
+            DataSourceChannel dataSourceChannel = DataSourcePluginManager.getDataSourceChannel(dbType);
             if (null == dataSourceChannel) {
                 throw new RuntimeException(String.format("datasource plugin '%s' is not found", dbType.getName()));
             }
@@ -83,7 +78,7 @@ public static Connection getPooledConnection(DbType dbType,
 
     public static AdHocDataSourceClient getAdHocDataSourceClient(DbType dbType, ConnectionParam connectionParam) {
         BaseConnectionParam baseConnectionParam = (BaseConnectionParam) connectionParam;
-        DataSourceChannel dataSourceChannel = dataSourcePluginManager.getDataSourceChannel(dbType);
+        DataSourceChannel dataSourceChannel = DataSourcePluginManager.getDataSourceChannel(dbType);
         if (null == dataSourceChannel) {
             throw new RuntimeException(String.format("datasource plugin '%s' is not found", dbType.getName()));
         }

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-api/src/main/java/org/apache/dolphinscheduler/plugin/datasource/api/plugin/DataSourcePluginManager.java

@@ -19,47 +19,74 @@
 
 import static java.lang.String.format;
 
+import org.apache.dolphinscheduler.plugin.datasource.api.datasource.DataSourceProcessor;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceChannel;
 import org.apache.dolphinscheduler.spi.datasource.DataSourceChannelFactory;
 import org.apache.dolphinscheduler.spi.enums.DbType;
 import org.apache.dolphinscheduler.spi.plugin.PrioritySPIFactory;
 
+import org.apache.commons.collections4.MapUtils;
+
 import java.util.Map;
+import java.util.ServiceLoader;
 import java.util.concurrent.ConcurrentHashMap;
 
+import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 public class DataSourcePluginManager {
 
-    private final Map<String, DataSourceChannel> datasourceChannelMap = new ConcurrentHashMap<>();
-
-    public DataSourceChannel getDataSourceChannel(final DbType dbType) {
-        return datasourceChannelMap.get(dbType.getName());
-    }
+    private static final Map<String, DataSourceChannel> datasourceChannelMap = new ConcurrentHashMap<>();
 
-    public void installPlugin() {
+    private static final Map<String, DataSourceProcessor> dataSourceProcessorMap = new ConcurrentHashMap<>();
 
-        PrioritySPIFactory<DataSourceChannelFactory> prioritySPIFactory =
-                new PrioritySPIFactory<>(DataSourceChannelFactory.class);
-        for (Map.Entry<String, DataSourceChannelFactory> entry : prioritySPIFactory.getSPIMap().entrySet()) {
-            final DataSourceChannelFactory factory = entry.getValue();
-            final String name = entry.getKey();
+    static {
+        loadDataSourcePlugin();
+    }
 
-            log.info("Registering datasource plugin: {}", name);
+    public static DataSourceChannel getDataSourceChannel(@NonNull DbType dbType) {
+        return datasourceChannelMap.get(dbType.getName());
+    }
 
-            if (datasourceChannelMap.containsKey(name)) {
-                throw new IllegalStateException(format("Duplicate datasource plugins named '%s'", name));
-            }
+    public static DataSourceProcessor getDataSourceProcessor(@NonNull DbType dbType) {
+        return dataSourceProcessorMap.get(dbType.getName());
+    }
 
-            loadDatasourceClient(factory);
+    public static void loadDataSourcePlugin() {
+        initializeDataSourceChannel();
+        initializeDataSourceProcessor();
+    }
 
-            log.info("Registered datasource plugin: {}", name);
+    private static synchronized void initializeDataSourceChannel() {
+        if (MapUtils.isNotEmpty(datasourceChannelMap)) {
+            return;
         }
+        new PrioritySPIFactory<>(DataSourceChannelFactory.class).getSPIMap().forEach(
+                (dataSourceChannelName, dataSourceChannelFactory) -> {
+                    if (datasourceChannelMap.containsKey(dataSourceChannelName)) {
+                        throw new IllegalStateException(
+                                format("Duplicate datasource channel named '%s'", dataSourceChannelName));
+                    }
+                    datasourceChannelMap.put(dataSourceChannelName, dataSourceChannelFactory.create());
+                    log.info("Registered datasource channel: {}", dataSourceChannelName);
+                });
     }
 
-    private void loadDatasourceClient(DataSourceChannelFactory datasourceChannelFactory) {
-        DataSourceChannel datasourceChannel = datasourceChannelFactory.create();
-        datasourceChannelMap.put(datasourceChannelFactory.getName(), datasourceChannel);
+    private static synchronized void initializeDataSourceProcessor() {
+        if (MapUtils.isNotEmpty(dataSourceProcessorMap)) {
+            return;
+        }
+
+        ServiceLoader.load(DataSourceProcessor.class).forEach(factory -> {
+            final String name = factory.getDbType().getName();
+            if (dataSourceProcessorMap.containsKey(name)) {
+                throw new IllegalStateException(format("Duplicate datasource processor named '%s'", name));
+            }
+            DataSourceProcessor dataSourceProcessor = factory.create();
+            dataSourceProcessorMap.put(name, dataSourceProcessor);
+            log.info("Success register datasource processor -> {}", name);
+        });
     }
+
 }