[Improve][e2e] Unified e2e IT for Redis

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/pom.xml

@@ -17,8 +17,8 @@
          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-connector-v2-e2e</artifactId>
         <groupId>org.apache.seatunnel</groupId>
+        <artifactId>seatunnel-connector-v2-e2e</artifactId>
         <version>${revision}</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-assert-e2e/src/test/resources/assertion/fakesource_to_assert.conf

@@ -14,9 +14,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-######
-###### This config file is a demonstration of streaming processing in seatunnel config
-######
 
 env {
   execution.parallelism = 1

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

@@ -18,36 +18,20 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <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-redis-flink-e2e</artifactId>
+    <artifactId>connector-redis-e2e</artifactId>
 
     <dependencies>
-        <dependency>
-            <groupId>org.apache.seatunnel</groupId>
-            <artifactId>connector-flink-e2e-base</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-redis</artifactId>
             <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.apache.seatunnel</groupId>
-            <artifactId>connector-assert</artifactId>
-            <version>${project.version}</version>
-            <scope>test</scope>
-        </dependency>
     </dependencies>
-
 </project>

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisIT.java

@@ -0,0 +1,184 @@
+/*
+ * 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.connector.redis;
+
+import org.apache.seatunnel.api.table.type.ArrayType;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.DecimalType;
+import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.type.MapType;
+import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+import org.apache.seatunnel.format.json.JsonSerializationSchema;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+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;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+import redis.clients.jedis.Jedis;
+
+import java.io.IOException;
+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.stream.Stream;
+
+import scala.Tuple2;
+
+@DisabledOnContainer(value = "spark:2.4.3", disabledReason = "json-format conflicts with the Jackson version of Spark-2.4.3, see:https://github.com/apache/incubator-seatunnel/issues/2929")
+@Slf4j
+public class RedisIT extends TestSuiteBase implements TestResource {
+    private static final String IMAGE = "redis:latest";
+    private static final String HOST = "redis-e2e";
+    private static final int PORT = 6379;
+    private static final String PASSWORD = "SeaTunnel";
+
+    private static final Tuple2<SeaTunnelRowType, List<SeaTunnelRow>> TEST_DATASET = generateTestDataSet();
+
+    private GenericContainer<?> redisContainer;
+
+    private Jedis jedis;
+
+    @BeforeAll
+    @Override
+    public void startUp() throws Exception {
+        this.redisContainer = new GenericContainer<>(DockerImageName.parse(IMAGE))
+            .withNetwork(NETWORK)
+            .withNetworkAliases(HOST)
+            .withExposedPorts(PORT)
+            .withLogConsumer(new Slf4jLogConsumer(log))
+            .withCommand(String.format("redis-server --requirepass %s", PASSWORD))
+            .waitingFor(new HostPortWaitStrategy()
+                .withStartupTimeout(Duration.ofMinutes(2)));
+        Startables.deepStart(Stream.of(redisContainer)).join();
+        log.info("Redis container started");
+        this.initJedis();
+        this.initSourceData();
+    }
+
+    private void initSourceData() {
+        JsonSerializationSchema jsonSerializationSchema = new JsonSerializationSchema(TEST_DATASET._1());
+        List<SeaTunnelRow> rows = TEST_DATASET._2();
+        for (int i = 0; i < rows.size(); i++) {
+            jedis.set("key_test" + i, new String(jsonSerializationSchema.serialize(rows.get(i))));
+        }
+    }
+
+    private static Tuple2<SeaTunnelRowType, List<SeaTunnelRow>> generateTestDataSet() {
+        SeaTunnelRowType rowType = new SeaTunnelRowType(
+            new String[]{
+                "id",
+                "c_map",
+                "c_array",
+                "c_string",
+                "c_boolean",
+                "c_tinyint",
+                "c_smallint",
+                "c_int",
+                "c_bigint",
+                "c_float",
+                "c_double",
+                "c_decimal",
+                "c_bytes",
+                "c_date",
+                "c_timestamp"
+            },
+            new SeaTunnelDataType[]{
+                BasicType.LONG_TYPE,
+                new MapType<>(BasicType.STRING_TYPE, BasicType.SHORT_TYPE),
+                ArrayType.BYTE_ARRAY_TYPE,
+                BasicType.STRING_TYPE,
+                BasicType.BOOLEAN_TYPE,
+                BasicType.BYTE_TYPE,
+                BasicType.SHORT_TYPE,
+                BasicType.INT_TYPE,
+                BasicType.LONG_TYPE,
+                BasicType.FLOAT_TYPE,
+                BasicType.DOUBLE_TYPE,
+                new DecimalType(2, 1),
+                PrimitiveByteArrayType.INSTANCE,
+                LocalTimeType.LOCAL_DATE_TYPE,
+                LocalTimeType.LOCAL_DATE_TIME_TYPE
+            }
+        );
+
+        List<SeaTunnelRow> rows = new ArrayList<>();
+        for (int i = 0; i < 100; i++) {
+            SeaTunnelRow row = new SeaTunnelRow(
+                new Object[]{
+                    Long.valueOf(i),
+                    Collections.singletonMap("key", Short.parseShort("1")),
+                    new Byte[]{Byte.parseByte("1")},
+                    "string",
+                    Boolean.FALSE,
+                    Byte.parseByte("1"),
+                    Short.parseShort("1"),
+                    Integer.parseInt("1"),
+                    Long.parseLong("1"),
+                    Float.parseFloat("1.1"),
+                    Double.parseDouble("1.1"),
+                    BigDecimal.valueOf(11, 1),
+                    "test".getBytes(),
+                    LocalDate.now(),
+                    LocalDateTime.now()
+                });
+            rows.add(row);
+        }
+        return Tuple2.apply(rowType, rows);
+    }
+
+    private void initJedis() {
+        Jedis jedis = new Jedis(redisContainer.getHost(), redisContainer.getFirstMappedPort());
+        jedis.auth(PASSWORD);
+        this.jedis = jedis;
+    }
+
+    @AfterAll
+    @Override
+    public void tearDown() throws Exception {
+        jedis.close();
+        redisContainer.close();
+    }
+
+    @TestTemplate
+    public void testRedis(TestContainer container) throws IOException, InterruptedException {
+        Container.ExecResult execResult = container.executeJob("/redis-to-redis.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(100, jedis.llen("key_list"));
+        // Clear data to prevent data duplication in the next TestContainer
+        jedis.del("key_list");
+        Assertions.assertEquals(0, jedis.llen("key_list"));
+    }
+}

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-redis-spark-e2e/src/test/resources/log4j.properties → seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/redis-to-redis.conf

@@ -14,9 +14,38 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-# Set everything to be logged to the console
-log4j.rootCategory=INFO, console
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.target=System.err
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
+
+env {
+    execution.parallelism = 1
+    job.mode = "BATCH"
+
+    #spark config
+    spark.app.name = "SeaTunnel"
+    spark.executor.instances = 2
+    spark.executor.cores = 1
+    spark.executor.memory = "1g"
+    spark.master = local
+}
+
+source {
+    Redis {
+        host = "redis-e2e"
+        port = 6379
+        auth = "SeaTunnel"
+        keys = "key_test*"
+        data_type = key
+    }
+}
+
+transform {
+}
+
+sink {
+    Redis {
+        host = "redis-e2e"
+        port = 6379
+        auth = "SeaTunnel"
+        key = "key_list"
+        data_type = list
+    }
+}

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

@@ -26,6 +26,7 @@
     <modules>
         <module>connector-assert-e2e</module>
         <module>connector-jdbc-it</module>
+        <module>connector-redis-e2e</module>
     </modules>
 
     <artifactId>seatunnel-connector-v2-e2e</artifactId>

seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/junit/ContainerTestingExtension.java

@@ -25,10 +25,14 @@
 import org.junit.jupiter.api.extension.BeforeAllCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
 import org.junit.platform.commons.support.AnnotationSupport;
+import org.junit.platform.commons.util.AnnotationUtils;
 
 import java.lang.annotation.Annotation;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
+import java.util.stream.Collectors;
 
 public class ContainerTestingExtension implements BeforeAllCallback, AfterAllCallback {
     public static final ExtensionContext.Namespace TEST_RESOURCE_NAMESPACE =
@@ -57,7 +61,15 @@ public void beforeAll(ExtensionContext context) throws Exception {
             TestContainersFactory.class);
 
         checkExactlyOneAnnotatedField(containersFactories, TestContainers.class);
-        List<TestContainer> testContainers = containersFactories.get(0).create();
+
+        // Filters disabled containers
+        final List<String> disabledContainers = new ArrayList<>();
+        AnnotationUtils.findAnnotation(context.getRequiredTestInstance().getClass(), DisabledOnContainer.class)
+            .ifPresent(annotation -> Collections.addAll(disabledContainers, annotation.value()));
+        List<TestContainer> testContainers = containersFactories.get(0).create()
+            .stream()
+            .filter(container -> !disabledContainers.contains(container.identifier()))
+            .collect(Collectors.toList());
         context.getStore(TEST_RESOURCE_NAMESPACE)
             .put(TEST_CONTAINERS_STORE_KEY, testContainers);
     }

seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/junit/DisabledOnContainer.java

@@ -0,0 +1,44 @@
+/*
+ * 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.junit;
+
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target({ ElementType.TYPE, ElementType.METHOD })
+@Retention(RetentionPolicy.RUNTIME)
+public @interface DisabledOnContainer {
+
+    /**
+     * {@link TestContainer#identifier()}
+     */
+    String[] value();
+
+    /**
+     * Custom reason to provide if the test container is disabled.
+     *
+     * <p>If a custom reason is supplied, it will be combined with the default
+     * reason for this annotation. If a custom reason is not supplied, the default
+     * reason will be used.
+     */
+    String disabledReason() default "";
+}