This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12652-704b3fec9de38dcb95732616ab258d76ada89f1f in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 5008be82623ef599c24c541638904ffd461e5245 Author: Jast <[email protected]> AuthorDate: Thu Oct 8 13:23:35 2026 +0000 [Test][Zeta] Replace fixed waits with condition-based waits in engine e2e tests (#12652) Co-authored-by: root <[email protected]> --- .../seatunnel/engine/e2e/CheckpointEnableIT.java | 21 ++++++++++--- .../seatunnel/engine/e2e/SeaTunnelSlotIT.java | 36 +++++++++------------- .../engine/e2e/classloader/ClassLoaderITBase.java | 2 +- .../seatunnel/engine/e2e/joblog/JobLogIT.java | 9 ++++-- .../resourceIsolation/WorkerTagClusterTest.java | 36 ++++++++++------------ 5 files changed, 56 insertions(+), 48 deletions(-) diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointEnableIT.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointEnableIT.java index 39d5f136d0..2e4762da30 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointEnableIT.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointEnableIT.java @@ -110,9 +110,15 @@ public class CheckpointEnableIT extends TestSuiteBase { } }); - // wait obtain job id - Thread.sleep(15000); - Assertions.assertTrue(container.getServerLogs().contains("checkpoint is enabled")); + // wait obtain job id; poll for the log line instead of a fixed sleep, since the + // job may take longer to reach RUNNING under CI load + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertTrue( + container + .getServerLogs() + .contains("checkpoint is enabled"))); Assertions.assertEquals(0, container.savepointJob(String.valueOf(jobId)).getExitCode()); Assertions.assertEquals(0, startFuture.get().getExitCode()); // restore job @@ -164,8 +170,13 @@ public class CheckpointEnableIT extends TestSuiteBase { } }); - Thread.sleep(15000); - Assertions.assertTrue(container.getServerLogs().contains("checkpoint is enabled")); + await().atMost(60000, TimeUnit.MILLISECONDS) + .untilAsserted( + () -> + Assertions.assertTrue( + container + .getServerLogs() + .contains("checkpoint is enabled"))); Assertions.assertEquals(0, container.savepointJob(String.valueOf(jobId)).getExitCode()); // restore job diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SeaTunnelSlotIT.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SeaTunnelSlotIT.java index c6000d4d9c..4186b54568 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SeaTunnelSlotIT.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SeaTunnelSlotIT.java @@ -73,13 +73,11 @@ public class SeaTunnelSlotIT { Awaitility.await() .atMost(300000, TimeUnit.MILLISECONDS) .untilAsserted( - () -> { - Thread.sleep(2000); - Assertions.assertTrue( - objectCompletableFuture.isDone() - && JobStatus.FAILED.equals( - objectCompletableFuture.get())); - }); + () -> + Assertions.assertTrue( + objectCompletableFuture.isDone() + && JobStatus.FAILED.equals( + objectCompletableFuture.get()))); } finally { if (engineClient != null) { @@ -126,13 +124,11 @@ public class SeaTunnelSlotIT { Awaitility.await() .atMost(300000, TimeUnit.MILLISECONDS) .untilAsserted( - () -> { - Thread.sleep(2000); - Assertions.assertTrue( - objectCompletableFuture.isDone() - && JobStatus.FINISHED.equals( - objectCompletableFuture.get())); - }); + () -> + Assertions.assertTrue( + objectCompletableFuture.isDone() + && JobStatus.FINISHED.equals( + objectCompletableFuture.get()))); } finally { if (engineClient != null) { @@ -205,13 +201,11 @@ public class SeaTunnelSlotIT { Awaitility.await() .atMost(300000, TimeUnit.MILLISECONDS) .untilAsserted( - () -> { - Thread.sleep(2000); - Assertions.assertTrue( - oversizedJobComplete.isDone() - && JobStatus.FAILED.equals( - oversizedJobComplete.get())); - }); + () -> + Assertions.assertTrue( + oversizedJobComplete.isDone() + && JobStatus.FAILED.equals( + oversizedJobComplete.get()))); // The oversized job's own resource shortfall must be fully resolved on the master // before this proves anything: submit a second, minimal job that needs far fewer diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java index f0ff5d78bd..1d155dad08 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java @@ -87,6 +87,7 @@ public abstract class ClassLoaderITBase extends SeaTunnelEngineContainer { SEATUNNEL_HOME); Awaitility.await() .atMost(2, TimeUnit.MINUTES) + .pollInterval(1, TimeUnit.SECONDS) .untilAsserted( () -> { Response response = @@ -97,7 +98,6 @@ public abstract class ClassLoaderITBase extends SeaTunnelEngineContainer { + server.getFirstMappedPort() + "/hazelcast/rest/cluster"); response.then().statusCode(200); - Thread.sleep(10000); Assertions.assertEquals( 1, response.jsonPath().getList("members").size()); }); diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/joblog/JobLogIT.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/joblog/JobLogIT.java index 8763bc5d43..d12e4681ac 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/joblog/JobLogIT.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/joblog/JobLogIT.java @@ -161,13 +161,18 @@ public class JobLogIT extends SeaTunnelEngineContainer { Tuple2.tuple2(false, "job-" + CUSTOM_JOB_ID2 + ".log"), Tuple2.tuple2(false, "job-" + CUSTOM_JOB_ID3 + ".log")); assertFileLogClean(before); - Thread.sleep(90000); + // The retained log file of the finished batch job appears asynchronously; poll + // for it instead of a fixed sleep that fails when the retention task is slower + // than expected under CI load. List<Tuple2<Boolean, String>> after = Lists.newArrayList( Tuple2.tuple2(true, "job-" + CUSTOM_JOB_ID + ".log"), Tuple2.tuple2(false, "job-" + CUSTOM_JOB_ID2 + ".log"), Tuple2.tuple2(false, "job-" + CUSTOM_JOB_ID3 + ".log")); - assertFileLogClean(after); + Awaitility.await() + .atMost(3, TimeUnit.MINUTES) + .pollInterval(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertFileLogClean(after)); } private void assertConsoleLog() { diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/resourceIsolation/WorkerTagClusterTest.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/resourceIsolation/WorkerTagClusterTest.java index d173cb7a57..58ada293eb 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/resourceIsolation/WorkerTagClusterTest.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/resourceIsolation/WorkerTagClusterTest.java @@ -25,7 +25,6 @@ import org.apache.seatunnel.engine.server.SeaTunnelServerStarter; import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager; import org.awaitility.Awaitility; -import org.awaitility.core.ThrowingRunnable; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; @@ -93,26 +92,25 @@ public class WorkerTagClusterTest { public void testTagFilter(Map<String, String> tagFilter, int expectedWorkerCount) throws Exception { - // waiting all node added to cluster + // waiting all node added to cluster; poll without the in-poll sleep so each + // attempt checks the condition immediately, and give cluster formation and + // worker registration enough headroom under CI load Awaitility.await() - .atMost(10000, TimeUnit.MILLISECONDS) + .atMost(60000, TimeUnit.MILLISECONDS) + .pollInterval(1, TimeUnit.SECONDS) .untilAsserted( - new ThrowingRunnable() { - @Override - public void run() throws Throwable { - Thread.sleep(2000); - // check master and worker node - Assertions.assertEquals( - 2, masterNode1.getCluster().getMembers().size()); - NodeEngineImpl nodeEngine = masterNode1.node.nodeEngine; - SeaTunnelServer server = - nodeEngine.getService(SeaTunnelServer.SERVICE_NAME); - ResourceManager resourceManager = - server.getCoordinatorService().getResourceManager(); - // if tag matched, then worker count is 1 else 0 - int workerCount = resourceManager.workerCount(tagFilter); - Assertions.assertEquals(expectedWorkerCount, workerCount); - } + () -> { + // check master and worker node + Assertions.assertEquals( + 2, masterNode1.getCluster().getMembers().size()); + NodeEngineImpl nodeEngine = masterNode1.node.nodeEngine; + SeaTunnelServer server = + nodeEngine.getService(SeaTunnelServer.SERVICE_NAME); + ResourceManager resourceManager = + server.getCoordinatorService().getResourceManager(); + // if tag matched, then worker count is 1 else 0 + int workerCount = resourceManager.workerCount(tagFilter); + Assertions.assertEquals(expectedWorkerCount, workerCount); }); }
