[Improve][e2e] Container only copy required connector jars (apache#2675)

* [Improve][e2e] flink container only copy required connector jars

* rename flink-e2e-common

* remove useless imported

* [Improve][e2e] flink sql container refactoring

* remove useless imported

* remove useless

* [Improve][e2e] spark container only copy required connector jars

* change for code review

* Use e2e-common module directly

* checkstyle

* code format

seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/ReadonlyConfig.java

@@ -66,18 +66,24 @@ public <T> T get(Option<T> option) {
         return getOptional(option).orElseGet(option::defaultValue);
     }
 
-    @SuppressWarnings("MagicNumber")
     public Map<String, String> toMap() {
         if (confData.isEmpty()) {
             return Collections.emptyMap();
         }
 
+        Map<String, String> result = new HashMap<>();
+        toMap(result);
+        return result;
+    }
+
+    public void toMap(Map<String, String> result) {
+        if (confData.isEmpty()) {
+            return;
+        }
         Map<String, Object> flatteningMap = flatteningMap(confData);
-        Map<String, String> result = new HashMap<>((flatteningMap.size() << 2) / 3 + 1);
         for (Map.Entry<String, Object> entry : flatteningMap.entrySet()) {
             result.put(entry.getKey(), convertToJsonString(entry.getValue()));
         }
-        return result;
     }
 
     @SuppressWarnings("unchecked")

seatunnel-e2e/pom.xml

@@ -27,15 +27,17 @@
     <packaging>pom</packaging>
 
     <modules>
-        <module>seatunnel-flink-e2e</module>
-        <module>seatunnel-spark-e2e</module>
+        <module>seatunnel-e2e-common</module>
         <module>seatunnel-flink-connector-v2-e2e</module>
-        <module>seatunnel-spark-connector-v2-e2e</module>
+        <module>seatunnel-flink-e2e</module>
         <module>seatunnel-flink-sql-e2e</module>
+        <module>seatunnel-spark-connector-v2-e2e</module>
+        <module>seatunnel-spark-e2e</module>
     </modules>
 
     <properties>
         <junit4.version>4.13.2</junit4.version>
+        <maven-jar-plugin.version>2.4</maven-jar-plugin.version>
     </properties>
     <dependencies>
         <dependency>

seatunnel-e2e/seatunnel-e2e-common/pom.xml

@@ -0,0 +1,55 @@
+<?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">
+    <parent>
+        <artifactId>seatunnel-e2e</artifactId>
+        <groupId>org.apache.seatunnel</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>seatunnel-e2e-common</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <version>${maven-jar-plugin.version}</version>
+                <configuration>
+                    <skip>false</skip>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>test-jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

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

@@ -0,0 +1,103 @@
+/*
+ * 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.e2e.common;
+
+import static org.apache.seatunnel.e2e.common.ContainerUtil.PROJECT_ROOT_PATH;
+import static org.apache.seatunnel.e2e.common.ContainerUtil.adaptPathForWin;
+import static org.apache.seatunnel.e2e.common.ContainerUtil.copyConfigFileToContainer;
+import static org.apache.seatunnel.e2e.common.ContainerUtil.copyConnectorJarToContainer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+
+public abstract class AbstractContainer {
+    protected static final Logger LOG = LoggerFactory.getLogger(AbstractContainer.class);
+    protected static final String START_ROOT_MODULE_NAME = "seatunnel-core";
+
+    protected final String startModuleName;
+
+    protected final String startModuleFullPath;
+
+    public AbstractContainer() {
+        String[] modules = getStartModulePath().split(File.separator);
+        this.startModuleName = modules[modules.length - 1];
+        this.startModuleFullPath = PROJECT_ROOT_PATH + File.separator +
+            START_ROOT_MODULE_NAME + File.separator + getStartModulePath();
+    }
+
+    protected abstract String getDockerImage();
+
+    protected abstract String getStartModulePath();
+
+    protected abstract String getStartShellName();
+
+    protected abstract String getConnectorModulePath();
+
+    protected abstract String getConnectorType();
+
+    protected abstract String getConnectorNamePrefix();
+
+    protected abstract String getSeaTunnelHomeInContainer();
+
+    protected abstract List<String> getExtraStartShellCommands();
+
+    protected void copySeaTunnelStarter(GenericContainer<?> container) {
+        String[] modules = getStartModulePath().split(File.separator);
+        final String startModuleName = modules[modules.length - 1];
+        ContainerUtil.copySeaTunnelStarter(container,
+            startModuleName,
+            PROJECT_ROOT_PATH + File.separator + START_ROOT_MODULE_NAME + File.separator + getStartModulePath(),
+            getSeaTunnelHomeInContainer(),
+            getStartShellName());
+    }
+
+    protected Container.ExecResult executeJob(GenericContainer<?> container, String confFile) throws IOException, InterruptedException {
+        final String confInContainerPath = copyConfigFileToContainer(container, confFile);
+        // copy connectors
+        copyConnectorJarToContainer(container,
+            confFile,
+            getConnectorModulePath(),
+            getConnectorNamePrefix(),
+            getConnectorType(),
+            getSeaTunnelHomeInContainer());
+        return executeCommand(container, confInContainerPath);
+    }
+
+    protected Container.ExecResult executeCommand(GenericContainer<?> container, String configPath) throws IOException, InterruptedException {
+        final List<String> command = new ArrayList<>();
+        String binPath = Paths.get(getSeaTunnelHomeInContainer(), "bin", getStartShellName()).toString();
+        // base command
+        command.add(adaptPathForWin(binPath));
+        command.add("--config");
+        command.add(adaptPathForWin(configPath));
+        command.addAll(getExtraStartShellCommands());
+
+        Container.ExecResult execResult = container.execInContainer("bash", "-c", String.join(" ", command));
+        LOG.info(execResult.getStdout());
+        LOG.error(execResult.getStderr());
+        return execResult;
+    }
+}

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

@@ -0,0 +1,121 @@
+/*
+ * 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.e2e.common;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Stream;
+
+/**
+ * This class is the base class of FlinkEnvironment test.
+ * The before method will create a Flink cluster, and after method will close the Flink cluster.
+ * You can use {@link AbstractFlinkContainer#executeSeaTunnelFlinkJob} to submit a seatunnel config and run a seatunnel job.
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public abstract class AbstractFlinkContainer extends AbstractContainer {
+
+    protected static final Logger LOG = LoggerFactory.getLogger(AbstractFlinkContainer.class);
+
+    protected static final String FLINK_SEATUNNEL_HOME = "/tmp/flink/seatunnel";
+
+    protected static final Network NETWORK = Network.newNetwork();
+
+    protected static final List<String> DEFAULT_FLINK_PROPERTIES = Arrays.asList(
+        "jobmanager.rpc.address: jobmanager",
+        "taskmanager.numberOfTaskSlots: 10",
+        "parallelism.default: 4",
+        "env.java.opts: -Doracle.jdbc.timezoneAsRegion=false");
+
+    protected static final String DEFAULT_DOCKER_IMAGE = "flink:1.13.6-scala_2.11";
+
+    protected GenericContainer<?> jobManager;
+    protected GenericContainer<?> taskManager;
+
+    @Override
+    protected String getDockerImage() {
+        return DEFAULT_DOCKER_IMAGE;
+    }
+
+    @Override
+    protected String getSeaTunnelHomeInContainer() {
+        return FLINK_SEATUNNEL_HOME;
+    }
+
+    @BeforeAll
+    public void before() {
+        final String dockerImage = getDockerImage();
+        final String properties = String.join("\n", getFlinkProperties());
+        jobManager = new GenericContainer<>(dockerImage)
+            .withCommand("jobmanager")
+            .withNetwork(NETWORK)
+            .withNetworkAliases("jobmanager")
+            .withExposedPorts()
+            .withEnv("FLINK_PROPERTIES", properties)
+            .withLogConsumer(new Slf4jLogConsumer(LOG));
+
+        taskManager =
+            new GenericContainer<>(dockerImage)
+                .withCommand("taskmanager")
+                .withNetwork(NETWORK)
+                .withNetworkAliases("taskmanager")
+                .withEnv("FLINK_PROPERTIES", properties)
+                .dependsOn(jobManager)
+                .withLogConsumer(new Slf4jLogConsumer(LOG));
+
+        Startables.deepStart(Stream.of(jobManager)).join();
+        Startables.deepStart(Stream.of(taskManager)).join();
+        copySeaTunnelStarter(jobManager);
+        LOG.info("Flink containers are started.");
+    }
+
+    protected List<String> getFlinkProperties() {
+        return DEFAULT_FLINK_PROPERTIES;
+    }
+
+    @AfterAll
+    public void close() {
+        if (taskManager != null) {
+            taskManager.stop();
+        }
+        if (jobManager != null) {
+            jobManager.stop();
+        }
+    }
+
+    @Override
+    protected List<String> getExtraStartShellCommands() {
+        return Collections.emptyList();
+    }
+
+    public Container.ExecResult executeSeaTunnelFlinkJob(String confFile) throws IOException, InterruptedException {
+        return executeJob(jobManager, confFile);
+    }
+}