[flink] 15/33: [hotfix][tests] Add DeclarativeSlotPoolServiceTest.testCloseReleasesAllSlotsForAllRegisteredTaskManagers

[tr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 6be9d3eda10332a594cb09fd06651fc2fe9e9438
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Jul 13 16:39:50 2021 +0200

    [hotfix][tests] Add DeclarativeSlotPoolServiceTest.testCloseReleasesAllSlotsForAllRegisteredTaskManagers
    
    This commit adds a test which tests that all slots for all registered TaskManagers are
    released upon calling DeclarativeSlotPoolService.close().
---
 .../slotpool/DeclarativeSlotPoolServiceTest.java   | 31 ++++++++++++++++++++++
 1 file changed, 31 insertions(+)

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceTest.java
index f0b0ee5..6d9c84b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolServiceTest.java
@@ -51,15 +51,19 @@ import org.junit.Test;
 
 import javax.annotation.Nonnull;
 
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Optional;
+import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertFalse;
@@ -274,6 +278,33 @@ public class DeclarativeSlotPoolServiceTest extends TestLogger {
         }
     }
 
+    @Test
+    public void testCloseReleasesAllSlotsForAllRegisteredTaskManagers() throws Exception {
+        final Queue<ResourceID> releasedSlotsFor = new ArrayDeque<>(2);
+        try (DeclarativeSlotPoolService declarativeSlotPoolService =
+                createDeclarativeSlotPoolService(
+                        new TestingDeclarativeSlotPoolFactory(
+                                new TestingDeclarativeSlotPoolBuilder()
+                                        .setReleaseSlotsFunction(
+                                                (resourceID, e) -> {
+                                                    releasedSlotsFor.offer(resourceID);
+                                                    return ResourceCounter.empty();
+                                                })))) {
+
+            final List<ResourceID> taskManagerResourceIds =
+                    Arrays.asList(
+                            ResourceID.generate(), ResourceID.generate(), ResourceID.generate());
+
+            for (ResourceID taskManagerResourceId : taskManagerResourceIds) {
+                declarativeSlotPoolService.registerTaskManager(taskManagerResourceId);
+            }
+
+            declarativeSlotPoolService.close();
+
+            assertThat(releasedSlotsFor, containsInAnyOrder(taskManagerResourceIds.toArray()));
+        }
+    }
+
     private DeclarativeSlotPoolService createDeclarativeSlotPoolService() throws Exception {
         return createDeclarativeSlotPoolService(new DefaultDeclarativeSlotPoolFactory());
     }