Skip to content

[Improve][E2E] Improve redis test for delete hash #9946

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.

Sign up for GitHub

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

Open
wants to merge 3 commits into
base: dev
Choose a base branch
from
Open

[Improve][E2E] Improve redis test for delete hash #9946

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4e93d64
[improve] Improve redis test for delete hash
chl-wxp Oct 14, 2025
b215732
[improve] update test
chl-wxp Oct 16, 2025
4c857b2
[feature] update writer
chl-wxp Oct 20, 2025
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
222 changes: 214 additions & 8 deletions ...s-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,16 +18,26 @@

import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;

import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.Column;
import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.catalog.TableSchema;
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.RowKind;
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.common.utils.JsonUtils;
import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisDataType;
import org.apache.seatunnel.connectors.seatunnel.redis.config.RedisParameters;
import org.apache.seatunnel.connectors.seatunnel.redis.sink.RedisSinkWriter;
import org.apache.seatunnel.e2e.common.TestResource;
import org.apache.seatunnel.e2e.common.TestSuiteBase;
import org.apache.seatunnel.e2e.common.container.EngineType;
Expand All @@ -38,6 +48,7 @@
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
Expand All @@ -50,6 +61,9 @@
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.DockerLoggerFactory;

import com.github.dockerjava.api.model.ExposedPort;
import com.github.dockerjava.api.model.PortBinding;
import com.github.dockerjava.api.model.Ports;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;

Expand All @@ -72,6 +86,7 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static org.apache.seatunnel.connectors.seatunnel.redis.config.RedisBaseOptions.CONNECTOR_IDENTITY;
import static org.awaitility.Awaitility.await;

@Slf4j
Expand All @@ -98,12 +113,20 @@ public void startUp() {
.withNetwork(NETWORK)
.withNetworkAliases(host)
.withExposedPorts(port)
.withCreateContainerCmdModifier(
cmd ->
cmd.getHostConfig()
.withPortBindings(
new PortBinding(
Ports.Binding.bindPort(port),
new ExposedPort(port))))
.withLogConsumer(
new Slf4jLogConsumer(DockerLoggerFactory.getLogger(imageName)))
.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();
Expand Down Expand Up @@ -610,14 +633,24 @@ public void testCustomHashKeyAndValueWriteRedis(TestContainer container)
jedis.del("custom-hash-check");
}

