[Improve][e2e] Add driver-jar to lib (#2719)

* [Improve][e2e] add driver-jar to lib

* discovery third-party jars

* Remove excess code

* modify e2e test case

* modify e2e test case

* fix e2e error,if i run docker in another machine, the case will error

* fix some bug

* fix some bug

* fix some bug

* use root to start spark container.

* move `addURLToClassLoader` into  `registerPlugin`

* move `ADD_URL_TO_CLASSLOADER` into  `FlinkCommon`

Author: liugddx

seatunnel-common/src/main/java/org/apache/seatunnel/common/config/Common.java

@@ -145,7 +145,7 @@ public static Path pluginLibDir(String pluginName) {
     /**
      * return plugin's dependent jars, which located in 'plugins/${pluginName}/lib/*'.
      */
-    public static List<Path> getPluginsJarDependencies(){
+    public static List<Path> getPluginsJarDependencies() {
         Path pluginRootDir = Common.pluginRootDir();
         if (!Files.exists(pluginRootDir) || !Files.isDirectory(pluginRootDir)) {
             return Collections.emptyList();

seatunnel-connectors-v2/connector-jdbc/pom.xml

@@ -28,7 +28,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>connector-jdbc</artifactId>
-    
+
     <properties>
         <phoenix.version>5.2.5-HBase-2.x</phoenix.version>
         <mysql.version>8.0.16</mysql.version>
@@ -49,6 +49,7 @@
             <groupId>org.postgresql</groupId>
             <artifactId>postgresql</artifactId>
             <version>${postgresql.version}</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>com.dameng</groupId>

seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/config/FlinkCommon.java

@@ -0,0 +1,42 @@
+/*
+ * 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.seatunnel.core.starter.flink.config;
+
+import org.apache.seatunnel.common.utils.ReflectionUtils;
+
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.function.BiConsumer;
+
+public class FlinkCommon {
+
+    /**
+     * Add jar url to classloader. The different engine should have different logic to add url into
+     * their own classloader
+     */
+    public static BiConsumer<ClassLoader, URL> ADD_URL_TO_CLASSLOADER = (classLoader, url) -> {
+        if (classLoader.getClass().getName().endsWith("SafetyNetWrapperClassLoader")) {
+            URLClassLoader c = (URLClassLoader) ReflectionUtils.getField(classLoader, "inner").get();
+            ReflectionUtils.invoke(c, "addURL", url);
+        } else if (classLoader instanceof URLClassLoader) {
+            ReflectionUtils.invoke(classLoader, "addURL", url);
+        } else {
+            throw new RuntimeException("Unsupported classloader: " + classLoader.getClass().getName());
+        }
+    };
+}

seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/AbstractPluginExecuteProcessor.java

@@ -21,7 +21,7 @@
 import static org.apache.seatunnel.apis.base.plugin.Plugin.SOURCE_TABLE_NAME;
 
 import org.apache.seatunnel.api.common.JobContext;
-import org.apache.seatunnel.common.utils.ReflectionUtils;
+import org.apache.seatunnel.core.starter.flink.config.FlinkCommon;
 import org.apache.seatunnel.flink.FlinkEnvironment;
 import org.apache.seatunnel.flink.util.TableUtil;
 
@@ -33,7 +33,6 @@
 import org.apache.flink.types.Row;
 
 import java.net.URL;
-import java.net.URLClassLoader;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.BiConsumer;
@@ -47,16 +46,7 @@ public abstract class AbstractPluginExecuteProcessor<T> implements PluginExecute
     protected static final String ENGINE_TYPE = "seatunnel";
     protected static final String PLUGIN_NAME = "plugin_name";
 
-    protected final BiConsumer<ClassLoader, URL> addUrlToClassloader = (classLoader, url) -> {
-        if (classLoader.getClass().getName().endsWith("SafetyNetWrapperClassLoader")) {
-            URLClassLoader c = (URLClassLoader) ReflectionUtils.getField(classLoader, "inner").get();
-            ReflectionUtils.invoke(c, "addURL", url);
-        } else if (classLoader instanceof URLClassLoader) {
-            ReflectionUtils.invoke(classLoader, "addURL", url);
-        } else {
-            throw new RuntimeException("Unsupported classloader: " + classLoader.getClass().getName());
-        }
-    };
+    protected final BiConsumer<ClassLoader, URL> addUrlToClassloader = FlinkCommon.ADD_URL_TO_CLASSLOADER;
 
     protected AbstractPluginExecuteProcessor(FlinkEnvironment flinkEnvironment,
                                              JobContext jobContext,

seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkExecution.java

@@ -22,6 +22,7 @@
 import org.apache.seatunnel.common.config.Common;
 import org.apache.seatunnel.core.starter.exception.TaskExecuteException;
 import org.apache.seatunnel.core.starter.execution.TaskExecution;
+import org.apache.seatunnel.core.starter.flink.config.FlinkCommon;
 import org.apache.seatunnel.core.starter.flink.config.FlinkEnvironmentFactory;
 import org.apache.seatunnel.flink.FlinkEnvironment;
 
@@ -54,10 +55,10 @@ public FlinkExecution(Config config) {
         this.flinkEnvironment = new FlinkEnvironmentFactory(config).getEnvironment();
         JobContext jobContext = new JobContext();
         jobContext.setJobMode(flinkEnvironment.getJobMode());
+        registerPlugin();
         this.sourcePluginExecuteProcessor = new SourceExecuteProcessor(flinkEnvironment, jobContext, config.getConfigList(Constants.SOURCE));
         this.transformPluginExecuteProcessor = new TransformExecuteProcessor(flinkEnvironment, jobContext, config.getConfigList(Constants.TRANSFORM));
         this.sinkPluginExecuteProcessor = new SinkExecuteProcessor(flinkEnvironment, jobContext, config.getConfigList(Constants.SINK));
-        registerPlugin();
     }
 
     @Override
@@ -75,7 +76,7 @@ public void execute() throws TaskExecuteException {
         }
     }
 
-    private void registerPlugin(){
+    private void registerPlugin() {
         List<URL> pluginsJarDependencies = Common.getPluginsJarDependencies().stream()
             .map(Path::toUri)
             .map(uri -> {
@@ -86,6 +87,9 @@ private void registerPlugin(){
                 }
             })
             .collect(Collectors.toList());
+
+        pluginsJarDependencies.forEach(url -> FlinkCommon.ADD_URL_TO_CLASSLOADER.accept(Thread.currentThread().getContextClassLoader(), url));
+
         flinkEnvironment.registerPlugin(pluginsJarDependencies);
     }
 }

seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/AbstractContainer.java

@@ -64,6 +64,10 @@ public AbstractContainer() {
 
     protected abstract List<String> getExtraStartShellCommands();
 
+    protected void executeExtraCommands(GenericContainer<?> container) throws IOException, InterruptedException {
+        //do nothing
+    }
+
     protected void copySeaTunnelStarter(GenericContainer<?> container) {
         ContainerUtil.copySeaTunnelStarter(container,
             this.startModuleName,
@@ -81,6 +85,8 @@ protected Container.ExecResult executeJob(GenericContainer<?> container, String
             getConnectorNamePrefix(),
             getConnectorType(),
             getSeaTunnelHomeInContainer());
+        // execute extra commands
+        executeExtraCommands(container);
         return executeCommand(container, confInContainerPath);
     }
 

seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/AbstractSparkContainer.java

@@ -61,7 +61,8 @@ public void before() {
             .withNetworkAliases("spark-master")
             .withExposedPorts()
             .withEnv("SPARK_MODE", "master")
-            .withLogConsumer(new Slf4jLogConsumer(LOG));
+            .withLogConsumer(new Slf4jLogConsumer(LOG))
+            .withCreateContainerCmdModifier(cmd -> cmd.withUser("root"));
         // In most case we can just use standalone mode to execute a spark job, if we want to use cluster mode, we need to
         // start a worker.
         Startables.deepStart(Stream.of(master)).join();
@@ -79,7 +80,7 @@ public void close() {
     @Override
     protected List<String> getExtraStartShellCommands() {
         return Arrays.asList("--master local",
-                             "--deploy-mode client");
+            "--deploy-mode client");
     }
 
     public Container.ExecResult executeSeaTunnelSparkJob(String confFile) throws IOException, InterruptedException {

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/pom.xml

@@ -72,6 +72,7 @@
             <groupId>org.postgresql</groupId>
             <artifactId>postgresql</artifactId>
             <version>${postgresql.version}</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>com.dameng</groupId>
@@ -80,4 +81,4 @@
         </dependency>
     </dependencies>
 
-</project>
+</project>

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/jdbc/FakeSourceToJdbcIT.java

@@ -29,6 +29,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.PostgreSQLContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.lifecycle.Startables;
@@ -47,14 +48,15 @@
 public class FakeSourceToJdbcIT extends FlinkContainer {
     private static final Logger LOGGER = LoggerFactory.getLogger(FakeSourceToJdbcIT.class);
     private PostgreSQLContainer<?> psl;
+    private static final String THIRD_PARTY_PLUGINS_URL = "https://repo1.maven.org/maven2/org/postgresql/postgresql/42.3.3/postgresql-42.3.3.jar";
 
     @SuppressWarnings("checkstyle:MagicNumber")
     @BeforeEach
     public void startPostgreSqlContainer() throws InterruptedException, ClassNotFoundException, SQLException {
         psl = new PostgreSQLContainer<>(DockerImageName.parse("postgres:alpine3.16"))
-                .withNetwork(NETWORK)
-                .withNetworkAliases("postgresql")
-                .withLogConsumer(new Slf4jLogConsumer(LOGGER));
+            .withNetwork(NETWORK)
+            .withNetworkAliases("postgresql")
+            .withLogConsumer(new Slf4jLogConsumer(LOGGER));
         Startables.deepStart(Stream.of(psl)).join();
         LOGGER.info("PostgreSql container started");
         Class.forName(psl.getDriverClassName());
@@ -70,8 +72,8 @@ private void initializeJdbcTable() {
         try (Connection connection = DriverManager.getConnection(psl.getJdbcUrl(), psl.getUsername(), psl.getPassword())) {
             Statement statement = connection.createStatement();
             String sql = "CREATE TABLE test (\n" +
-                    "  name varchar(255) NOT NULL\n" +
-                    ")";
+                "  name varchar(255) NOT NULL\n" +
+                ")";
             statement.execute(sql);
         } catch (SQLException e) {
             throw new RuntimeException("Initializing PostgreSql table failed!", e);
@@ -101,4 +103,10 @@ public void closePostgreSqlContainer() {
             psl.stop();
         }
     }
+
+    @Override
+    protected void executeExtraCommands(GenericContainer<?> container) throws IOException, InterruptedException {
+        Container.ExecResult extraCommands = container.execInContainer("bash", "-c", "mkdir -p /tmp/flink/seatunnel/plugins/Jdbc/lib && cd /tmp/flink/seatunnel/plugins/Jdbc/lib && curl -O " + THIRD_PARTY_PLUGINS_URL);
+        Assertions.assertEquals(0, extraCommands.getExitCode());
+    }
 }

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/jdbc/JdbcDmdbIT.java

@@ -51,8 +51,7 @@ public class JdbcDmdbIT extends FlinkContainer {
     private static final String DOCKER_IMAGE = "laglangyue/dmdb8";
     private static final String DRIVER_CLASS = "dm.jdbc.driver.DmDriver";
     private static final String HOST = "flink_e2e_dmdb";
-    private static final String LOCAL_HOST = "localhost";
-    private static final String URL = "jdbc:dm://" + LOCAL_HOST + ":5236";
+    private static final String URL = "jdbc:dm://%s:5236";
     private static final String USERNAME = "SYSDBA";
     private static final String PASSWORD = "SYSDBA";
     private static final String DATABASE = "SYSDBA";
@@ -81,7 +80,8 @@ public void startDmdbContainer() throws ClassNotFoundException, SQLException {
     }
 
     private void initializeJdbcConnection() throws SQLException {
-        jdbcConnection = DriverManager.getConnection(URL, USERNAME, PASSWORD);
+        jdbcConnection = DriverManager.getConnection(String.format(
+            URL, dbServer.getHost()), USERNAME, PASSWORD);
     }
 
     /**
@@ -110,8 +110,7 @@ private void initializeJdbcTable() {
     }
 
     private void assertHasData(String table) {
-        try (Connection connection = DriverManager.getConnection(URL, USERNAME, PASSWORD)) {
-            Statement statement = connection.createStatement();
+        try (Statement statement = jdbcConnection.createStatement();) {
             String sql = String.format("select * from %s.%s limit 1", DATABASE, table);
             ResultSet source = statement.executeQuery(sql);
             Assertions.assertTrue(source.next());