[Chore] Merge the DataSourcePluginManager and DataSourceProcessorManager (#17975)

Author: ruanwenjun

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-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);
+        });
     }
+
 }

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

@@ -20,12 +20,11 @@
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.BaseDataSourceParamDTO;
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.DataSourceProcessor;
-import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider;
+import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourcePluginManager;
 import org.apache.dolphinscheduler.spi.datasource.ConnectionParam;
 import org.apache.dolphinscheduler.spi.enums.DbType;
 
 import java.sql.Connection;
-import java.util.Map;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -75,12 +74,7 @@ public static BaseDataSourceParamDTO buildDatasourceParamDTO(DbType dbType, Stri
     }
 
     public static DataSourceProcessor getDatasourceProcessor(DbType dbType) {
-        Map<String, DataSourceProcessor> dataSourceProcessorMap =
-                DataSourceProcessorProvider.getDataSourceProcessorMap();
-        if (!dataSourceProcessorMap.containsKey(dbType.name())) {
-            throw new IllegalArgumentException("illegal datasource type");
-        }
-        return dataSourceProcessorMap.get(dbType.name());
+        return DataSourcePluginManager.getDataSourceProcessor(dbType);
     }
 
     /**

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

@@ -30,15 +30,12 @@
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
-import org.mockito.junit.jupiter.MockitoExtension;
 
-@ExtendWith(MockitoExtension.class)
 public class DatabendDataSourceProcessorTest {
 
-    private DatabendDataSourceProcessor databendDataSourceProcessor = new DatabendDataSourceProcessor();
+    private final DatabendDataSourceProcessor databendDataSourceProcessor = new DatabendDataSourceProcessor();
 
     @Test
     public void testCheckDatasourceParam() {

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

@@ -26,7 +26,7 @@
 import org.apache.dolphinscheduler.dao.DaoConfiguration;
 import org.apache.dolphinscheduler.meter.metrics.MetricsProvider;
 import org.apache.dolphinscheduler.meter.metrics.SystemMetrics;
-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;
@@ -123,7 +123,7 @@ public void initialized() {
 
         // install task plugin
         TaskPluginManager.loadTaskPlugin();
-        DataSourceProcessorProvider.initialize();
+        DataSourcePluginManager.loadDataSourcePlugin();
 
         // self tolerant
         this.masterRegistryClient.start();

dolphinscheduler-datasource-plugin/dolphinscheduler-datasource-databend/src/test/java/org/apache/dolphinscheduler/plugin/datasource/databend/param/DatabendDataSourceProcessorTest.java

@@ -23,7 +23,7 @@
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.plugin.datasource.api.datasource.DataSourceProcessor;
 import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClientProvider;
-import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider;
+import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourcePluginManager;
 import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
 import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack;
 import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
@@ -95,7 +95,7 @@ public void handle(TaskCallBack taskCallBack) throws TaskException {
                 procedureParameters.getLocalParams());
 
         DbType dbType = DbType.valueOf(procedureParameters.getType());
-        DataSourceProcessor dataSourceProcessor = DataSourceProcessorProvider.getDataSourceProcessor(dbType);
+        DataSourceProcessor dataSourceProcessor = DataSourcePluginManager.getDataSourceProcessor(dbType);
         ConnectionParam connectionParams =
                 dataSourceProcessor.createConnectionParams(procedureTaskExecutionContext.getConnectionParams());
         try (Connection connection = DataSourceClientProvider.getAdHocConnection(dbType, connectionParams)) {