@TestTemplate
public void testFakeToRedisDeleteHashTest(TestContainer container)
throws IOException, InterruptedException {
Container.ExecResult execResult =
container.executeJob("/fake-to-redis-test-delete-hash.conf");
Assertions.assertEquals(0, execResult.getExitCode());
Assertions.assertEquals(2, jedis.hlen("hash_check"));
jedis.del("hash_check");
@Test
public void testFakeToRedisDeleteHashTest() throws IOException {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please reuse

seatunnel/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/sink/SinkFlowTestUtils.java

Line 41 in 720ed0c

public class SinkFlowTestUtils {
, we already has this tool to verify sink by unit test.

Also you refer some demo

seatunnel/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileTest.java

Line 58 in 720ed0c

public class LocalFileTest {

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

seatunnel/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/sink/SinkFlowTestUtils.java

ok,I will update.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please reuse

seatunnel/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/sink/SinkFlowTestUtils.java

Line 41 in 720ed0c

public class SinkFlowTestUtils {

, we already has this tool to verify sink by unit test.
Also you refer some demo

seatunnel/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileTest.java

Line 58 in 720ed0c

public class LocalFileTest {

updated

String key = "hash_check";
final RedisSinkWriter redisSinkWriter = getRedisSinkWriter(RedisDataType.HASH, key);
List<SeaTunnelRow> rowList =
Arrays.asList(
getSeaTunnelRowInsert1(),
getSeaTunnelRowInsert2(),
getSeaTunnelRowInsert3(),
getSeaTunnelRowUpdateBefore(),
getSeaTunnelRowUpdateAfter(),
getSeaTunnelRowDelete());
for (SeaTunnelRow seaTunnelRow : rowList) {
redisSinkWriter.write(seaTunnelRow);
}
redisSinkWriter.close();
Assertions.assertEquals(2, jedis.hlen(key));
jedis.del(key);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you think converting this e2e test to a unit test would have any impact on overall stability or integration coverage?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it is not necessary to submit configuration files to implement some functional points in redis. Unit testing can be used to satisfy this requirement. In addition, nearly 40 test tasks will slow down the e2e efficiency of this module.

}

@TestTemplate
Expand Down Expand Up @@ -809,4 +842,177 @@ public void testFakeToRedisHashTypeValueIsNullTest(TestContainer container)
}

public abstract RedisContainerInfo getRedisContainerInfo();

private RedisSinkWriter getRedisSinkWriter(RedisDataType dataType, String key) {
return new RedisSinkWriter(
getCatalogTable(0, key).getSeaTunnelRowType(), getRedisParameters(dataType, key));
}

private RedisParameters getRedisParameters(RedisDataType dataType, String key) {
final RedisParameters redisParameters = new RedisParameters();
redisParameters.buildWithConfig(getReadonlyConfig(dataType, key));
return redisParameters;
}

private ReadonlyConfig getReadonlyConfig(RedisDataType dataType, String key) {
Map<String, Object> map = new HashMap<>();
map.put("host", redisContainer.getHost());
map.put("port", port);
map.put("db_num", 0);
map.put("auth", password);
map.put("key", key);
map.put("data_type", dataType.name());
map.put("hash_key_field", "id");
map.put("batch_size", 33);
return ReadonlyConfig.fromMap(map);
}

private SeaTunnelRow getSeaTunnelRowInsert1() {
return new SeaTunnelRow(
new Object[] {
1,
true,
(byte) 1,
(short) 2,
3,
4L,
4.3f,
5.3d,
BigDecimal.valueOf(6.3).setScale(1),
"NEW",
LocalDateTime.parse("2020-02-02T02:02:02")
});
}

private SeaTunnelRow getSeaTunnelRowInsert2() {
return new SeaTunnelRow(
new Object[] {
2,
true,
(byte) 1,
(short) 2,
3,
4L,
4.3f,
5.3d,
BigDecimal.valueOf(6.3).setScale(1),
"NEW",
LocalDateTime.parse("2020-02-02T02:02:02")
});
}

private SeaTunnelRow getSeaTunnelRowInsert3() {
return new SeaTunnelRow(
new Object[] {
3,
true,
(byte) 1,
(short) 2,
3,
4L,
4.3f,
5.3d,
BigDecimal.valueOf(6.3).setScale(1),
"NEW",
LocalDateTime.parse("2020-02-02T02:02:02")
});
}

private SeaTunnelRow getSeaTunnelRowUpdateBefore() {
final SeaTunnelRow seaTunnelRow =
new SeaTunnelRow(
new Object[] {
1,
true,
(byte) 1,
(short) 2,
3,
4L,
4.3f,
5.3d,
BigDecimal.valueOf(6.3).setScale(1),
"NEW",
LocalDateTime.parse("2020-02-02T02:02:02")
});
seaTunnelRow.setRowKind(RowKind.UPDATE_BEFORE);
return seaTunnelRow;
}

private SeaTunnelRow getSeaTunnelRowUpdateAfter() {
final SeaTunnelRow seaTunnelRow =
new SeaTunnelRow(
new Object[] {
1,
true,
(byte) 2,
(short) 2,
3,
4L,
4.3f,
5.3d,
BigDecimal.valueOf(6.3).setScale(1),
"NEW",
LocalDateTime.parse("2020-02-02T02:02:02")
});
seaTunnelRow.setRowKind(RowKind.UPDATE_AFTER);
return seaTunnelRow;
}

private SeaTunnelRow getSeaTunnelRowDelete() {
final SeaTunnelRow seaTunnelRow =
new SeaTunnelRow(
new Object[] {
2,
true,
(byte) 1,
(short) 2,
3,
4L,
4.3f,
5.3d,
BigDecimal.valueOf(6.3).setScale(1),
"NEW",
LocalDateTime.parse("2020-02-02T02:02:02")
});
seaTunnelRow.setRowKind(RowKind.DELETE);
return seaTunnelRow;
}

private CatalogTable getCatalogTable(Integer dbNum, String key) {
return CatalogTable.of(
TableIdentifier.of(CONNECTOR_IDENTITY, dbNum.toString(), key),
getTableSchema(),
new HashMap<>(),
new ArrayList<>(),
"");
}

private TableSchema getTableSchema() {
return new TableSchema(getColumns(), null, null);
}

private List<Column> getColumns() {
List<Column> columns = new ArrayList<>();
columns.add(new PhysicalColumn("id", BasicType.INT_TYPE, 32L, 0, true, "", ""));
columns.add(new PhysicalColumn("val_bool", BasicType.BOOLEAN_TYPE, 1L, 0, true, "", ""));
columns.add(new PhysicalColumn("val_int8", BasicType.BYTE_TYPE, 8L, 0, true, "", ""));
columns.add(new PhysicalColumn("val_int16", BasicType.SHORT_TYPE, 16L, 0, true, "", ""));
columns.add(new PhysicalColumn("val_int32", BasicType.INT_TYPE, 32L, 0, true, "", ""));
columns.add(new PhysicalColumn("val_int64", BasicType.LONG_TYPE, 64L, 0, true, "", ""));
columns.add(new PhysicalColumn("val_float", BasicType.FLOAT_TYPE, 32L, 0, true, "", ""));
columns.add(new PhysicalColumn("val_double", BasicType.DOUBLE_TYPE, 64L, 0, true, "", ""));
columns.add(
new PhysicalColumn("val_decimal", new DecimalType(16, 1), 16L, 1, true, "", ""));
columns.add(new PhysicalColumn("val_string", BasicType.STRING_TYPE, 0L, 0, true, "", ""));
columns.add(
new PhysicalColumn(
"val_unixtime_micros",
LocalTimeType.LOCAL_DATE_TIME_TYPE,
64L,
6,
true,
"",
""));
return columns;
}
}
87 changes: 0 additions & 87 deletions ...nnector-v2-e2e/connector-redis-e2e/src/test/resources/fake-to-redis-test-delete-hash.conf
View file
Open in desktop

This file was deleted.