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 all 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
8 changes: 8 additions & 0 deletions seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,5 +33,13 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>connector-common</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
</dependencies>
</project>
199 changes: 191 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.sink.RedisSinkFactory;
import org.apache.seatunnel.connectors.seatunnel.sink.SinkFlowTestUtils;
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 Down Expand Up @@ -72,6 +83,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 Down Expand Up @@ -104,6 +116,7 @@ public void startUp() {
.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 +623,22 @@ 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";
SinkFlowTestUtils.runBatchWithCheckpointDisabled(
getCatalogTable(0, key),
getReadonlyConfig(RedisDataType.HASH, key),
new RedisSinkFactory(),
Arrays.asList(
getSeaTunnelRowInsert1(),
getSeaTunnelRowInsert2(),
getSeaTunnelRowInsert3(),
getSeaTunnelRowUpdateBefore(),
getSeaTunnelRowUpdateAfter(),
getSeaTunnelRowDelete()));
Assertions.assertEquals(2, jedis.hlen(key));
jedis.del(key);
}

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

public abstract RedisContainerInfo getRedisContainerInfo();

private ReadonlyConfig getReadonlyConfig(RedisDataType dataType, String key) {
Map<String, Object> map = new HashMap<>();
map.put("host", redisContainer.getHost());
map.put("port", redisContainer.getFirstMappedPort());
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.