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-13066-794a025397be8e7fee672af46a5718d7c1382aa5 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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 =
