Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Improve][e2e] Unified e2e IT for Redis #2924

Merged
merged 3 commits into from
Sep 28, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Original file line number Diff line number Diff line change
@@ -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"));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
}
1 change: 1 addition & 0 deletions seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Expand Down Expand Up @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
@@ -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 "";
}
Loading