[fix][zeta] add and update ut

seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/AbstractSeaTunnelServerTest.java

@@ -81,7 +81,7 @@ public void before() {
         Config hazelcastConfig = Config.loadFromString(yaml);
         hazelcastConfig.setClusterName(
                 TestUtils.getClusterName("AbstractSeaTunnelServerTest_" + name));
-        SeaTunnelConfig seaTunnelConfig = ConfigProvider.locateAndGetSeaTunnelConfig();
+        SeaTunnelConfig seaTunnelConfig = loadSeaTunnelConfig();
         seaTunnelConfig.setHazelcastConfig(hazelcastConfig);
         instance = SeaTunnelServerStarter.createHazelcastInstance(seaTunnelConfig);
         nodeEngine = instance.node.nodeEngine;

seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointStorageTest.java

@@ -48,7 +48,7 @@ public class CheckpointStorageTest extends AbstractSeaTunnelServerTest {
     public SeaTunnelConfig loadSeaTunnelConfig() {
         SeaTunnelConfig seaTunnelConfig = super.loadSeaTunnelConfig();
         CheckpointConfig checkpointConfig = seaTunnelConfig.getEngineConfig().getCheckpointConfig();
-        // set a bigger interval in here and config file to avoid auto trigger checkpoint affect
+        // set a big interval in here and config file to avoid auto trigger checkpoint affect
         // test result
         checkpointConfig.setCheckpointInterval(Integer.MAX_VALUE);
         seaTunnelConfig.getEngineConfig().setCheckpointConfig(checkpointConfig);
@@ -61,7 +61,6 @@ public void testGenerateFileWhenSavepoint()
         long jobId = System.currentTimeMillis();
         CheckpointConfig checkpointConfig =
                 server.getSeaTunnelConfig().getEngineConfig().getCheckpointConfig();
-        server.getSeaTunnelConfig().getEngineConfig().setCheckpointConfig(checkpointConfig);
 
         CheckpointStorage checkpointStorage =
                 FactoryUtil.discoverFactory(
@@ -96,7 +95,6 @@ public void testBatchJob() throws CheckpointStorageException {
         long jobId = System.currentTimeMillis();
         CheckpointConfig checkpointConfig =
                 server.getSeaTunnelConfig().getEngineConfig().getCheckpointConfig();
-        server.getSeaTunnelConfig().getEngineConfig().setCheckpointConfig(checkpointConfig);
 
         CheckpointStorage checkpointStorage =
                 FactoryUtil.discoverFactory(

seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/FixSlotResourceTest.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.engine.server.resourcemanager;
+
+import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
+import org.apache.seatunnel.engine.common.config.server.SlotServiceConfig;
+import org.apache.seatunnel.engine.server.AbstractSeaTunnelServerTest;
+import org.apache.seatunnel.engine.server.resourcemanager.resource.ResourceProfile;
+import org.apache.seatunnel.engine.server.resourcemanager.resource.SlotProfile;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+public class FixSlotResourceTest extends AbstractSeaTunnelServerTest {
+
+    @Override
+    public SeaTunnelConfig loadSeaTunnelConfig() {
+        SeaTunnelConfig seaTunnelConfig = super.loadSeaTunnelConfig();
+        SlotServiceConfig slotServiceConfig =
+                seaTunnelConfig.getEngineConfig().getSlotServiceConfig();
+        slotServiceConfig.setDynamicSlot(false);
+        slotServiceConfig.setSlotNum(3);
+        seaTunnelConfig.getEngineConfig().setSlotServiceConfig(slotServiceConfig);
+        return seaTunnelConfig;
+    }
+
+    @Test
+    public void testEnoughResource() throws ExecutionException, InterruptedException {
+        List<ResourceProfile> resourceProfiles = new ArrayList<>();
+        resourceProfiles.add(new ResourceProfile());
+        resourceProfiles.add(new ResourceProfile());
+        resourceProfiles.add(new ResourceProfile());
+        List<SlotProfile> slotProfiles =
+                server.getCoordinatorService()
+                        .getResourceManager()
+                        .applyResources(1L, resourceProfiles)
+                        .get();
+        Assertions.assertEquals(slotProfiles.size(), 3);
+    }
+
+    @Test
+    public void testNotEnoughResource() throws ExecutionException, InterruptedException {
+        List<ResourceProfile> resourceProfiles = new ArrayList<>();
+        resourceProfiles.add(new ResourceProfile());
+        resourceProfiles.add(new ResourceProfile());
+        resourceProfiles.add(new ResourceProfile());
+        resourceProfiles.add(new ResourceProfile());
+        try {
+            server.getCoordinatorService()
+                    .getResourceManager()
+                    .applyResources(1L, resourceProfiles)
+                    .get();
+        } catch (ExecutionException e) {
+            Assertions.assertTrue(e.getMessage().contains("NoEnoughResourceException"));
+        }
+        resourceProfiles.remove(0);
+        List<SlotProfile> slotProfiles =
+                server.getCoordinatorService()
+                        .getResourceManager()
+                        .applyResources(1L, resourceProfiles)
+                        .get();
+        Assertions.assertEquals(slotProfiles.size(), 3);
+    }
+}