[Improve][Test] Fix test unstable on ResourceManger and EventReport module (#6620)

Author: Hisoka-X

Diff for: .github/workflows/backend.yml

@@ -556,7 +556,7 @@ jobs:
       - name: run seatunnel zeta integration test
         if: needs.changes.outputs.api == 'true'
         run: |
-          ./mvnw -T 1 -B verify -DskipUT=true -DskipIT=false -D"license.skipAddThirdParty"=true --no-snapshot-updates -pl :connector-seatunnel-e2e-base -am -Pci
+          ./mvnw -T 1 -B verify -DskipUT=true -DskipIT=false -D"license.skipAddThirdParty"=true --no-snapshot-updates -pl :connector-seatunnel-e2e-base,:connector-console-seatunnel-e2e -am -Pci
         env:
           MAVEN_OPTS: -Xmx4096m
   engine-k8s-it:
@@ -578,6 +578,8 @@ jobs:
         env:
           KUBECONFIG: /etc/rancher/k3s/k3s.yaml
       - uses: actions/checkout@v2
+      - name: free disk space
+        run: tools/github/free_disk_space.sh
       - name: Set up JDK ${{ matrix.java }}
         uses: actions/setup-java@v3
         with:

Diff for: seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainer.java

@@ -248,6 +248,8 @@ private static boolean isSystemThread(String s) {
                 || s.startsWith("Timer-")
                 || s.contains("InterruptTimer")
                 || s.contains("Java2D Disposer")
+                || s.contains("OkHttp ConnectionPool")
+                || s.startsWith("http-report-event-scheduler")
                 || s.contains(
                         "org.apache.hadoop.fs.FileSystem$Statistics$StatisticsDataReferenceCleaner")
                 || s.startsWith("Log4j2-TF-")

Diff for: seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManagerFunctionTest.java

@@ -27,11 +27,15 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
+import com.hazelcast.cluster.Address;
+
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
 
-public class ResourceManagerTest extends AbstractSeaTunnelServerTest {
+public class ResourceManagerTest extends AbstractSeaTunnelServerTest<ResourceManagerTest> {
 
     private ResourceManager resourceManager;
 
@@ -77,4 +81,27 @@ public void testApplyRequest() throws ExecutionException, InterruptedException {
                                         new ResourceProfile(CPU.of(0), Memory.of(Long.MAX_VALUE)))
                                 .get());
     }
+
+    @Test
+    public void testApplyResourceWithRandomResult()
+            throws ExecutionException, InterruptedException {
+        FakeResourceManager resourceManager = new FakeResourceManager(nodeEngine);
+
+        List<ResourceProfile> resourceProfiles = new ArrayList<>();
+        resourceProfiles.add(new ResourceProfile());
+        resourceProfiles.add(new ResourceProfile());
+        resourceProfiles.add(new ResourceProfile());
+        resourceProfiles.add(new ResourceProfile());
+        resourceProfiles.add(new ResourceProfile());
+        List<SlotProfile> slotProfiles = resourceManager.applyResources(1L, resourceProfiles).get();
+        Assertions.assertEquals(slotProfiles.size(), 5);
+
+        boolean hasDifferentWorker = false;
+        for (int i = 0; i < 5; i++) {
+            Set<Address> addresses =
+                    slotProfiles.stream().map(SlotProfile::getWorker).collect(Collectors.toSet());
+            hasDifferentWorker = addresses.size() > 1;
+        }
+        Assertions.assertTrue(hasDifferentWorker, "should have different worker for each slot");
+    }
 }