Merge pull request #22 from TongchengOpenSource/dynamic_workerprofile

[Improve][Zeta] Improve the Success Rate of Resource Allocation for Large Tasks in Dynamic Slot Mode.
TongchengOpenSource · Dec 9, 2024 · ea0a400 · ea0a400
2 parents 150d30d + 6973abd
commit ea0a400
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 1 deletion.
diff --git a/...-e2e/connector-hive-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hive/HiveIT.java b/...-e2e/connector-hive-e2e/src/test/java/org/apache/seatunnel/e2e/connector/hive/HiveIT.java
@@ -228,6 +228,8 @@ private void executeJob(TestContainer container, String job1, String job2)
     }
 
     @TestTemplate
+    @Disabled(
+            "[HDFS/COS/OSS/S3] is not available in CI, if you want to run this test, please set up your own environment in the test case file, hadoop_hive_conf_path_local and ip below}")
     public void testFakeSinkHiveOnHDFS(TestContainer container) throws Exception {
         executeJob(container, "/fake_to_hive_on_hdfs.conf", "/hive_on_hdfs_to_assert.conf");
     }

diff --git a/.../main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceRequestHandler.java b/.../main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceRequestHandler.java
@@ -33,6 +33,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -230,7 +231,17 @@ public Optional<WorkerProfile> preCheckWorkerResource(ResourceProfile r) {
                     workerProfiles.stream()
                             .filter(WorkerProfile::isDynamicSlot)
                             .filter(worker -> worker.getUnassignedResource().enoughThan(r))
-                            .findAny();
+                            .min(
+                                    Comparator.comparing(
+                                                    (WorkerProfile profile) ->
+                                                            profile.getUnassignedResource()
+                                                                    .getHeapMemory()
+                                                                    .getBytes())
+                                            .thenComparing(
+                                                    profile ->
+                                                            profile.getUnassignedResource()
+                                                                    .getCpu()
+                                                                    .getCore()));
         }
 
         return workerProfile;