This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 42bafc590c [Test][Zeta] Poll for worker registration and system-load
collection in allocate strategy ITs (#13066)
42bafc590c is described below
commit 42bafc590cf2d9ed8c787e23143c58285d7953ab
Author: Jast <[email protected]>
AuthorDate: Fri Oct 30 04:50:40 2026 +0000
[Test][Zeta] Poll for worker registration and system-load collection in
allocate strategy ITs (#13066)
---
.../SlotRatioAllocateStrategyIT.java | 35 +++++++++----
.../SystemLoadAllocateStrategyIT.java | 57 +++++++++++++++++-----
.../allocation/strategy/SystemLoadStrategy.java | 10 ++++
3 files changed, 81 insertions(+), 21 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/allocatestrategy/SlotRatioAllocateStrategyIT.java
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/allocatestrategy/SlotRatioAllocateStrategyIT.java
index c870cfbd6d..51ed027431 100644
---
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/allocatestrategy/SlotRatioAllocateStrategyIT.java
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/allocatestrategy/SlotRatioAllocateStrategyIT.java
@@ -110,8 +110,31 @@ public class SlotRatioAllocateStrategyIT {
Assertions.assertEquals(
2,
finalNode.getCluster().getMembers().size()));
- // Waiting for worker heartbeat registration
- Thread.sleep(10000);
+ // The resource manager only knows about a worker after its first
heartbeat, and
+ // job submission needs both workers to be registered, so poll for
registration
+ // instead of waiting a fixed time.
+ NodeEngineImpl nodeEngine = node1.node.nodeEngine;
+ Address node1Address = node1.node.address;
+ Address node2Address = node2.node.address;
+ SeaTunnelServer server =
nodeEngine.getService(SeaTunnelServer.SERVICE_NAME);
+ ResourceManager resourceManager =
server.getCoordinatorService().getResourceManager();
+
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(
+ resourceManager
+ .getRegisterWorker()
+ .containsKey(node1Address),
+ "Node1 has not registered yet");
+ Assertions.assertTrue(
+ resourceManager
+ .getRegisterWorker()
+ .containsKey(node2Address),
+ "Node2 has not registered yet");
+ });
+
Common.setDeployMode(DeployMode.CLIENT);
JobConfig jobConfig = new JobConfig();
jobConfig.setName(testCaseName);
@@ -133,14 +156,6 @@ public class SlotRatioAllocateStrategyIT {
seaTunnelConfig)
.execute();
- NodeEngineImpl nodeEngine = node1.node.nodeEngine;
- Address node2Address = node2.node.address;
- Address node1Address = node1.node.address;
-
- // Get the number of occupied slots through resourceManager
- SeaTunnelServer server =
nodeEngine.getService(SeaTunnelServer.SERVICE_NAME);
- ResourceManager resourceManager =
server.getCoordinatorService().getResourceManager();
-
// SLOT_RATION strategy, the task will eventually occupy 5 slots
and will be distributed
// to two nodes, one node occupies 2 slots and the other occupies
3 slots.
Awaitility.await()
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/allocatestrategy/SystemLoadAllocateStrategyIT.java
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/allocatestrategy/SystemLoadAllocateStrategyIT.java
index 5380442e83..fe7ccafb71 100644
---
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/allocatestrategy/SystemLoadAllocateStrategyIT.java
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/allocatestrategy/SystemLoadAllocateStrategyIT.java
@@ -30,7 +30,9 @@ import
org.apache.seatunnel.engine.common.config.server.SlotServiceConfig;
import org.apache.seatunnel.engine.e2e.TestUtils;
import org.apache.seatunnel.engine.server.SeaTunnelServer;
import org.apache.seatunnel.engine.server.SeaTunnelServerStarter;
+import
org.apache.seatunnel.engine.server.resourcemanager.AbstractResourceManager;
import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager;
+import
org.apache.seatunnel.engine.server.resourcemanager.allocation.strategy.SystemLoadStrategy;
import org.apache.seatunnel.engine.server.resourcemanager.worker.WorkerProfile;
import org.awaitility.Awaitility;
@@ -117,8 +119,31 @@ public class SystemLoadAllocateStrategyIT {
Assertions.assertEquals(
2,
finalNode.getCluster().getMembers().size()));
- // Waiting for worker heartbeat registration
- Thread.sleep(10000);
+ // The resource manager only knows about a worker after its first
heartbeat, and
+ // job submission needs both workers to be registered, so poll for
registration
+ // instead of waiting a fixed time.
+ NodeEngineImpl nodeEngine = node1.node.nodeEngine;
+ Address node1Address = node1.node.address;
+ Address node2Address = node2.node.address;
+ SeaTunnelServer server =
nodeEngine.getService(SeaTunnelServer.SERVICE_NAME);
+ ResourceManager resourceManager =
server.getCoordinatorService().getResourceManager();
+
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(
+ resourceManager
+ .getRegisterWorker()
+ .containsKey(node1Address),
+ "Node1 has not registered yet");
+ Assertions.assertTrue(
+ resourceManager
+ .getRegisterWorker()
+ .containsKey(node2Address),
+ "Node2 has not registered yet");
+ });
+
Common.setDeployMode(DeployMode.CLIENT);
JobConfig jobConfig = new JobConfig();
jobConfig.setName(testCaseName);
@@ -152,13 +177,6 @@ public class SystemLoadAllocateStrategyIT {
seaTunnelConfig)
.execute();
- NodeEngineImpl nodeEngine = node1.node.nodeEngine;
- Address node2Address = node2.node.address;
- Address node1Address = node1.node.address;
-
- SeaTunnelServer server =
nodeEngine.getService(SeaTunnelServer.SERVICE_NAME);
- ResourceManager resourceManager =
server.getCoordinatorService().getResourceManager();
-
Awaitility.await()
.atMost(600, TimeUnit.SECONDS)
.untilAsserted(
@@ -176,8 +194,25 @@ public class SystemLoadAllocateStrategyIT {
});
log.info("The first step is completed");
- // Waiting to collect the node's System Load information
- Thread.sleep(60000);
+ // The SYSTEM_LOAD strategy weights a worker without load samples
with the lowest
+ // priority, so submitting the second job before both workers have
reported their
+ // system load can assign all slots to a single node. Poll until
both workers have
+ // reported load samples instead of waiting a fixed time (a sample
is sent every
+ // SYSTEM_LOAD_SEND_INTERVAL heartbeats).
+ SystemLoadStrategy systemLoadStrategy =
+ (SystemLoadStrategy)
+ ((AbstractResourceManager)
resourceManager).getSlotAllocationStrategy();
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(
+
systemLoadStrategy.hasLoadInfo(node1Address),
+ "Node1 has not reported system load
yet");
+ Assertions.assertTrue(
+
systemLoadStrategy.hasLoadInfo(node2Address),
+ "Node2 has not reported system load
yet");
+ });
// Start a task that occupies 4 slots
jobConfig = new JobConfig();
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/allocation/strategy/SystemLoadStrategy.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/allocation/strategy/SystemLoadStrategy.java
index c36f36d048..8bbd80616a 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/allocation/strategy/SystemLoadStrategy.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/allocation/strategy/SystemLoadStrategy.java
@@ -52,6 +52,16 @@ public class SystemLoadStrategy implements
SlotAllocationStrategy {
workerLoadMap.computeIfAbsent(address, k ->
EvictingQueue.create(5)).add(systemLoadInfo);
}
+ /**
+ * Returns whether at least one load sample has been collected for the
given worker. A worker
+ * without load samples gets the lowest weight in {@link #selectWorker},
so callers that need
+ * load-based allocation decisions can wait on this signal.
+ */
+ public boolean hasLoadInfo(Address address) {
+ EvictingQueue<SystemLoadInfo> loadInfo = workerLoadMap.get(address);
+ return loadInfo != null && !loadInfo.isEmpty();
+ }
+
@Override
public Optional<WorkerProfile> selectWorker(List<WorkerProfile>
availableWorkers) {
Optional<WorkerProfile> workerProfile =
