[Fix][Zeta] Fix hybrid deployment can not get worker when init (#7235)

apache · Jul 19, 2024 · d39ad93 · d39ad93
1 parent 2058f5d
commit d39ad93
Show file tree

Hide file tree

Showing 5 changed files with 39 additions and 15 deletions.
diff --git a/...l-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelClientTest.java b/...l-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelClientTest.java
@@ -392,11 +392,14 @@ public void testSetJobIdDuplicate() {
                     Assertions.assertThrows(
                             Exception.class,
                             () -> jobExecutionEnvWithSameJobId.execute().waitForJobCompleteV2());
-            Assertions.assertEquals(
-                    String.format(
-                            "The job id %s has already been submitted and is not starting with a savepoint.",
-                            jobId),
-                    exception.getCause().getMessage());
+            Assertions.assertTrue(
+                    exception
+                            .getCause()
+                            .getMessage()
+                            .contains(
+                                    String.format(
+                                            "The job id %s has already been submitted and is not starting with a savepoint.",
+                                            jobId)));
         } catch (Exception e) {
             throw new RuntimeException(e);
         } finally {

diff --git a/...unnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java b/...unnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/SeaTunnelServer.java
@@ -84,6 +84,12 @@ public SeaTunnelServer(@NonNull SeaTunnelConfig seaTunnelConfig) {
 
     /** Lazy load for Slot Service */
     public SlotService getSlotService() {
+        // If the node is master node, the slot service is not needed.
+        if (EngineConfig.ClusterRole.MASTER.ordinal()
+                == seaTunnelConfig.getEngineConfig().getClusterRole().ordinal()) {
+            return null;
+        }
+
         if (slotService == null) {
             synchronized (this) {
                 if (slotService == null) {

diff --git a/...main/java/org/apache/seatunnel/engine/server/resourcemanager/AbstractResourceManager.java b/...main/java/org/apache/seatunnel/engine/server/resourcemanager/AbstractResourceManager.java
@@ -74,25 +74,31 @@ public void init() {
 
     private void initWorker() {
         log.info("initWorker... ");
-        List<Address> aliveWorker =
+        List<Address> aliveNode =
                 nodeEngine.getClusterService().getMembers().stream()
-                        .filter(Member::isLiteMember)
                         .map(Member::getAddress)
                         .collect(Collectors.toList());
-        log.info("initWorker live nodes: " + aliveWorker);
+        log.info("init live nodes: {}", aliveNode);
         List<CompletableFuture<Void>> futures =
-                aliveWorker.stream()
+                aliveNode.stream()
                         .map(
-                                worker ->
-                                        sendToMember(new SyncWorkerProfileOperation(), worker)
+                                node ->
+                                        sendToMember(new SyncWorkerProfileOperation(), node)
                                                 .thenAccept(
                                                         p -> {
-                                                            registerWorker.put(
-                                                                    worker, (WorkerProfile) p);
+                                                            if (p != null) {
+                                                                registerWorker.put(
+                                                                        node, (WorkerProfile) p);
+                                                                log.info(
+                                                                        "received new worker register: "
+                                                                                + ((WorkerProfile)
+                                                                                                p)
+                                                                                        .getAddress());
+                                                            }
                                                         }))
                         .collect(Collectors.toList());
         futures.forEach(CompletableFuture::join);
-        log.info("registerWorker: " + registerWorker);
+        log.info("registerWorker: {}", registerWorker);
     }
 
     @Override

diff --git a/.../apache/seatunnel/engine/server/resourcemanager/opeartion/SyncWorkerProfileOperation.java b/.../apache/seatunnel/engine/server/resourcemanager/opeartion/SyncWorkerProfileOperation.java
@@ -33,7 +33,11 @@ public class SyncWorkerProfileOperation extends Operation implements IdentifiedD
     @Override
     public void run() throws Exception {
         SeaTunnelServer server = getService();
-        result = server.getSlotService().getWorkerProfile();
+        if (server.getSlotService() != null) {
+            result = server.getSlotService().getWorkerProfile();
+        } else {
+            result = null;
+        }
     }
 
     @Override

diff --git a/...src/test/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManagerTest.java b/...src/test/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManagerTest.java
@@ -54,6 +54,11 @@ public void before() {
         server.getSlotService();
     }
 
+    @Test
+    public void testHaveWorkerWhenUseHybridDeployment() {
+        Assertions.assertEquals(1, resourceManager.workerCount(null));
+    }
+
     @Test
     public void testApplyRequest() throws ExecutionException, InterruptedException {
         List<ResourceProfile> resourceProfiles = new ArrayList<>();