[Improve][Connector-V2][MongoDB] Refactor MongoDB connector e2e test …

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/pom.xml → seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/pom.xml

@@ -17,31 +17,28 @@
         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">
-    <modelVersion>4.0.0</modelVersion>
     <parent>
         <groupId>org.apache.seatunnel</groupId>
-        <artifactId>seatunnel-flink-connector-v2-e2e</artifactId>
+        <artifactId>seatunnel-connector-v2-e2e</artifactId>
         <version>${revision}</version>
     </parent>
+    <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>connector-mongodb-flink-e2e</artifactId>
+    <artifactId>connector-mongodb-e2e</artifactId>
 
     <dependencies>
+
+        <!-- SeaTunnel connectors -->
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
-            <artifactId>connector-flink-e2e-base</artifactId>
+            <artifactId>connector-mongodb</artifactId>
             <version>${project.version}</version>
-            <classifier>tests</classifier>
-            <type>test-jar</type>
             <scope>test</scope>
         </dependency>
-
-        <!-- SeaTunnel connectors -->
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
-            <artifactId>connector-mongodb</artifactId>
+            <artifactId>connector-assert</artifactId>
             <version>${project.version}</version>
-            <scope>test</scope>
         </dependency>
     </dependencies>
 </project>

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-mongodb-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/mongodb/MongodbIT.java → seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.seatunnel.e2e.spark.v2.mongodb;
+package org.apache.seatunnel.e2e.connector.v2.mongodb;
 
 import static java.net.HttpURLConnection.HTTP_OK;
 import static java.net.HttpURLConnection.HTTP_UNAUTHORIZED;
@@ -31,20 +31,20 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.connectors.seatunnel.mongodb.data.DefaultSerializer;
 import org.apache.seatunnel.connectors.seatunnel.mongodb.data.Serializer;
-import org.apache.seatunnel.e2e.spark.SparkContainer;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
 
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
 import com.mongodb.client.MongoCollection;
-import com.mongodb.client.MongoCursor;
-import com.mongodb.client.model.Sorts;
 import lombok.extern.slf4j.Slf4j;
 import org.awaitility.Awaitility;
 import org.bson.Document;
-import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
 import org.testcontainers.containers.Container;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
@@ -57,103 +57,61 @@
 import java.math.BigDecimal;
 import java.time.Duration;
 import java.time.LocalDate;
