You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2023/04/25 07:42:37 UTC
[flink] 01/04: [hotfix][tests] Migrate SlotSelectionStrategy and SlotPool related tests to JUnit5
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 25926d1f90df59d013d9ada3d2613c8d778c42d1
Author: Weihua Hu <hu...@gmail.com>
AuthorDate: Fri Apr 21 19:06:57 2023 +0800
[hotfix][tests] Migrate SlotSelectionStrategy and SlotPool related tests to JUnit5
---
...ocationPreferenceSlotSelectionStrategyTest.java | 139 ++++-----
...reviousAllocationSlotSelectionStrategyTest.java | 35 +--
.../types/SlotSelectionStrategyTestBase.java | 9 +-
.../slotpool/BlocklistDeclarativeSlotPoolTest.java | 2 +-
.../slotpool/DefaultAllocatedSlotPoolTest.java | 178 +++++-------
.../slotpool/DefaultDeclarativeSlotPoolTest.java | 320 +++++++++------------
.../jobmaster/slotpool/SlotPoolTestUtils.java | 27 ++
.../runtime/scheduler/DefaultSchedulerTest.java | 2 +-
.../scheduler/adaptive/AdaptiveSchedulerTest.java | 2 +-
9 files changed, 321 insertions(+), 393 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/LocationPreferenceSlotSelectionStrategyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/LocationPreferenceSlotSelectionStrategyTest.java
index e6caaaf7725..c090b375ab7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/LocationPreferenceSlotSelectionStrategyTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/LocationPreferenceSlotSelectionStrategyTest.java
@@ -19,42 +19,35 @@
package org.apache.flink.runtime.clusterframework.types;
import org.apache.flink.runtime.jobmanager.scheduler.Locality;
-import org.apache.flink.runtime.jobmaster.SlotInfo;
import org.apache.flink.runtime.jobmaster.slotpool.LocationPreferenceSlotSelectionStrategy;
+import org.apache.flink.runtime.jobmaster.slotpool.SlotInfoWithUtilization;
import org.apache.flink.runtime.jobmaster.slotpool.SlotSelectionStrategy;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
-import org.hamcrest.FeatureMatcher;
-import org.hamcrest.Matcher;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import java.net.InetAddress;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Optional;
+import java.util.Set;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link LocationPreferenceSlotSelectionStrategy}. */
-public class LocationPreferenceSlotSelectionStrategyTest extends SlotSelectionStrategyTestBase {
+class LocationPreferenceSlotSelectionStrategyTest extends SlotSelectionStrategyTestBase {
- public LocationPreferenceSlotSelectionStrategyTest() {
- super(LocationPreferenceSlotSelectionStrategy.createDefault());
- }
-
- protected LocationPreferenceSlotSelectionStrategyTest(
- SlotSelectionStrategy slotSelectionStrategy) {
- super(slotSelectionStrategy);
+ @BeforeEach
+ void setUp() {
+ this.selectionStrategy = LocationPreferenceSlotSelectionStrategy.createDefault();
}
@Test
- public void testPhysicalSlotResourceProfileRespected() {
+ void testPhysicalSlotResourceProfileRespected() {
- SlotProfile slotProfile =
+ final SlotProfile slotProfile =
SlotProfile.priorAllocation(
resourceProfile,
biggerResourceProfile,
@@ -63,17 +56,23 @@ public class LocationPreferenceSlotSelectionStrategyTest extends SlotSelectionSt
Collections.emptySet());
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
- Assert.assertTrue(
- match.get()
- .getSlotInfo()
- .getResourceProfile()
- .isMatching(slotProfile.getPhysicalSlotResourceProfile()));
+ assertThat(match)
+ .hasValueSatisfying(
+ slotInfoAndLocality ->
+ assertThat(
+ slotInfoAndLocality
+ .getSlotInfo()
+ .getResourceProfile()
+ .isMatching(
+ slotProfile
+ .getPhysicalSlotResourceProfile()))
+ .isTrue());
ResourceProfile evenBiggerResourceProfile =
ResourceProfile.fromResources(
biggerResourceProfile.getCpuCores().getValue().doubleValue() + 1.0,
resourceProfile.getTaskHeapMemory().getMebiBytes());
- slotProfile =
+ final SlotProfile slotProfileNotMatching =
SlotProfile.priorAllocation(
resourceProfile,
evenBiggerResourceProfile,
@@ -81,49 +80,33 @@ public class LocationPreferenceSlotSelectionStrategyTest extends SlotSelectionSt
Collections.emptyList(),
Collections.emptySet());
- match = runMatching(slotProfile);
- Assert.assertFalse(match.isPresent());
+ match = runMatching(slotProfileNotMatching);
+ assertThat(match).isNotPresent();
}
@Test
- public void matchNoRequirements() {
+ void matchNoRequirements() {
SlotProfile slotProfile = SlotProfileTestingUtils.noRequirements();
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
- Assert.assertTrue(match.isPresent());
- final SlotSelectionStrategy.SlotInfoAndLocality slotInfoAndLocality = match.get();
- assertThat(candidates, hasItem(withSlotInfo(slotInfoAndLocality.getSlotInfo())));
- assertThat(slotInfoAndLocality, hasLocality(Locality.UNCONSTRAINED));
+ assertMatchingSlotLocalityAndInCandidates(match, Locality.UNCONSTRAINED, candidates);
}
@Test
- public void returnsHostLocalMatchingIfExactTMLocationCannotBeFulfilled() {
+ void returnsHostLocalMatchingIfExactTMLocationCannotBeFulfilled() {
SlotProfile slotProfile =
SlotProfileTestingUtils.preferredLocality(
resourceProfile, Collections.singletonList(tmlX));
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
- Assert.assertTrue(match.isPresent());
- final SlotSelectionStrategy.SlotInfoAndLocality slotInfoAndLocality = match.get();
- final SlotInfo slotInfo = slotInfoAndLocality.getSlotInfo();
- assertThat(candidates, hasItem(withSlotInfo(slotInfo)));
- assertThat(slotInfoAndLocality, hasLocality(Locality.HOST_LOCAL));
- }
-
- private Matcher<SlotSelectionStrategy.SlotInfoAndLocality> hasLocality(Locality locality) {
- return new LocalityFeatureMatcher(locality);
- }
-
- private Matcher<SlotSelectionStrategy.SlotInfoAndResources> withSlotInfo(SlotInfo slotInfo) {
- return new SlotInfoFeatureMatcher(slotInfo);
+ assertMatchingSlotLocalityAndInCandidates(match, Locality.HOST_LOCAL, candidates);
}
@Test
- public void
- returnsNonLocalMatchingIfResourceProfileCanBeFulfilledButNotTheTMLocationPreferences()
- throws Exception {
+ void returnsNonLocalMatchingIfResourceProfileCanBeFulfilledButNotTheTMLocationPreferences()
+ throws Exception {
final InetAddress nonHostLocalInetAddress =
InetAddress.getByAddress(new byte[] {10, 0, 0, 24});
final TaskManagerLocation nonLocalTm =
@@ -134,39 +117,36 @@ public class LocationPreferenceSlotSelectionStrategyTest extends SlotSelectionSt
resourceProfile, Collections.singletonList(nonLocalTm));
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
- Assert.assertTrue(match.isPresent());
- final SlotSelectionStrategy.SlotInfoAndLocality slotInfoAndLocality = match.get();
- assertThat(candidates, hasItem(withSlotInfo(slotInfoAndLocality.getSlotInfo())));
- assertThat(slotInfoAndLocality, hasLocality(Locality.NON_LOCAL));
+ assertMatchingSlotLocalityAndInCandidates(match, Locality.NON_LOCAL, candidates);
}
@Test
- public void matchPreferredLocation() {
+ void matchPreferredLocation() {
SlotProfile slotProfile =
SlotProfileTestingUtils.preferredLocality(
biggerResourceProfile, Collections.singletonList(tml2));
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
- Assert.assertEquals(slotInfo2, match.get().getSlotInfo());
+ assertMatchingSlotEqualsToSlotInfo(match, slotInfo2);
slotProfile =
SlotProfileTestingUtils.preferredLocality(
resourceProfile, Arrays.asList(tmlX, tml4));
match = runMatching(slotProfile);
- Assert.assertEquals(slotInfo4, match.get().getSlotInfo());
+ assertMatchingSlotEqualsToSlotInfo(match, slotInfo4);
slotProfile =
SlotProfileTestingUtils.preferredLocality(
resourceProfile, Arrays.asList(tml3, tml1, tml3, tmlX));
match = runMatching(slotProfile);
- Assert.assertEquals(slotInfo3, match.get().getSlotInfo());
+ assertMatchingSlotEqualsToSlotInfo(match, slotInfo3);
}
@Test
- public void matchPreviousLocationAvailableButAlsoBlacklisted() {
+ void matchPreviousLocationAvailableButAlsoBlacklisted() {
HashSet<AllocationID> blacklisted = new HashSet<>(4);
blacklisted.add(aid1);
blacklisted.add(aid2);
@@ -182,30 +162,33 @@ public class LocationPreferenceSlotSelectionStrategyTest extends SlotSelectionSt
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
// available previous allocation should override blacklisting
- Assert.assertEquals(slotInfo3, match.get().getSlotInfo());
+ assertMatchingSlotEqualsToSlotInfo(match, slotInfo3);
}
- private static class SlotInfoFeatureMatcher
- extends FeatureMatcher<SlotSelectionStrategy.SlotInfoAndResources, SlotInfo> {
- SlotInfoFeatureMatcher(SlotInfo slotInfo) {
- super(is(slotInfo), "Slot info of a SlotInfoAndResources instance", "slotInfo");
- }
-
- @Override
- protected SlotInfo featureValueOf(SlotSelectionStrategy.SlotInfoAndResources actual) {
- return actual.getSlotInfo();
- }
+ protected static void assertMatchingSlotEqualsToSlotInfo(
+ Optional<SlotSelectionStrategy.SlotInfoAndLocality> matchingSlot,
+ SlotInfoWithUtilization slotInfo) {
+ assertThat(matchingSlot)
+ .hasValueSatisfying(
+ slotInfoAndLocality ->
+ assertThat(slotInfoAndLocality.getSlotInfo()).isEqualTo(slotInfo));
}
- private static class LocalityFeatureMatcher
- extends FeatureMatcher<SlotSelectionStrategy.SlotInfoAndLocality, Locality> {
- LocalityFeatureMatcher(Locality locality) {
- super(is(locality), "Locality of SlotInfoAndLocality instance", "locality");
- }
-
- @Override
- protected Locality featureValueOf(SlotSelectionStrategy.SlotInfoAndLocality actual) {
- return actual.getLocality();
- }
+ protected static void assertMatchingSlotLocalityAndInCandidates(
+ Optional<SlotSelectionStrategy.SlotInfoAndLocality> matchingSlot,
+ Locality locality,
+ Set<SlotSelectionStrategy.SlotInfoAndResources> candidates) {
+ assertThat(matchingSlot)
+ .hasValueSatisfying(
+ slotInfoAndLocality -> {
+ assertThat(slotInfoAndLocality.getLocality()).isEqualTo(locality);
+ assertThat(candidates)
+ .anySatisfy(
+ slotInfoAndResources ->
+ assertThat(slotInfoAndResources.getSlotInfo())
+ .isEqualTo(
+ slotInfoAndLocality
+ .getSlotInfo()));
+ });
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/PreviousAllocationSlotSelectionStrategyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/PreviousAllocationSlotSelectionStrategyTest.java
index a62be78b040..2e029718cc5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/PreviousAllocationSlotSelectionStrategyTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/PreviousAllocationSlotSelectionStrategyTest.java
@@ -21,24 +21,28 @@ package org.apache.flink.runtime.clusterframework.types;
import org.apache.flink.runtime.jobmaster.slotpool.PreviousAllocationSlotSelectionStrategy;
import org.apache.flink.runtime.jobmaster.slotpool.SlotSelectionStrategy;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Optional;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Tests for {@link PreviousAllocationSlotSelectionStrategy}. */
-public class PreviousAllocationSlotSelectionStrategyTest
+class PreviousAllocationSlotSelectionStrategyTest
extends LocationPreferenceSlotSelectionStrategyTest {
- public PreviousAllocationSlotSelectionStrategyTest() {
- super(PreviousAllocationSlotSelectionStrategy.create());
+ @BeforeEach
+ @Override
+ void setUp() {
+ this.selectionStrategy = PreviousAllocationSlotSelectionStrategy.create();
}
@Test
- public void matchPreviousAllocationOverridesPreferredLocation() {
+ void matchPreviousAllocationOverridesPreferredLocation() {
SlotProfile slotProfile =
SlotProfile.priorAllocation(
@@ -48,8 +52,7 @@ public class PreviousAllocationSlotSelectionStrategyTest
Collections.singleton(aid3),
Collections.emptySet());
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
-
- Assert.assertEquals(slotInfo3, match.get().getSlotInfo());
+ assertMatchingSlotEqualsToSlotInfo(match, slotInfo3);
slotProfile =
SlotProfile.priorAllocation(
@@ -59,12 +62,11 @@ public class PreviousAllocationSlotSelectionStrategyTest
new HashSet<>(Arrays.asList(aidX, aid2)),
Collections.emptySet());
match = runMatching(slotProfile);
-
- Assert.assertEquals(slotInfo2, match.get().getSlotInfo());
+ assertMatchingSlotEqualsToSlotInfo(match, slotInfo2);
}
@Test
- public void matchPreviousLocationNotAvailableButByLocality() {
+ void matchPreviousLocationNotAvailableButByLocality() {
SlotProfile slotProfile =
SlotProfile.priorAllocation(
@@ -74,12 +76,11 @@ public class PreviousAllocationSlotSelectionStrategyTest
Collections.singleton(aidX),
Collections.emptySet());
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
-
- Assert.assertEquals(slotInfo4, match.get().getSlotInfo());
+ assertMatchingSlotEqualsToSlotInfo(match, slotInfo4);
}
@Test
- public void matchPreviousLocationNotAvailableAndAllOthersBlacklisted() {
+ void matchPreviousLocationNotAvailableAndAllOthersBlacklisted() {
HashSet<AllocationID> blacklisted = new HashSet<>(4);
blacklisted.add(aid1);
blacklisted.add(aid2);
@@ -95,11 +96,11 @@ public class PreviousAllocationSlotSelectionStrategyTest
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
// there should be no valid option left and we expect null as return
- Assert.assertFalse(match.isPresent());
+ assertThat(match).isNotPresent();
}
@Test
- public void matchPreviousLocationNotAvailableAndSomeOthersBlacklisted() {
+ void matchPreviousLocationNotAvailableAndSomeOthersBlacklisted() {
HashSet<AllocationID> blacklisted = new HashSet<>(3);
blacklisted.add(aid2);
blacklisted.add(aid3);
@@ -114,6 +115,6 @@ public class PreviousAllocationSlotSelectionStrategyTest
Optional<SlotSelectionStrategy.SlotInfoAndLocality> match = runMatching(slotProfile);
// we expect that the candidate that is not blacklisted is returned
- Assert.assertEquals(slotInfo1, match.get().getSlotInfo());
+ assertMatchingSlotEqualsToSlotInfo(match, slotInfo1);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/SlotSelectionStrategyTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/SlotSelectionStrategyTestBase.java
index d4cd8d995d0..53a80750881 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/SlotSelectionStrategyTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/SlotSelectionStrategyTestBase.java
@@ -24,7 +24,6 @@ import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
import org.apache.flink.runtime.jobmaster.slotpool.SlotInfoWithUtilization;
import org.apache.flink.runtime.jobmaster.slotpool.SlotSelectionStrategy;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
-import org.apache.flink.util.TestLogger;
import java.net.InetAddress;
import java.util.Collections;
@@ -33,7 +32,7 @@ import java.util.Optional;
import java.util.Set;
/** Test base for {@link SlotSelectionStrategy}. */
-public abstract class SlotSelectionStrategyTestBase extends TestLogger {
+abstract class SlotSelectionStrategyTestBase {
protected final ResourceProfile resourceProfile = ResourceProfile.fromResources(2, 1024);
protected final ResourceProfile biggerResourceProfile = ResourceProfile.fromResources(3, 1024);
@@ -74,11 +73,7 @@ public abstract class SlotSelectionStrategyTestBase extends TestLogger {
protected final Set<SlotSelectionStrategy.SlotInfoAndResources> candidates =
Collections.unmodifiableSet(createCandidates());
- protected final SlotSelectionStrategy selectionStrategy;
-
- public SlotSelectionStrategyTestBase(SlotSelectionStrategy slotSelectionStrategy) {
- this.selectionStrategy = slotSelectionStrategy;
- }
+ protected SlotSelectionStrategy selectionStrategy;
private Set<SlotSelectionStrategy.SlotInfoAndResources> createCandidates() {
Set<SlotSelectionStrategy.SlotInfoAndResources> candidates = new HashSet<>(4);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/BlocklistDeclarativeSlotPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/BlocklistDeclarativeSlotPoolTest.java
index 5decc5ac1d1..47afbca838d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/BlocklistDeclarativeSlotPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/BlocklistDeclarativeSlotPoolTest.java
@@ -47,9 +47,9 @@ import java.util.stream.Collectors;
import static org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPoolTest.FreeSlotConsumer;
import static org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPoolTest.NewSlotsService;
import static org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPoolTest.createResourceRequirements;
-import static org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPoolTest.createSlotOffersForResourceRequirements;
import static org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPoolTest.drainNewSlotService;
import static org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPoolTest.increaseRequirementsAndOfferSlotsToSlotPool;
+import static org.apache.flink.runtime.jobmaster.slotpool.SlotPoolTestUtils.createSlotOffersForResourceRequirements;
import static org.apache.flink.shaded.guava30.com.google.common.collect.Iterables.getOnlyElement;
import static org.assertj.core.api.Assertions.assertThat;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultAllocatedSlotPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultAllocatedSlotPoolTest.java
index d26d5bc4def..2cd04102f19 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultAllocatedSlotPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultAllocatedSlotPoolTest.java
@@ -23,18 +23,13 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.jobmaster.RpcTaskManagerGateway;
-import org.apache.flink.runtime.jobmaster.SlotInfo;
import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables;
-import org.hamcrest.Description;
-import org.hamcrest.Matcher;
-import org.hamcrest.TypeSafeMatcher;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -49,20 +44,15 @@ import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
-import static org.hamcrest.CoreMatchers.hasItems;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.Matchers.closeTo;
-import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.sameInstance;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.offset;
/** Tests for the {@link DefaultAllocatedSlotPool}. */
-public class DefaultAllocatedSlotPoolTest extends TestLogger {
+class DefaultAllocatedSlotPoolTest {
@Test
- public void testAddSlots() {
+ void testAddSlots() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final Collection<AllocatedSlot> slots = createAllocatedSlots();
@@ -74,7 +64,7 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
}
@Test
- public void testRemoveSlot() {
+ void testRemoveSlot() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final Collection<AllocatedSlot> slots = createAllocatedSlots();
@@ -91,7 +81,7 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
}
@Test
- public void testRemoveSlots() {
+ void testRemoveSlots() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final ResourceID owner = ResourceID.generate();
@@ -107,7 +97,7 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
}
@Test
- public void testRemoveSlotsReturnValue() {
+ void testRemoveSlotsReturnValue() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final ResourceID owner = ResourceID.generate();
@@ -120,36 +110,37 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
final AllocatedSlotPool.AllocatedSlotsAndReservationStatus
allocatedSlotsAndReservationStatus = slotPool.removeSlots(owner);
- assertThat(allocatedSlotsAndReservationStatus.getAllocatedSlots(), hasItems(slot1, slot2));
- assertThat(allocatedSlotsAndReservationStatus.wasFree(slot1.getAllocationId()), is(false));
- assertThat(allocatedSlotsAndReservationStatus.wasFree(slot2.getAllocationId()), is(true));
+ assertThat(allocatedSlotsAndReservationStatus.getAllocatedSlots())
+ .containsExactlyInAnyOrder(slot1, slot2);
+ assertThat(allocatedSlotsAndReservationStatus.wasFree(slot1.getAllocationId())).isFalse();
+ assertThat(allocatedSlotsAndReservationStatus.wasFree(slot2.getAllocationId())).isTrue();
}
@Test
- public void testContainsSlots() {
+ void testContainsSlots() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final ResourceID owner = ResourceID.generate();
final AllocatedSlot allocatedSlot = createAllocatedSlot(owner);
slotPool.addSlots(Collections.singleton(allocatedSlot), 0);
- assertTrue(slotPool.containsSlots(owner));
- assertFalse(slotPool.containsSlots(ResourceID.generate()));
+ assertThat(slotPool.containsSlots(owner)).isTrue();
+ assertThat(slotPool.containsSlots(ResourceID.generate())).isFalse();
}
@Test
- public void testContainsSlot() {
+ void testContainsSlot() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final AllocatedSlot allocatedSlot = createAllocatedSlot(null);
slotPool.addSlots(Collections.singleton(allocatedSlot), 0);
- assertTrue(slotPool.containsSlot(allocatedSlot.getAllocationId()));
- assertFalse(slotPool.containsSlot(new AllocationID()));
+ assertThat(slotPool.containsSlot(allocatedSlot.getAllocationId())).isTrue();
+ assertThat(slotPool.containsSlot(new AllocationID())).isFalse();
}
@Test
- public void testReserveFreeSlot() {
+ void testReserveFreeSlot() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final Collection<AllocatedSlot> allSlots = createAllocatedSlots();
final Collection<AllocatedSlot> freeSlots = new ArrayList<>(allSlots);
@@ -159,27 +150,27 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
slotPool.addSlots(allSlots, 0);
- assertThat(
- slotPool.reserveFreeSlot(allocatedSlot.getAllocationId()),
- sameInstance(allocatedSlot));
+ assertThat(slotPool.reserveFreeSlot(allocatedSlot.getAllocationId()))
+ .isEqualTo(allocatedSlot);
assertSlotPoolContainsFreeSlots(slotPool, freeSlots);
assertSlotPoolContainsSlots(slotPool, allSlots);
}
- @Test(expected = IllegalStateException.class)
- public void testReserveNonFreeSlotFails() {
+ @Test
+ void testReserveNonFreeSlotFails() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final AllocatedSlot slot = createAllocatedSlot(null);
slotPool.addSlots(Collections.singleton(slot), 0);
slotPool.reserveFreeSlot(slot.getAllocationId());
- slotPool.reserveFreeSlot(slot.getAllocationId());
+ assertThatThrownBy(() -> slotPool.reserveFreeSlot(slot.getAllocationId()))
+ .isInstanceOf(IllegalStateException.class);
}
@Test
- public void testFreeingOfReservedSlot() {
+ void testFreeingOfReservedSlot() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final Collection<AllocatedSlot> slots = createAllocatedSlots();
@@ -191,7 +182,7 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
slotPool.reserveFreeSlot(slot.getAllocationId());
final int releaseTime = 1;
- assertTrue(slotPool.freeReservedSlot(slot.getAllocationId(), releaseTime).isPresent());
+ assertThat(slotPool.freeReservedSlot(slot.getAllocationId(), releaseTime)).isPresent();
assertSlotPoolContainsFreeSlots(slotPool, slots);
for (AllocatedSlotPool.FreeSlotInfo freeSlotInfo : slotPool.getFreeSlotsInformation()) {
@@ -202,89 +193,90 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
time = initialTime;
}
- assertThat(freeSlotInfo.getFreeSince(), is(time));
+ assertThat(freeSlotInfo.getFreeSince()).isEqualTo(time);
}
}
@Test
- public void testFreeingOfFreeSlotIsIgnored() {
+ void testFreeingOfFreeSlotIsIgnored() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final AllocatedSlot slot = createAllocatedSlot(null);
slotPool.addSlots(Collections.singleton(slot), 0);
- assertFalse(slotPool.freeReservedSlot(slot.getAllocationId(), 1).isPresent());
+ assertThat(slotPool.freeReservedSlot(slot.getAllocationId(), 1)).isNotPresent();
final AllocatedSlotPool.FreeSlotInfo freeSlotInfo =
Iterables.getOnlyElement(slotPool.getFreeSlotsInformation());
- assertThat(freeSlotInfo.getFreeSince(), is(0L));
+ assertThat(freeSlotInfo.getFreeSince()).isEqualTo(0L);
}
@Test
- public void testSlotUtilizationCalculation() {
+ void testSlotUtilizationCalculation() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final ResourceID owner = ResourceID.generate();
final Collection<AllocatedSlot> slots = createAllocatedSlotsWithOwner(owner);
slotPool.addSlots(slots, 0);
- for (AllocatedSlotPool.FreeSlotInfo freeSlotInfo : slotPool.getFreeSlotsInformation()) {
- assertThat(freeSlotInfo.asSlotInfo().getTaskExecutorUtilization(), closeTo(0, 0.1));
- }
+ assertThat(slotPool.getFreeSlotsInformation())
+ .allSatisfy(
+ freeSlotInfo ->
+ assertThat(freeSlotInfo.asSlotInfo().getTaskExecutorUtilization())
+ .isCloseTo(0, offset(0.1)));
int numAllocatedSlots = 0;
for (AllocatedSlot slot : slots) {
- assertThat(slotPool.reserveFreeSlot(slot.getAllocationId()), sameInstance(slot));
+ assertThat(slotPool.reserveFreeSlot(slot.getAllocationId())).isEqualTo(slot);
numAllocatedSlots++;
for (AllocatedSlotPool.FreeSlotInfo freeSlotInfo : slotPool.getFreeSlotsInformation()) {
final double utilization = (double) numAllocatedSlots / slots.size();
- assertThat(
- freeSlotInfo.asSlotInfo().getTaskExecutorUtilization(),
- closeTo(utilization, 0.1));
+ assertThat(freeSlotInfo.asSlotInfo().getTaskExecutorUtilization())
+ .isCloseTo(utilization, offset(0.1));
}
}
}
@Test
- public void testRemoveSlotsOfUnknownOwnerIsIgnored() {
+ void testRemoveSlotsOfUnknownOwnerIsIgnored() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
slotPool.removeSlots(ResourceID.generate());
}
@Test
- public void testContainsFreeSlotReturnsTrueIfSlotIsFree() {
+ void testContainsFreeSlotReturnsTrueIfSlotIsFree() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final AllocatedSlot allocatedSlot = createAllocatedSlot(ResourceID.generate());
slotPool.addSlots(Collections.singleton(allocatedSlot), 0);
- assertTrue(slotPool.containsFreeSlot(allocatedSlot.getAllocationId()));
+ assertThat(slotPool.containsFreeSlot(allocatedSlot.getAllocationId())).isTrue();
}
@Test
- public void testContainsFreeSlotReturnsFalseIfSlotDoesNotExist() {
+ void testContainsFreeSlotReturnsFalseIfSlotDoesNotExist() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
- assertFalse(slotPool.containsFreeSlot(new AllocationID()));
+ assertThat(slotPool.containsFreeSlot(new AllocationID())).isFalse();
}
@Test
- public void testContainsFreeSlotReturnsFalseIfSlotIsReserved() {
+ void testContainsFreeSlotReturnsFalseIfSlotIsReserved() {
final DefaultAllocatedSlotPool slotPool = new DefaultAllocatedSlotPool();
final AllocatedSlot allocatedSlot = createAllocatedSlot(ResourceID.generate());
slotPool.addSlots(Collections.singleton(allocatedSlot), 0);
slotPool.reserveFreeSlot(allocatedSlot.getAllocationId());
- assertFalse(slotPool.containsFreeSlot(allocatedSlot.getAllocationId()));
+ assertThat(slotPool.containsFreeSlot(allocatedSlot.getAllocationId())).isFalse();
}
private void assertSlotPoolContainsSlots(
DefaultAllocatedSlotPool slotPool, Collection<AllocatedSlot> slots) {
- assertThat(slotPool.getAllSlotsInformation(), hasSize(slots.size()));
+ assertThat(slotPool.getAllSlotsInformation()).hasSize(slots.size());
final Map<AllocationID, AllocatedSlot> slotsPerAllocationId =
slots.stream()
@@ -292,13 +284,11 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
Collectors.toMap(
AllocatedSlot::getAllocationId, Function.identity()));
- for (SlotInfo slotInfo : slotPool.getAllSlotsInformation()) {
- assertTrue(slotsPerAllocationId.containsKey(slotInfo.getAllocationId()));
- final AllocatedSlot allocatedSlot =
- slotsPerAllocationId.get(slotInfo.getAllocationId());
-
- assertThat(slotInfo, matchesPhysicalSlot(allocatedSlot));
- }
+ assertThat(slotPool.getAllSlotsInformation())
+ .allSatisfy(
+ slotInfo ->
+ assertThat(slotsPerAllocationId.get(slotInfo.getAllocationId()))
+ .isEqualTo(slotInfo));
}
private void assertSlotPoolContainsFreeSlots(
@@ -306,7 +296,7 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
final Collection<AllocatedSlotPool.FreeSlotInfo> freeSlotsInformation =
slotPool.getFreeSlotsInformation();
- assertThat(freeSlotsInformation, hasSize(allocatedSlots.size()));
+ assertThat(freeSlotsInformation).hasSize(allocatedSlots.size());
final Map<AllocationID, AllocatedSlot> allocatedSlotMap =
allocatedSlots.stream()
@@ -314,17 +304,23 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
Collectors.toMap(
AllocatedSlot::getAllocationId, Function.identity()));
- for (AllocatedSlotPool.FreeSlotInfo freeSlotInfo : freeSlotsInformation) {
- assertTrue(allocatedSlotMap.containsKey(freeSlotInfo.getAllocationId()));
-
- assertThat(
- freeSlotInfo.asSlotInfo(),
- matchesPhysicalSlot(allocatedSlotMap.get(freeSlotInfo.getAllocationId())));
- }
- }
-
- static Matcher<SlotInfo> matchesPhysicalSlot(PhysicalSlot allocatedSlot) {
- return new SlotInfoMatcher(allocatedSlot);
+ assertThat(freeSlotsInformation)
+ .allSatisfy(
+ freeSlotInfo -> {
+ AllocatedSlot allocatedSlot =
+ allocatedSlotMap.get(freeSlotInfo.getAllocationId());
+ assertThat(allocatedSlot).isNotNull();
+ SlotInfoWithUtilization slotInfoWithUtilization =
+ freeSlotInfo.asSlotInfo();
+ assertThat(allocatedSlot.getAllocationId())
+ .isEqualTo(slotInfoWithUtilization.getAllocationId());
+ assertThat(allocatedSlot.getPhysicalSlotNumber())
+ .isEqualTo(slotInfoWithUtilization.getPhysicalSlotNumber());
+ assertThat(allocatedSlot.getResourceProfile())
+ .isEqualTo(slotInfoWithUtilization.getResourceProfile());
+ assertThat(allocatedSlot.getTaskManagerLocation())
+ .isEqualTo(slotInfoWithUtilization.getTaskManagerLocation());
+ });
}
private Collection<AllocatedSlot> createAllocatedSlots() {
@@ -356,34 +352,4 @@ public class DefaultAllocatedSlotPoolTest extends TestLogger {
new TestingTaskExecutorGatewayBuilder().createTestingTaskExecutorGateway(),
JobMasterId.generate()));
}
-
- static class SlotInfoMatcher extends TypeSafeMatcher<SlotInfo> {
-
- private final PhysicalSlot physicalSlot;
-
- SlotInfoMatcher(PhysicalSlot physicalSlot) {
- this.physicalSlot = physicalSlot;
- }
-
- @Override
- public void describeTo(Description description) {
- description.appendText("SlotInfo with values: ");
- description.appendValueList(
- "{",
- ",",
- "}",
- physicalSlot.getAllocationId(),
- physicalSlot.getPhysicalSlotNumber(),
- physicalSlot.getResourceProfile(),
- physicalSlot.getTaskManagerLocation());
- }
-
- @Override
- protected boolean matchesSafely(SlotInfo item) {
- return item.getAllocationId().equals(physicalSlot.getAllocationId())
- && item.getPhysicalSlotNumber() == physicalSlot.getPhysicalSlotNumber()
- && item.getResourceProfile().equals(physicalSlot.getResourceProfile())
- && item.getTaskManagerLocation().equals(physicalSlot.getTaskManagerLocation());
- }
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclarativeSlotPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclarativeSlotPoolTest.java
index 8127c7eb2c5..dcf2d2eeb67 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclarativeSlotPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclarativeSlotPoolTest.java
@@ -32,14 +32,11 @@ import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.util.ResourceCounter;
import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.function.QuadConsumer;
import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables;
-import org.hamcrest.Description;
-import org.hamcrest.TypeSafeMatcher;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -57,19 +54,11 @@ import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
-import static org.hamcrest.CoreMatchers.hasItems;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.CoreMatchers.not;
-import static org.hamcrest.CoreMatchers.sameInstance;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.Matchers.hasSize;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.apache.flink.runtime.jobmaster.slotpool.SlotPoolTestUtils.createSlotOffersForResourceRequirements;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link DefaultDeclarativeSlotPool}. */
-public class DefaultDeclarativeSlotPoolTest extends TestLogger {
+class DefaultDeclarativeSlotPoolTest {
private static final ResourceProfile RESOURCE_PROFILE_1 =
ResourceProfile.newBuilder().setCpuCores(1.7).build();
@@ -77,7 +66,7 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
ResourceProfile.newBuilder().setManagedMemoryMB(100).build();
@Test
- public void testIncreasingResourceRequirementsWillSendResourceRequirementNotification()
+ void testIncreasingResourceRequirementsWillSendResourceRequirementNotification()
throws InterruptedException {
final NewResourceRequirementsService requirementsListener =
new NewResourceRequirementsService();
@@ -88,19 +77,17 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
slotPool.increaseResourceRequirementsBy(increment1);
slotPool.increaseResourceRequirementsBy(increment2);
- assertThat(
- requirementsListener.takeResourceRequirements(),
- is(toResourceRequirements(increment1)));
+ assertThat(requirementsListener.takeResourceRequirements())
+ .isEqualTo(toResourceRequirements(increment1));
final ResourceCounter totalResources = increment1.add(increment2);
- assertThat(
- requirementsListener.takeResourceRequirements(),
- is(toResourceRequirements(totalResources)));
- assertThat(requirementsListener.hasNextResourceRequirements(), is(false));
+ assertThat(requirementsListener.takeResourceRequirements())
+ .isEqualTo(toResourceRequirements(totalResources));
+ assertThat(requirementsListener.hasNextResourceRequirements()).isFalse();
}
@Test
- public void testDecreasingResourceRequirementsWillSendResourceRequirementNotification()
+ void testDecreasingResourceRequirementsWillSendResourceRequirementNotification()
throws InterruptedException {
final NewResourceRequirementsService requirementsListener =
new NewResourceRequirementsService();
@@ -116,32 +103,28 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
slotPool.decreaseResourceRequirementsBy(decrement);
final ResourceCounter totalResources = increment.subtract(decrement);
- assertThat(
- requirementsListener.takeResourceRequirements(),
- is(toResourceRequirements(totalResources)));
- assertThat(requirementsListener.hasNextResourceRequirements(), is(false));
+ assertThat(requirementsListener.takeResourceRequirements())
+ .isEqualTo(toResourceRequirements(totalResources));
+ assertThat(requirementsListener.hasNextResourceRequirements()).isFalse();
}
@Test
- public void testGetResourceRequirements() {
+ void testGetResourceRequirements() {
final DefaultDeclarativeSlotPool slotPool =
DefaultDeclarativeSlotPoolBuilder.builder().build();
- assertThat(
- slotPool.getResourceRequirements(),
- is(toResourceRequirements(ResourceCounter.empty())));
+ assertThat(slotPool.getResourceRequirements()).isEmpty();
final ResourceCounter resourceRequirements = createResourceRequirements();
slotPool.increaseResourceRequirementsBy(resourceRequirements);
- assertThat(
- slotPool.getResourceRequirements(),
- is(toResourceRequirements(resourceRequirements)));
+ assertThat(slotPool.getResourceRequirements())
+ .isEqualTo(toResourceRequirements(resourceRequirements));
}
@Test
- public void testOfferSlots() throws InterruptedException {
+ void testOfferSlots() throws InterruptedException {
final NewSlotsService notifyNewSlots = new NewSlotsService();
final DefaultDeclarativeSlotPool slotPool =
createDefaultDeclarativeSlotPoolWithNewSlotsListener(notifyNewSlots);
@@ -156,26 +139,46 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
final Collection<SlotOffer> acceptedSlots =
SlotPoolTestUtils.offerSlots(slotPool, slotOffers);
- assertThat(acceptedSlots, containsInAnyOrder(slotOffers.toArray()));
-
- final Collection<PhysicalSlot> newSlots = drainNewSlotService(notifyNewSlots);
-
- assertThat(
- newSlots,
- containsInAnyOrder(
- slotOffers.stream()
- .map(DefaultDeclarativeSlotPoolTest::matchesSlotOffer)
- .collect(Collectors.toList())));
- assertThat(
- slotPool.getAllSlotsInformation(),
- containsInAnyOrder(
- newSlots.stream()
- .map(DefaultAllocatedSlotPoolTest::matchesPhysicalSlot)
- .collect(Collectors.toList())));
+ assertThat(acceptedSlots).containsExactlyInAnyOrderElementsOf(slotOffers);
+
+ final Map<AllocationID, PhysicalSlot> newSlotsById =
+ drainNewSlotService(notifyNewSlots).stream()
+ .collect(
+ Collectors.toMap(
+ PhysicalSlot::getAllocationId, Function.identity()));
+
+ assertThat(slotOffers)
+ .hasSize(newSlotsById.size())
+ .allSatisfy(
+ slotOffer -> {
+ PhysicalSlot slot = newSlotsById.get(slotOffer.getAllocationId());
+ assertThat(slot).isNotNull();
+ assertThat(slotOffer.getAllocationId())
+ .isEqualTo(slot.getAllocationId());
+ assertThat(slotOffer.getSlotIndex())
+ .isEqualTo(slot.getPhysicalSlotNumber());
+ assertThat(slotOffer.getResourceProfile())
+ .isEqualTo(slot.getResourceProfile());
+ });
+ assertThat(slotPool.getAllSlotsInformation())
+ .hasSize(newSlotsById.size())
+ .allSatisfy(
+ slotInfo -> {
+ PhysicalSlot slot = newSlotsById.get(slotInfo.getAllocationId());
+ assertThat(slot).isNotNull();
+ assertThat(slotInfo.getAllocationId())
+ .isEqualTo(slot.getAllocationId());
+ assertThat(slotInfo.getPhysicalSlotNumber())
+ .isEqualTo(slot.getPhysicalSlotNumber());
+ assertThat(slotInfo.getResourceProfile())
+ .isEqualTo(slot.getResourceProfile());
+ assertThat(slotInfo.getTaskManagerLocation())
+ .isEqualTo(slot.getTaskManagerLocation());
+ });
}
@Test
- public void testDuplicateSlotOfferings() throws InterruptedException {
+ void testDuplicateSlotOfferings() throws InterruptedException {
final NewSlotsService notifyNewSlots = new NewSlotsService();
final DefaultDeclarativeSlotPool slotPool =
createDefaultDeclarativeSlotPoolWithNewSlotsListener(notifyNewSlots);
@@ -194,13 +197,13 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
final Collection<SlotOffer> acceptedSlots =
SlotPoolTestUtils.offerSlots(slotPool, slotOffers);
- assertThat(acceptedSlots, containsInAnyOrder(slotOffers.toArray()));
+ assertThat(acceptedSlots).containsExactlyInAnyOrderElementsOf(slotOffers);
// duplicate slots should not trigger notify new slots
- assertFalse(notifyNewSlots.hasNextNewSlots());
+ assertThat(notifyNewSlots.hasNextNewSlots()).isFalse();
}
@Test
- public void testOfferingTooManySlotsWillRejectSuperfluousSlots() {
+ void testOfferingTooManySlotsWillRejectSuperfluousSlots() {
final NewSlotsService notifyNewSlots = new NewSlotsService();
final DefaultDeclarativeSlotPool slotPool =
createDefaultDeclarativeSlotPoolWithNewSlotsListener(notifyNewSlots);
@@ -223,16 +226,17 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
.map(SlotOffer::getResourceProfile)
.collect(Collectors.groupingBy(Function.identity(), Collectors.counting()));
- for (Map.Entry<ResourceProfile, Integer> resourceCount :
- resourceRequirements.getResourcesWithCount()) {
- assertThat(
- resourceProfileCount.getOrDefault(resourceCount.getKey(), 0L),
- is((long) resourceCount.getValue()));
- }
+ assertThat(resourceRequirements.getResourcesWithCount())
+ .allSatisfy(
+ resourceCount ->
+ assertThat(
+ resourceProfileCount.getOrDefault(
+ resourceCount.getKey(), 0L))
+ .isEqualTo((long) resourceCount.getValue()));
}
@Test
- public void testReleaseSlotsRemovesSlots() throws InterruptedException {
+ void testReleaseSlotsRemovesSlots() throws InterruptedException {
final NewResourceRequirementsService notifyNewResourceRequirements =
new NewResourceRequirementsService();
final DefaultDeclarativeSlotPool slotPool =
@@ -246,11 +250,11 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
slotPool.releaseSlots(
taskManagerLocation.getResourceID(), new FlinkException("Test failure"));
- assertThat(slotPool.getAllSlotsInformation(), is(empty()));
+ assertThat(slotPool.getAllSlotsInformation()).isEmpty();
}
@Test
- public void testReleaseSlotsReturnsSlot() {
+ void testReleaseSlotsReturnsSlot() {
final DefaultDeclarativeSlotPool slotPool =
DefaultDeclarativeSlotPoolBuilder.builder().build();
@@ -275,13 +279,15 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
final Collection<AllocationID> freedSlots = freeSlotConsumer.drainFreedSlots();
- assertThat(
- freedSlots,
- containsInAnyOrder(slotOffers.stream().map(SlotOffer::getAllocationId).toArray()));
+ assertThat(freedSlots)
+ .containsExactlyInAnyOrderElementsOf(
+ slotOffers.stream()
+ .map(SlotOffer::getAllocationId)
+ .collect(Collectors.toList()));
}
@Test
- public void testReleaseSlotsOnlyReturnsFulfilledRequirementsOfReservedSlots() {
+ void testReleaseSlotsOnlyReturnsFulfilledRequirementsOfReservedSlots() {
withSlotPoolContainingOneTaskManagerWithTwoSlotsWithUniqueResourceProfiles(
(slotPool, freeSlot, slotToReserve, taskManagerLocation) -> {
slotPool.reserveFreeSlot(
@@ -295,17 +301,18 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
new FlinkException("Test failure"));
assertThat(
- fulfilledRequirements.getResourceCount(freeSlot.getResourceProfile()),
- is(0));
+ fulfilledRequirements.getResourceCount(
+ freeSlot.getResourceProfile()))
+ .isEqualTo(0);
assertThat(
- fulfilledRequirements.getResourceCount(
- slotToReserve.getResourceProfile()),
- is(1));
+ fulfilledRequirements.getResourceCount(
+ slotToReserve.getResourceProfile()))
+ .isEqualTo(1);
});
}
@Test
- public void testReleaseSlotOnlyReturnsFulfilledRequirementsOfReservedSlots() {
+ void testReleaseSlotOnlyReturnsFulfilledRequirementsOfReservedSlots() {
withSlotPoolContainingOneTaskManagerWithTwoSlotsWithUniqueResourceProfiles(
(slotPool, freeSlot, slotToReserve, ignored) -> {
slotPool.reserveFreeSlot(
@@ -321,11 +328,11 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
slotToReserve.getAllocationId(),
new FlinkException("Test failure"));
- assertThat(fulfilledRequirementsOfFreeSlot.getResources(), is(empty()));
+ assertThat(fulfilledRequirementsOfFreeSlot.getResources()).isEmpty();
assertThat(
- fulfilledRequirementsOfReservedSlot.getResourceCount(
- slotToReserve.getResourceProfile()),
- is(1));
+ fulfilledRequirementsOfReservedSlot.getResourceCount(
+ slotToReserve.getResourceProfile()))
+ .isEqualTo(1);
});
}
@@ -360,8 +367,7 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
}
@Test
- public void testReleaseSlotDecreasesFulfilledResourceRequirements()
- throws InterruptedException {
+ void testReleaseSlotDecreasesFulfilledResourceRequirements() throws InterruptedException {
final NewSlotsService notifyNewSlots = new NewSlotsService();
final DefaultDeclarativeSlotPool slotPool =
createDefaultDeclarativeSlotPoolWithNewSlotsListener(notifyNewSlots);
@@ -377,11 +383,12 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
final ResourceCounter finalResourceRequirements =
resourceRequirements.subtract(physicalSlot.getResourceProfile(), 1);
- assertThat(slotPool.getFulfilledResourceRequirements(), is(finalResourceRequirements));
+ assertThat(slotPool.getFulfilledResourceRequirements())
+ .isEqualTo(finalResourceRequirements);
}
@Test
- public void testReleaseSlotReturnsSlot() throws InterruptedException {
+ void testReleaseSlotReturnsSlot() throws InterruptedException {
final NewSlotsService notifyNewSlots = new NewSlotsService();
final DefaultDeclarativeSlotPool slotPool =
createDefaultDeclarativeSlotPoolWithNewSlotsListener(notifyNewSlots);
@@ -407,11 +414,11 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
final AllocationID freedSlot = Iterables.getOnlyElement(freeSlotConsumer.drainFreedSlots());
- assertThat(freedSlot, is(physicalSlot.getAllocationId()));
+ assertThat(freedSlot).isEqualTo(physicalSlot.getAllocationId());
}
@Test
- public void testReturnIdleSlotsAfterTimeout() {
+ void testReturnIdleSlotsAfterTimeout() {
final Time idleSlotTimeout = Time.seconds(10);
final long offerTime = 0;
final DefaultDeclarativeSlotPool slotPool =
@@ -440,22 +447,22 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
final Collection<AllocationID> freedSlots = freeSlotConsumer.drainFreedSlots();
- assertThat(acceptedSlots, is(not(empty())));
- assertThat(
- freedSlots,
- containsInAnyOrder(
- acceptedSlots.stream().map(SlotOffer::getAllocationId).toArray()));
+ assertThat(freedSlots)
+ .containsExactlyInAnyOrderElementsOf(
+ acceptedSlots.stream()
+ .map(SlotOffer::getAllocationId)
+ .collect(Collectors.toList()));
assertNoAvailableAndRequiredResources(slotPool);
}
private void assertNoAvailableAndRequiredResources(DefaultDeclarativeSlotPool slotPool) {
- assertTrue(slotPool.getFulfilledResourceRequirements().isEmpty());
- assertTrue(slotPool.getResourceRequirements().isEmpty());
- assertThat(slotPool.getAllSlotsInformation(), is(empty()));
+ assertThat(slotPool.getFulfilledResourceRequirements().isEmpty()).isTrue();
+ assertThat(slotPool.getResourceRequirements()).isEmpty();
+ assertThat(slotPool.getAllSlotsInformation()).isEmpty();
}
@Test
- public void testOnlyReturnExcessIdleSlots() {
+ void testOnlyReturnExcessIdleSlots() {
final Time idleSlotTimeout = Time.seconds(10);
final long offerTime = 0;
final DefaultDeclarativeSlotPool slotPool =
@@ -478,12 +485,12 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
slotPool.releaseIdleSlots(offerTime + idleSlotTimeout.toMilliseconds());
- assertThat(acceptedSlots, is(not(empty())));
- assertThat(slotPool.getFulfilledResourceRequirements(), is(requiredResources));
+ assertThat(acceptedSlots).isNotEmpty();
+ assertThat(slotPool.getFulfilledResourceRequirements()).isEqualTo(requiredResources);
}
@Test
- public void testFreedSlotWillBeUsedToFulfillOutstandingResourceRequirementsOfSameProfile()
+ void testFreedSlotWillBeUsedToFulfillOutstandingResourceRequirementsOfSameProfile()
throws InterruptedException {
final NewSlotsService notifyNewSlots = new NewSlotsService();
final DefaultDeclarativeSlotPool slotPool =
@@ -502,7 +509,7 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
final Collection<PhysicalSlot> recycledSlots = drainNewSlotService(notifyNewSlots);
- assertThat(Iterables.getOnlyElement(recycledSlots), sameInstance(newSlot));
+ assertThat(Iterables.getOnlyElement(recycledSlots)).isEqualTo(newSlot);
final Collection<SlotOffer> newSlotOffers =
createSlotOffersForResourceRequirements(initialRequirements);
@@ -516,12 +523,12 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
SlotPoolTestUtils.createTaskManagerGateway(null),
0);
- assertThat(acceptedSlots, is(empty()));
- assertTrue(slotPool.calculateUnfulfilledResources().isEmpty());
+ assertThat(acceptedSlots).isEmpty();
+ assertThat(slotPool.calculateUnfulfilledResources().isEmpty()).isTrue();
}
@Test
- public void testRegisterSlotsAcceptsAllSlots() {
+ void testRegisterSlotsAcceptsAllSlots() {
final DefaultDeclarativeSlotPool declarativeSlotPool = createDefaultDeclarativeSlotPool();
final int numberSlots = 10;
final Collection<SlotOffer> slots =
@@ -537,15 +544,17 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
final Collection<? extends SlotInfo> allSlotsInformation =
declarativeSlotPool.getAllSlotsInformation();
- assertThat(allSlotsInformation, hasSize(numberSlots));
+ assertThat(allSlotsInformation).hasSize(numberSlots);
- for (SlotInfo slotInfo : allSlotsInformation) {
- assertThat(slotInfo.getResourceProfile(), is(RESOURCE_PROFILE_1));
- }
+ assertThat(allSlotsInformation)
+ .allSatisfy(
+ slotInfo ->
+ assertThat(slotInfo.getResourceProfile())
+ .isEqualTo(RESOURCE_PROFILE_1));
}
@Test
- public void testFreedSlotWillRemainAssignedToMatchedResourceProfile() {
+ void testFreedSlotWillRemainAssignedToMatchedResourceProfile() {
final DefaultDeclarativeSlotPool slotPool = new DefaultDeclarativeSlotPoolBuilder().build();
final ResourceProfile largeResourceProfile =
@@ -564,8 +573,9 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
slotPool.reserveFreeSlot(slot.getAllocationId(), largeResourceProfile);
assertThat(
- slotPool.getFulfilledResourceRequirements().getResourceCount(largeResourceProfile),
- is(1));
+ slotPool.getFulfilledResourceRequirements()
+ .getResourceCount(largeResourceProfile))
+ .isEqualTo(1);
slotPool.increaseResourceRequirementsBy(
ResourceCounter.withResource(smallResourceProfile, 1));
@@ -578,15 +588,17 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
// profile
slotPool.freeReservedSlot(slot.getAllocationId(), null, 1);
assertThat(
- slotPool.getFulfilledResourceRequirements().getResourceCount(largeResourceProfile),
- is(1));
+ slotPool.getFulfilledResourceRequirements()
+ .getResourceCount(largeResourceProfile))
+ .isEqualTo(1);
assertThat(
- slotPool.getFulfilledResourceRequirements().getResourceCount(smallResourceProfile),
- is(0));
+ slotPool.getFulfilledResourceRequirements()
+ .getResourceCount(smallResourceProfile))
+ .isEqualTo(0);
}
@Test
- public void testReserveFreeSlotForResourceUpdatesAvailableResourcesAndRequirements() {
+ void testReserveFreeSlotForResourceUpdatesAvailableResourcesAndRequirements() {
final DefaultDeclarativeSlotPool slotPool = new DefaultDeclarativeSlotPoolBuilder().build();
final ResourceProfile largeResourceProfile =
@@ -611,21 +623,20 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
slotPool.reserveFreeSlot(largeSlot.getAllocationId(), smallResourceProfile);
ResourceCounter availableResources = slotPool.getFulfilledResourceRequirements();
- assertThat(availableResources.getResourceCount(smallResourceProfile), is(1));
- assertThat(availableResources.getResourceCount(largeResourceProfile), is(0));
+ assertThat(availableResources.getResourceCount(smallResourceProfile)).isEqualTo(1);
+ assertThat(availableResources.getResourceCount(largeResourceProfile)).isEqualTo(0);
Collection<ResourceRequirement> currentResourceRequirements =
slotPool.getResourceRequirements();
// since we used one of the large slots for fulfilling another profile, we now need another
// large slot for fulfill the original requirement
// conversely we no longer need the small slot, because we are now using another slot for it
- assertThat(
- currentResourceRequirements,
- hasItems(ResourceRequirement.create(largeResourceProfile, 2)));
+ assertThat(currentResourceRequirements)
+ .containsExactly(ResourceRequirement.create(largeResourceProfile, 2));
}
@Test
- public void testSetResourceRequirementsForInitialResourceRequirements() {
+ void testSetResourceRequirementsForInitialResourceRequirements() {
final DefaultDeclarativeSlotPool slotPool = new DefaultDeclarativeSlotPoolBuilder().build();
final ResourceCounter resourceRequirements =
@@ -633,13 +644,12 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
slotPool.setResourceRequirements(resourceRequirements);
- assertThat(
- slotPool.getResourceRequirements(),
- is(toResourceRequirements(resourceRequirements)));
+ assertThat(slotPool.getResourceRequirements())
+ .isEqualTo(toResourceRequirements(resourceRequirements));
}
@Test
- public void testSetResourceRequirementsOverwritesPreviousValue() {
+ void testSetResourceRequirementsOverwritesPreviousValue() {
final DefaultDeclarativeSlotPool slotPool = new DefaultDeclarativeSlotPoolBuilder().build();
slotPool.setResourceRequirements(ResourceCounter.withResource(RESOURCE_PROFILE_1, 1));
@@ -648,13 +658,12 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
ResourceCounter.withResource(RESOURCE_PROFILE_2, 1);
slotPool.setResourceRequirements(resourceRequirements);
- assertThat(
- slotPool.getResourceRequirements(),
- is(toResourceRequirements(resourceRequirements)));
+ assertThat(slotPool.getResourceRequirements())
+ .isEqualTo(toResourceRequirements(resourceRequirements));
}
@Test
- public void testRegisterSlotsDoesNotAffectRequirements() {
+ void testRegisterSlotsDoesNotAffectRequirements() {
final DefaultDeclarativeSlotPool slotPool = new DefaultDeclarativeSlotPoolBuilder().build();
final ResourceProfile slotProfile = RESOURCE_PROFILE_1;
@@ -670,14 +679,14 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
final AllocationID allocationId =
slotPool.getFreeSlotsInformation().iterator().next().getAllocationId();
- assertThat(slotPool.getResourceRequirements(), is(empty()));
+ assertThat(slotPool.getResourceRequirements()).isEmpty();
slotPool.increaseResourceRequirementsBy(ResourceCounter.withResource(requestedProfile, 1));
slotPool.reserveFreeSlot(allocationId, requestedProfile);
slotPool.freeReservedSlot(allocationId, null, 1L);
slotPool.decreaseResourceRequirementsBy(ResourceCounter.withResource(requestedProfile, 1));
- assertThat(slotPool.getResourceRequirements(), is(empty()));
+ assertThat(slotPool.getResourceRequirements()).isEmpty();
}
@Nonnull
@@ -689,28 +698,6 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
return ResourceCounter.withResources(requirements);
}
- @Nonnull
- public static Collection<SlotOffer> createSlotOffersForResourceRequirements(
- ResourceCounter resourceRequirements) {
- Collection<SlotOffer> slotOffers = new ArrayList<>();
- int slotIndex = 0;
-
- for (Map.Entry<ResourceProfile, Integer> resourceWithCount :
- resourceRequirements.getResourcesWithCount()) {
- for (int i = 0; i < resourceWithCount.getValue(); i++) {
- ResourceProfile slotProfile = resourceWithCount.getKey();
- slotOffers.add(
- new SlotOffer(
- new AllocationID(),
- slotIndex++,
- slotProfile == ResourceProfile.UNKNOWN
- ? ResourceProfile.ANY
- : slotProfile));
- }
- }
- return slotOffers;
- }
-
@Nonnull
private static Collection<ResourceRequirement> toResourceRequirements(
ResourceCounter resourceCounter) {
@@ -782,10 +769,6 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
return newSlots;
}
- private static TypeSafeMatcher<PhysicalSlot> matchesSlotOffer(SlotOffer slotOffer) {
- return new PhysicalSlotSlotOfferMatcher(slotOffer);
- }
-
private static final class NewResourceRequirementsService
implements Consumer<Collection<ResourceRequirement>> {
@@ -827,33 +810,6 @@ public class DefaultDeclarativeSlotPoolTest extends TestLogger {
}
}
- private static class PhysicalSlotSlotOfferMatcher extends TypeSafeMatcher<PhysicalSlot> {
- private final SlotOffer slotOffer;
-
- public PhysicalSlotSlotOfferMatcher(SlotOffer slotOffer) {
- this.slotOffer = slotOffer;
- }
-
- @Override
- protected boolean matchesSafely(PhysicalSlot item) {
- return item.getAllocationId().equals(slotOffer.getAllocationId())
- && item.getResourceProfile().equals(slotOffer.getResourceProfile())
- && item.getPhysicalSlotNumber() == slotOffer.getSlotIndex();
- }
-
- @Override
- public void describeTo(Description description) {
- description.appendText("SlotOffer: ");
- description.appendValueList(
- "{",
- ",",
- "}",
- slotOffer.getAllocationId(),
- slotOffer.getResourceProfile(),
- slotOffer.getSlotIndex());
- }
- }
-
static class FreeSlotConsumer
implements BiFunction<AllocationID, Throwable, CompletableFuture<Acknowledge>> {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTestUtils.java
index 3754ac427e3..22b64701f69 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTestUtils.java
@@ -18,6 +18,8 @@
package org.apache.flink.runtime.jobmaster.slotpool;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.jobmaster.RpcTaskManagerGateway;
@@ -26,11 +28,14 @@ import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
import org.apache.flink.runtime.taskexecutor.slot.SlotOffer;
import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+import org.apache.flink.runtime.util.ResourceCounter;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Map;
/** Utilities for testing slot pool implementations. */
public final class SlotPoolTestUtils {
@@ -79,4 +84,26 @@ public final class SlotPoolTestUtils {
TaskManagerGateway taskManagerGateway) {
return slotPool.offerSlots(new LocalTaskManagerLocation(), taskManagerGateway, slotOffers);
}
+
+ @Nonnull
+ public static Collection<SlotOffer> createSlotOffersForResourceRequirements(
+ ResourceCounter resourceRequirements) {
+ Collection<SlotOffer> slotOffers = new ArrayList<>();
+ int slotIndex = 0;
+
+ for (Map.Entry<ResourceProfile, Integer> resourceWithCount :
+ resourceRequirements.getResourcesWithCount()) {
+ for (int i = 0; i < resourceWithCount.getValue(); i++) {
+ ResourceProfile slotProfile = resourceWithCount.getKey();
+ slotOffers.add(
+ new SlotOffer(
+ new AllocationID(),
+ slotIndex++,
+ slotProfile == ResourceProfile.UNKNOWN
+ ? ResourceProfile.ANY
+ : slotProfile));
+ }
+ }
+ return slotOffers;
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java
index 4f3212d1c24..bc6df6ee0a8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerTest.java
@@ -130,7 +130,7 @@ import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
-import static org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPoolTest.createSlotOffersForResourceRequirements;
+import static org.apache.flink.runtime.jobmaster.slotpool.SlotPoolTestUtils.createSlotOffersForResourceRequirements;
import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.acknowledgePendingCheckpoint;
import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.createFailedTaskExecutionState;
import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.enableCheckpointing;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
index ebbc936c1dc..e5abbcfada5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
@@ -139,7 +139,7 @@ import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createNoOpVertex;
import static org.apache.flink.runtime.jobgraph.JobGraphTestUtils.streamingJobGraph;
-import static org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPoolTest.createSlotOffersForResourceRequirements;
+import static org.apache.flink.runtime.jobmaster.slotpool.SlotPoolTestUtils.createSlotOffersForResourceRequirements;
import static org.apache.flink.runtime.jobmaster.slotpool.SlotPoolTestUtils.offerSlots;
import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.enableCheckpointing;
import static org.assertj.core.api.Assertions.assertThat;