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 4fc258f509 [Fix][Zeta] Stabilize flaky Java 8 unit tests (#11542)
4fc258f509 is described below
commit 4fc258f50969e14e217cdb977df99259631330bc
Author: zhiwei.niu <[email protected]>
AuthorDate: Fri Jul 24 18:31:07 2026 +0800
[Fix][Zeta] Stabilize flaky Java 8 unit tests (#11542)
---
.../client/SeaTunnelEngineClusterRoleTest.java | 4 +-
.../engine/server/ConnectorPackageServiceTest.java | 74 +++++++++++++---------
.../engine/server/master/JobMasterTest.java | 12 +++-
3 files changed, 57 insertions(+), 33 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelEngineClusterRoleTest.java
b/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelEngineClusterRoleTest.java
index 0b12e644d5..49b3754fd3 100644
---
a/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelEngineClusterRoleTest.java
+++
b/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelEngineClusterRoleTest.java
@@ -480,7 +480,9 @@ public class SeaTunnelEngineClusterRoleTest {
&& jobClient
.listJobStatus(true)
.contains("RUNNING")));
- jobClient.cancelJob(jobId);
+ // This test verifies job queries after a master handoff, not
graceful cancellation.
+ // Force cancellation avoids waiting for checkpoint teardown on
the promoted master.
+ jobClient.cancelJob(jobId, true);
await().atMost(120000, TimeUnit.MILLISECONDS)
.pollDelay(5, TimeUnit.SECONDS)
.pollInterval(2, TimeUnit.SECONDS)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/ConnectorPackageServiceTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/ConnectorPackageServiceTest.java
index 9eda75f6d9..73d3eb6115 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/ConnectorPackageServiceTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/ConnectorPackageServiceTest.java
@@ -61,6 +61,7 @@ import com.hazelcast.internal.serialization.Data;
import com.hazelcast.logging.ILogger;
import com.hazelcast.logging.Logger;
import com.hazelcast.spi.impl.NodeEngineImpl;
+import com.hazelcast.spi.properties.ClusterProperty;
import lombok.extern.slf4j.Slf4j;
import java.io.ByteArrayOutputStream;
@@ -119,6 +120,13 @@ public class ConnectorPackageServiceTest {
+ " connector-jar-expiry-time: 600";
SEATUNNEL_CONFIG =
ConfigProvider.locateAndGetSeaTunnelConfigFromString(yaml);
+ // Keep the failover test fast when a listener registration briefly
targets the old master.
+ SEATUNNEL_CONFIG
+ .getHazelcastConfig()
+
.setProperty(ClusterProperty.INVOCATION_MAX_RETRY_COUNT.getName(), "3");
+ SEATUNNEL_CONFIG
+ .getHazelcastConfig()
+ .setProperty(ClusterProperty.INVOCATION_RETRY_PAUSE.getName(),
"1");
}
@Test
@@ -128,38 +136,44 @@ public class ConnectorPackageServiceTest {
.setClusterName(
TestUtils.getClusterName(
"ConnectorPackageServiceTest_testMasterNodeActive"));
- HazelcastInstanceImpl instance1 =
-
SeaTunnelServerStarter.createHazelcastInstance(SEATUNNEL_CONFIG);
- HazelcastInstanceImpl instance2 =
-
SeaTunnelServerStarter.createHazelcastInstance(SEATUNNEL_CONFIG);
-
- SeaTunnelServer server1 =
-
instance1.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
- SeaTunnelServer server2 =
-
instance2.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
-
- Assertions.assertTrue(server1.isMasterNode());
- Assertions.assertTrue(server1.getConnectorPackageService() != null);
-
+ HazelcastInstanceImpl instance1 = null;
+ HazelcastInstanceImpl instance2 = null;
try {
- server2.getConnectorPackageService();
- } catch (Exception e) {
- Assertions.assertTrue(e instanceof SeaTunnelEngineException);
+ instance1 =
SeaTunnelServerStarter.createHazelcastInstance(SEATUNNEL_CONFIG);
+ instance2 =
SeaTunnelServerStarter.createHazelcastInstance(SEATUNNEL_CONFIG);
+
+ SeaTunnelServer server1 =
+
instance1.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
+ SeaTunnelServer server2 =
+
instance2.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
+
+ Assertions.assertTrue(server1.isMasterNode());
+ Assertions.assertNotNull(server1.getConnectorPackageService());
+ Assertions.assertThrows(
+ SeaTunnelEngineException.class,
server2::getConnectorPackageService);
+
+ // shutdown instance1
+ instance1.shutdown();
+ await().atMost(20000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+ try {
+
Assertions.assertTrue(server2.isMasterNode());
+
Assertions.assertNotNull(server2.getConnectorPackageService());
+ } catch (SeaTunnelEngineException e) {
+ Assertions.fail(
+ "Connector package service is not
ready on the new master",
+ e);
+ }
+ });
+ } finally {
+ if (instance2 != null && instance2.node.isRunning()) {
+ instance2.shutdown();
+ }
+ if (instance1 != null && instance1.node.isRunning()) {
+ instance1.shutdown();
+ }
}
-
- // shutdown instance1
- instance1.shutdown();
- await().atMost(20000, TimeUnit.MILLISECONDS)
- .untilAsserted(
- () -> {
- try {
- Assertions.assertTrue(server2.isMasterNode());
-
Assertions.assertTrue(server2.getConnectorPackageService() != null);
- } catch (SeaTunnelEngineException e) {
- Assertions.assertTrue(false);
- }
- });
- instance2.shutdown();
}
@Test
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java
index a394b9980b..fea61ab1c9 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java
@@ -53,6 +53,7 @@ import org.junit.jupiter.api.condition.OS;
import com.hazelcast.internal.serialization.Data;
import com.hazelcast.map.IMap;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -320,9 +321,10 @@ public class JobMasterTest extends
AbstractSeaTunnelServerTest {
private void assertCloseIdleTask(JobMaster jobMaster) {
SlotService slotService = server.getSlotService();
+ long jobId = jobMaster.getJobId();
// Savepoint restore can overlap old slot release with new task
scheduling for a short time.
await().atMost(60, TimeUnit.SECONDS)
- .until(() ->
slotService.getWorkerProfile().getAssignedSlots().length == 4);
+ .until(() -> getAssignedSlotCount(slotService, jobId) == 4);
Assertions.assertEquals(1,
jobMaster.getPhysicalPlan().getPipelineList().size());
SubPlan subPlan = jobMaster.getPhysicalPlan().getPipelineList().get(0);
@@ -357,7 +359,13 @@ public class JobMasterTest extends
AbstractSeaTunnelServerTest {
checkpointCoordinator.getClosedIdleTask().size()
>= expectedClosedIdleTaskSize);
await().atMost(60, TimeUnit.SECONDS)
- .until(() ->
slotService.getWorkerProfile().getAssignedSlots().length == 3);
+ .until(() -> getAssignedSlotCount(slotService, jobId) == 3);
+ }
+
+ private long getAssignedSlotCount(SlotService slotService, long jobId) {
+ return Arrays.stream(slotService.getWorkerProfile().getAssignedSlots())
+ .filter(slotProfile -> slotProfile.getOwnerJobID() == jobId)
+ .count();
}
private JobMaster newJobInstanceWithRunningState(long jobId) throws
InterruptedException {
