This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new bc61ca5f4e [Improve][e2e] Improve the stability of redis e2e testing
e2e (#10989)
bc61ca5f4e is described below
commit bc61ca5f4e10409a20e4d960883282751b2f01c0
Author: zhiwei.niu <[email protected]>
AuthorDate: Tue Jun 2 10:41:14 2026 +0800
[Improve][e2e] Improve the stability of redis e2e testing e2e (#10989)
---
.../seatunnel/e2e/connector/redis/Redis5IT.java | 3 +
.../seatunnel/e2e/connector/redis/Redis7IT.java | 3 +
.../e2e/connector/redis/RedisClusterIT.java | 92 ++++++++++++++--------
.../e2e/connector/redis/RedisMasterAndSlaveIT.java | 2 +
.../connector/redis/RedisTestCaseTemplateIT.java | 35 ++++----
5 files changed, 88 insertions(+), 47 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis5IT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis5IT.java
index b20700577e..47e28a0dbb 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis5IT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis5IT.java
@@ -18,6 +18,9 @@ package org.apache.seatunnel.e2e.connector.redis;
import
org.apache.seatunnel.connectors.seatunnel.redis.config.RedisContainerInfo;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+@ResourceLock("redis-standalone-e2e")
public class Redis5IT extends RedisTestCaseTemplateIT {
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis7IT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis7IT.java
index e6fb8a7989..7e065837ff 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis7IT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/Redis7IT.java
@@ -18,6 +18,9 @@ package org.apache.seatunnel.e2e.connector.redis;
import
org.apache.seatunnel.connectors.seatunnel.redis.config.RedisContainerInfo;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+@ResourceLock("redis-standalone-e2e")
public class Redis7IT extends RedisTestCaseTemplateIT {
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisClusterIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisClusterIT.java
index dcf4da1844..6b326c6dc4 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisClusterIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisClusterIT.java
@@ -36,6 +36,7 @@ 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.junit.jupiter.api.parallel.ResourceLock;
import org.testcontainers.containers.Container;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
@@ -69,6 +70,7 @@ import java.util.Set;
import java.util.stream.Stream;
@Slf4j
+@ResourceLock("redis-cluster-e2e")
public class RedisClusterIT extends TestSuiteBase implements TestResource {
private static final int REDIS_CLUSTER_SIZE = 3;
@@ -94,14 +96,13 @@ public class RedisClusterIT extends TestSuiteBase
implements TestResource {
private void setupRedisContainer() {
redisClusterNodes = new GenericContainer[REDIS_CLUSTER_SIZE];
+ String hostIp = getDockerHostIp();
for (int i = 0; i < REDIS_CLUSTER_SIZE; i++) {
- String nodeName = "redis-cluster-" + (i + 1);
+ String nodeName = "redis-cluster-" + i;
int redisPort = REDIS_PORTS[i];
int busPort = REDIS_BUS_PORTS[i];
- // Get the host machine's IP address
- String hostIp = getHostIpAddress();
String redisCommand =
String.format(
"redis-server --cluster-enabled yes --port %d
--protected-mode no "
@@ -127,7 +128,6 @@ public class RedisClusterIT extends TestSuiteBase
implements TestResource {
new HostPortWaitStrategy()
.withStartupTimeout(Duration.ofMinutes(2)));
- // Set the fixed port mapping
redisClusterNodes[i].setPortBindings(
Arrays.asList(redisPort + ":" + redisPort, busPort + ":" +
busPort));
}
@@ -138,13 +138,17 @@ public class RedisClusterIT extends TestSuiteBase
implements TestResource {
private void createRedisCluster() {
try {
- String hostIp = getHostIpAddress();
StringBuilder clusterCreateCmd =
new StringBuilder(
"redis-cli --cluster create --cluster-replicas 0
--cluster-yes ");
- for (int port : REDIS_PORTS) {
-
clusterCreateCmd.append(hostIp).append(":").append(port).append(" ");
+ for (int i = 0; i < REDIS_CLUSTER_SIZE; i++) {
+ clusterCreateCmd
+ .append("redis-cluster-")
+ .append(i)
+ .append(":")
+ .append(REDIS_PORTS[i])
+ .append(" ");
}
clusterCreateCmd.append("-a
").append(redisContainerInfo.getPassword());
@@ -154,14 +158,11 @@ public class RedisClusterIT extends TestSuiteBase
implements TestResource {
Container.ExecResult result =
redisClusterNodes[0].execInContainer("sh", "-c",
clusterCreateCmd.toString());
- // Wait for the cluster to be created
- Thread.sleep(5000);
-
if (result.getExitCode() != 0) {
throw new RuntimeException("Failed to create Redis cluster: "
+ result.getStderr());
}
- log.info("Redis cluster created successfully");
+ log.info("Redis cluster created, waiting for slot assignment via
CLUSTER INFO...");
} catch (Exception e) {
throw new RuntimeException("Error creating Redis cluster", e);
}
@@ -170,7 +171,7 @@ public class RedisClusterIT extends TestSuiteBase
implements TestResource {
private void waitForRedisClusterReady() {
log.info("Waiting for Redis cluster to be ready...");
- int maxRetries = 10;
+ int maxRetries = 30;
int retryCount = 0;
while (retryCount < maxRetries) {
@@ -185,29 +186,34 @@ public class RedisClusterIT extends TestSuiteBase
implements TestResource {
String.valueOf(REDIS_PORTS[i]),
"-a",
redisContainerInfo.getPassword(),
- "ping");
+ "cluster",
+ "info");
- if (!"PONG".equals(result.getStdout().trim())) {
+ String output = result.getStdout().trim();
+ if (!output.contains("cluster_state:ok")
+ || !output.contains("cluster_slots_ok:16384")) {
allReady = false;
break;
}
}
if (allReady) {
- log.info("All Redis nodes are ready after {} attempts",
retryCount + 1);
+ log.info(
+ "Redis cluster is fully ready after {} attempts
(all slots assigned)",
+ retryCount + 1);
return;
}
} catch (Exception e) {
log.debug(
- "Redis readiness check failed, attempt {}: {}",
+ "Redis cluster readiness check failed, attempt {}: {}",
retryCount + 1,
e.getMessage());
}
retryCount++;
try {
- Thread.sleep(3000);
+ Thread.sleep(2000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
@@ -219,7 +225,7 @@ public class RedisClusterIT extends TestSuiteBase
implements TestResource {
private void initJedisCluster() {
Set<HostAndPort> jedisClusterNodes = new HashSet<>();
- String hostIp = getHostIpAddress();
+ String hostIp = getDockerHostIp();
for (int port : REDIS_PORTS) {
jedisClusterNodes.add(new HostAndPort(hostIp, port));
}
@@ -318,7 +324,7 @@ public class RedisClusterIT extends TestSuiteBase
implements TestResource {
Assertions.assertEquals(100, amount);
} finally {
jedisCluster.del("key_set");
- Assertions.assertEquals(0, jedisCluster.llen("key_set"));
+ Assertions.assertFalse(jedisCluster.exists("key_set"));
}
}
@@ -513,24 +519,44 @@ public class RedisClusterIT extends TestSuiteBase
implements TestResource {
return Pair.of(rowType, rows);
}
- private String getHostIpAddress() {
- String ip = "";
+ private String getDockerHostIp() {
+ String fallback = null;
try {
- Enumeration<NetworkInterface> networkInterfaces =
- NetworkInterface.getNetworkInterfaces();
- while (networkInterfaces.hasMoreElements()) {
- NetworkInterface networkInterface =
networkInterfaces.nextElement();
- Enumeration<InetAddress> inetAddresses =
networkInterface.getInetAddresses();
- while (inetAddresses.hasMoreElements()) {
- InetAddress inetAddress = inetAddresses.nextElement();
- if (!inetAddress.isLoopbackAddress() && inetAddress
instanceof Inet4Address) {
- ip = inetAddress.getHostAddress();
+ Enumeration<NetworkInterface> interfaces =
NetworkInterface.getNetworkInterfaces();
+ while (interfaces.hasMoreElements()) {
+ NetworkInterface ni = interfaces.nextElement();
+ if (!ni.isUp() || ni.isLoopback() || ni.isVirtual()) {
+ continue;
+ }
+
+ String name = ni.getName();
+ if (name.startsWith("utun")
+ || name.startsWith("tun")
+ || name.startsWith("tap")
+ || name.startsWith("ppp")
+ || name.startsWith("docker")
+ || name.startsWith("br-")
+ || name.startsWith("veth")
+ || name.startsWith("vmnet")
+ || name.startsWith("virbr")) {
+ continue;
+ }
+
+ Enumeration<InetAddress> addrs = ni.getInetAddresses();
+ while (addrs.hasMoreElements()) {
+ InetAddress addr = addrs.nextElement();
+ if (addr instanceof Inet4Address &&
!addr.isLoopbackAddress()) {
+ String ip = addr.getHostAddress();
+ if ("en0".equals(name) || "eth0".equals(name) ||
"wlan0".equals(name)) {
+ return ip;
+ }
+ fallback = ip;
}
}
}
- } catch (SocketException ex) {
- ex.printStackTrace();
+ } catch (SocketException e) {
+ log.warn("Failed to enumerate network interfaces", e);
}
- return ip;
+ return fallback != null ? fallback : "127.0.0.1";
}
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisMasterAndSlaveIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisMasterAndSlaveIT.java
index 48ba3163f9..15a0ad460f 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisMasterAndSlaveIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisMasterAndSlaveIT.java
@@ -25,6 +25,7 @@ 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.junit.jupiter.api.parallel.ResourceLock;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
@@ -40,6 +41,7 @@ import java.util.Objects;
import java.util.stream.Stream;
@Slf4j
+@ResourceLock("redis-master-slave-e2e")
public class RedisMasterAndSlaveIT extends TestSuiteBase implements
TestResource {
private static RedisContainerInfo masterContainerInfo;
private static RedisContainerInfo slaveContainerInfo;
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java
index f16c8d526e..bb1650f93d 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/java/org/apache/seatunnel/e2e/connector/redis/RedisTestCaseTemplateIT.java
@@ -305,9 +305,9 @@ public abstract class RedisTestCaseTemplateIT extends
TestSuiteBase implements T
Container.ExecResult execResult =
container.executeJob("/redis-to-redis-expire.conf");
Assertions.assertEquals(0, execResult.getExitCode());
Assertions.assertEquals(100, jedis.llen("key_list"));
- // Clear data to prevent data duplication in the next TestContainer
- Thread.sleep(60 * 1000);
- Assertions.assertEquals(0, jedis.llen("key_list"));
+ await().atMost(90, TimeUnit.SECONDS)
+ .pollInterval(3, TimeUnit.SECONDS)
+ .untilAsserted(() -> Assertions.assertEquals(0,
jedis.llen("key_list")));
}
@TestTemplate
@@ -689,19 +689,26 @@ public abstract class RedisTestCaseTemplateIT extends
TestSuiteBase implements T
@DisabledOnOs(OS.WINDOWS)
public void testFakeToRedisInRealTimeTest(TestContainer container)
throws IOException, InterruptedException {
- CompletableFuture.supplyAsync(
- () -> {
- try {
-
container.executeJob("/fake-to-redis-test-in-real-time.conf");
- } catch (Exception e) {
- log.error("Commit task exception :" + e.getMessage());
- throw new RuntimeException(e);
- }
- return null;
- });
- await().atMost(60000, TimeUnit.MILLISECONDS)
+ CompletableFuture<Void> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ container.executeJob(
+
"/fake-to-redis-test-in-real-time.conf");
+ } catch (Exception e) {
+ log.error("Streaming job execution
failed", e);
+ throw new RuntimeException(e);
+ }
+ return null;
+ })
+ .thenAccept(v -> {});
+ await().atMost(120, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
.untilAsserted(
() -> {
+ Assertions.assertFalse(
+ jobFuture.isCompletedExceptionally(),
+ "Streaming job failed unexpectedly");
Assertions.assertEquals(3,
jedis.llen("list_check"));
});
jedis.del("list_check");
