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

Merged
merged 6 commits into from
Nov 3, 2025
Merged

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

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 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
e951649
[bugfix] fix some code
chl-wxp Oct 27, 2025
8284905
[bugfix] delete some code
chl-wxp Oct 27, 2025
7904ebc
[feature] delete some code
chl-wxp Nov 3, 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 {
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);
}

@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.