-import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 @Slf4j
-public class MongodbIT extends SparkContainer {
+public class MongodbIT extends TestSuiteBase implements TestResource {
 
     private static final String MONGODB_IMAGE = "mongo:latest";
-    private static final String MONGODB_CONTAINER_HOST = "spark_e2e_mongodb";
+    private static final String MONGODB_CONTAINER_HOST = "e2e_mongodb";
     private static final int MONGODB_PORT = 27017;
     private static final String MONGODB_DATABASE = "test_db";
     private static final String MONGODB_SOURCE_TABLE = "source_table";
-    private static final String MONGODB_SINK_TABLE = "sink_table";
 
-    private static final List<Document> TEST_DATASET = generateTestDataSet();
+    private static final List<Document> TEST_DATASET = generateTestDataSet(0, 10);
 
-    private MongoClient client;
     private GenericContainer<?> mongodbContainer;
+    private MongoClient client;
 
-    @BeforeEach
-    public void startMongoContainer() {
-        DockerImageName imageName = DockerImageName.parse(MONGODB_IMAGE);
-        mongodbContainer = new GenericContainer<>(imageName)
-            .withNetwork(NETWORK)
-            .withNetworkAliases(MONGODB_CONTAINER_HOST)
-            .withExposedPorts(MONGODB_PORT)
-            .waitingFor(new HttpWaitStrategy()
-                .forPort(MONGODB_PORT)
-                .forStatusCodeMatching(response -> response == HTTP_OK || response == HTTP_UNAUTHORIZED)
-                .withStartupTimeout(Duration.ofMinutes(2)))
-            .withLogConsumer(new Slf4jLogConsumer(DockerLoggerFactory.getLogger(MONGODB_IMAGE)));
-        Startables.deepStart(Stream.of(mongodbContainer)).join();
-        log.info("Mongodb container started");
+    @TestTemplate
+    public void testMongodbSourceToAssertSink(TestContainer container) throws IOException, InterruptedException {
+        Container.ExecResult execResult = container.executeJob("/mongodb_source_to_assert.conf");
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
+    }
 
-        Awaitility.given().ignoreExceptions()
-            .atLeast(100, TimeUnit.MILLISECONDS)
-            .pollInterval(500, TimeUnit.MILLISECONDS)
-            .atMost(180, TimeUnit.SECONDS)
-            .untilAsserted(this::initConnection);
-        this.initSourceData();
+    @TestTemplate
+    public void testMongodb(TestContainer container) throws IOException, InterruptedException {
+        Container.ExecResult execResult = container.executeJob("/mongodb_source_and_sink.conf");
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
     }
 
-    @Test
-    public void testMongodb() throws IOException, InterruptedException {
-        Container.ExecResult execResult = executeSeaTunnelSparkJob("/mongodb/mongodb_source_and_sink.conf");
-        Assertions.assertEquals(0, execResult.getExitCode());
-        Assertions.assertIterableEquals(
-            TEST_DATASET.stream()
-                .map(e -> {
-                    e.remove("_id");
-                    return e;
-                })
-                .collect(Collectors.toList()),
-            readSinkData().stream()
-                .map(e -> {
-                    e.remove("_id");
-                    return e;
-                })
-                .collect(Collectors.toList()));
+    @TestTemplate
+    public void testMongodbMatchQuery(TestContainer container) throws IOException, InterruptedException {
+        Container.ExecResult execResult = container.executeJob("/mongodb_source_matchQuery_and_sink.conf");
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
     }
 
     public void initConnection() {
         String host = mongodbContainer.getContainerIpAddress();
         int port = mongodbContainer.getFirstMappedPort();
         String url = String.format("mongodb://%s:%d/%s", host, port, MONGODB_DATABASE);
-
         client = MongoClients.create(url);
     }
 
-    private void initSourceData() {
+    private void initSourceData(String database, String table, List<Document> documents) {
         MongoCollection<Document> sourceTable = client
-            .getDatabase(MONGODB_DATABASE)
-            .getCollection(MONGODB_SOURCE_TABLE);
+            .getDatabase(database)
+            .getCollection(table);
 
         sourceTable.deleteMany(new Document());
-        sourceTable.insertMany(TEST_DATASET);
-    }
-
-    private List<Document> readSinkData() {
-        MongoCollection<Document> sinkTable = client
-            .getDatabase(MONGODB_DATABASE)
-            .getCollection(MONGODB_SINK_TABLE);
-        MongoCursor<Document> cursor = sinkTable.find()
-            .sort(Sorts.ascending("id"))
-            .cursor();
-        List<Document> documents = new ArrayList<>();
-        while (cursor.hasNext()) {
-            documents.add(cursor.next());
-        }
-        return documents;
+        sourceTable.insertMany(documents);
     }
 
-    private static List<Document> generateTestDataSet() {
+    private static List<Document> generateTestDataSet(int start, int end) {
         SeaTunnelRowType seatunnelRowType = new SeaTunnelRowType(
             new String[]{
                 "id",
@@ -169,8 +127,7 @@ private static List<Document> generateTestDataSet() {
                 "c_double",
                 "c_decimal",
                 "c_bytes",
-                "c_date",
-                "c_timestamp"
+                "c_date"
             },
             new SeaTunnelDataType[]{
                 BasicType.LONG_TYPE,
@@ -186,14 +143,13 @@ private static List<Document> generateTestDataSet() {
                 BasicType.DOUBLE_TYPE,
                 new DecimalType(2, 1),
                 PrimitiveByteArrayType.INSTANCE,
-                LocalTimeType.LOCAL_DATE_TYPE,
-                LocalTimeType.LOCAL_DATE_TIME_TYPE
+                LocalTimeType.LOCAL_DATE_TYPE
             }
         );
         Serializer serializer = new DefaultSerializer(seatunnelRowType);
 
         List<Document> documents = new ArrayList<>();
-        for (int i = 0; i < 100; i++) {
+        for (int i = start; i < end; i++) {
             SeaTunnelRow row = new SeaTunnelRow(
                 new Object[]{
                     Long.valueOf(i),
@@ -209,16 +165,40 @@ private static List<Document> generateTestDataSet() {
                     Double.parseDouble("1.1"),
                     BigDecimal.valueOf(11, 1),
                     "test".getBytes(),
-                    LocalDate.now(),
-                    LocalDateTime.now()
+                    LocalDate.now()
                 });
             documents.add(serializer.serialize(row));
         }
         return documents;
     }
 
-    @AfterEach
-    public void closeMongoContainer() {
+    @BeforeAll
+    @Override
+    public void startUp() {
+        DockerImageName imageName = DockerImageName.parse(MONGODB_IMAGE);
+        mongodbContainer = new GenericContainer<>(imageName)
+            .withNetwork(NETWORK)
+            .withNetworkAliases(MONGODB_CONTAINER_HOST)
+            .withExposedPorts(MONGODB_PORT)
+            .waitingFor(new HttpWaitStrategy()
+                .forPort(MONGODB_PORT)
+                .forStatusCodeMatching(response -> response == HTTP_OK || response == HTTP_UNAUTHORIZED)
+                .withStartupTimeout(Duration.ofMinutes(2)))
+            .withLogConsumer(new Slf4jLogConsumer(DockerLoggerFactory.getLogger(MONGODB_IMAGE)));
+        Startables.deepStart(Stream.of(mongodbContainer)).join();
+        log.info("Mongodb container started");
+
+        Awaitility.given().ignoreExceptions()
+            .atLeast(100, TimeUnit.MILLISECONDS)
+            .pollInterval(500, TimeUnit.MILLISECONDS)
+            .atMost(180, TimeUnit.SECONDS)
+            .untilAsserted(this::initConnection);
+        this.initSourceData(MONGODB_DATABASE, MONGODB_SOURCE_TABLE, TEST_DATASET);
+    }
+
+    @AfterAll
+    @Override
+    public void tearDown() {
         if (client != null) {
             client.close();
         }

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-mongodb-spark-e2e/src/test/resources/mongodb/mongodb_source_and_sink.conf → seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mongodb_source_and_sink.conf

@@ -16,20 +16,25 @@
 #
 
 ######
-###### This config file is a demonstration of streaming processing in seatunnel config
+###### This config file is a demonstration of batch processing in seatunnel config
 ######
 
 env {
   # You can set flink configuration here
   execution.parallelism = 1
   job.mode = "BATCH"
-  #execution.checkpoint.interval = 10000
-  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+
+  # You can set spark configuration here
+  spark.app.name = "SeaTunnel"
+  spark.executor.instances = 1
+  spark.executor.cores = 1
+  spark.executor.memory = "1g"
+  spark.master = local
 }
 
 source {
   MongoDB {
-    uri = "mongodb://spark_e2e_mongodb:27017/test_db?retryWrites=true&writeConcern=majority"
+    uri = "mongodb://e2e_mongodb:27017/test_db?retryWrites=true&writeConcern=majority"
     database = "test_db"
     collection = "source_table"
     schema = {
@@ -48,7 +53,6 @@ source {
         c_decimal = "decimal(2, 1)"
         c_bytes = bytes
         c_date = date
-        c_timestamp = timestamp
       }
     }
   }
@@ -59,7 +63,7 @@ transform {
 
 sink {
   MongoDB {
-    uri = "mongodb://spark_e2e_mongodb:27017/test_db?retryWrites=true&writeConcern=majority"
+    uri = "mongodb://e2e_mongodb:27017/test_db?retryWrites=true&writeConcern=majority"
     database = "test_db"
     collection = "sink_table"
   }

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/resources/mongodb/mongodb_source_matchQuery_and_sink.conf → seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mongodb_source_matchQuery_and_sink.conf

@@ -16,20 +16,25 @@
 #
 
 ######
-###### This config file is a demonstration of streaming processing in seatunnel config
+###### This config file is a demonstration of batch processing in seatunnel config
 ######
 
 env {
   # You can set flink configuration here
   execution.parallelism = 1
   job.mode = "BATCH"
-  #execution.checkpoint.interval = 10000
-  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+
+  # You can set spark configuration here
+  spark.app.name = "SeaTunnel"
+  spark.executor.instances = 1
+  spark.executor.cores = 1
+  spark.executor.memory = "1g"
+  spark.master = local
 }
 
 source {
   MongoDB {
-    uri = "mongodb://flink_e2e_mongodb:27017/test_db?retryWrites=true&writeConcern=majority"
+    uri = "mongodb://e2e_mongodb:27017/test_db?retryWrites=true&writeConcern=majority"
     database = "test_db"
     collection = "source_matchQuery_table"
     matchQuery = "{"id":3}"
@@ -59,7 +64,7 @@ transform {
 
 sink {
   MongoDB {
-    uri = "mongodb://flink_e2e_mongodb:27017/test_db?retryWrites=true&writeConcern=majority"
+    uri = "mongodb://e2e_mongodb:27017/test_db?retryWrites=true&writeConcern=majority"
     database = "test_db"
     collection = "sink_matchQuery_table"
   }