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/07 03:55:26 UTC
[flink] 01/03: [hotfix] Migrate some fine-grained slot manager tests to Junit5 and Assertj
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 ddc61595bd1c6a996dfde6563956b324d35b3602
Author: Weihua Hu <hu...@gmail.com>
AuthorDate: Thu Mar 16 16:00:27 2023 +0800
[hotfix] Migrate some fine-grained slot manager tests to Junit5 and Assertj
---
.../slotmanager/DefaultSlotStatusSyncerTest.java | 154 ++++++-------
.../FineGrainedTaskManagerTrackerTest.java | 237 +++++++++++----------
2 files changed, 191 insertions(+), 200 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java
index 7d37cb884fb..48e07d95997 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java
@@ -33,41 +33,34 @@ import org.apache.flink.runtime.taskexecutor.SlotStatus;
import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGateway;
import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
import org.apache.flink.util.concurrent.FutureUtils;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeoutException;
-import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.Matchers.instanceOf;
-import static org.hamcrest.core.Is.is;
-import static org.hamcrest.core.IsEqual.equalTo;
-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;
/** Tests for the {@link DefaultSlotStatusSyncer}. */
-public class DefaultSlotStatusSyncerTest extends TestLogger {
+class DefaultSlotStatusSyncerTest {
private static final Time TASK_MANAGER_REQUEST_TIMEOUT = Time.seconds(10);
private static final TaskExecutorConnection TASK_EXECUTOR_CONNECTION =
new TaskExecutorConnection(
ResourceID.generate(),
new TestingTaskExecutorGatewayBuilder().createTestingTaskExecutorGateway());
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
+ @RegisterExtension
+ static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_RESOURCE =
+ TestingUtils.defaultExecutorExtension();
@Test
- public void testAllocateSlot() throws Exception {
+ void testAllocateSlot() throws Exception {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
final CompletableFuture<
@@ -109,23 +102,21 @@ public class DefaultSlotStatusSyncerTest extends TestLogger {
"address",
ResourceProfile.ANY);
final AllocationID allocationId = requestFuture.get().f2;
- assertThat(
- resourceTracker.getAcquiredResources(jobId),
- contains(ResourceRequirement.create(ResourceProfile.ANY, 1)));
- assertTrue(taskManagerTracker.getAllocatedOrPendingSlot(allocationId).isPresent());
- assertThat(
- taskManagerTracker.getAllocatedOrPendingSlot(allocationId).get().getJobId(),
- is(jobId));
- assertThat(
- taskManagerTracker.getAllocatedOrPendingSlot(allocationId).get().getState(),
- is(SlotState.PENDING));
+ assertThat(resourceTracker.getAcquiredResources(jobId))
+ .contains(ResourceRequirement.create(ResourceProfile.ANY, 1));
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId))
+ .hasValueSatisfying(
+ slot -> {
+ assertThat(slot.getJobId()).isEqualTo(jobId);
+ assertThat(slot.getState()).isEqualTo(SlotState.PENDING);
+ });
responseFuture.complete(Acknowledge.get());
- assertFalse(allocatedFuture.isCompletedExceptionally());
+ assertThat(allocatedFuture).isNotCompletedExceptionally();
}
@Test
- public void testAllocateSlotFailsWithException() {
+ void testAllocateSlotFailsWithException() {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
final TestingTaskExecutorGateway taskExecutorGateway =
@@ -155,23 +146,19 @@ public class DefaultSlotStatusSyncerTest extends TestLogger {
jobId,
"address",
ResourceProfile.ANY);
- try {
- allocatedFuture.get();
- } catch (Exception e) {
- assertThat(e.getCause(), instanceOf(TimeoutException.class));
- }
- assertThat(resourceTracker.getAcquiredResources(jobId), is(empty()));
+
+ assertThatThrownBy(allocatedFuture::get).hasCauseInstanceOf(TimeoutException.class);
+ assertThat(resourceTracker.getAcquiredResources(jobId)).isEmpty();
assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(taskExecutorConnection.getInstanceID())
- .get()
- .getAllocatedSlots()
- .keySet(),
- is(empty()));
+ taskManagerTracker.getRegisteredTaskManager(
+ taskExecutorConnection.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAllocatedSlots()).isEmpty());
}
@Test
- public void testFreeSlot() {
+ void testFreeSlot() {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
final ResourceTracker resourceTracker = new DefaultResourceTracker();
@@ -195,18 +182,17 @@ public class DefaultSlotStatusSyncerTest extends TestLogger {
resourceTracker.notifyAcquiredResource(jobId, ResourceProfile.ANY);
slotStatusSyncer.freeSlot(allocationId);
- assertThat(resourceTracker.getAcquiredResources(jobId), is(empty()));
+ assertThat(resourceTracker.getAcquiredResources(jobId)).isEmpty();
assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(TASK_EXECUTOR_CONNECTION.getInstanceID())
- .get()
- .getAllocatedSlots()
- .keySet(),
- is(empty()));
+ taskManagerTracker.getRegisteredTaskManager(
+ TASK_EXECUTOR_CONNECTION.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAllocatedSlots()).isEmpty());
}
@Test
- public void testSlotStatusProcessing() {
+ void testSlotStatusProcessing() {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
final ResourceTracker resourceTracker = new DefaultResourceTracker();
@@ -245,29 +231,29 @@ public class DefaultSlotStatusSyncerTest extends TestLogger {
taskManagerTracker.addTaskManager(taskExecutorConnection, totalResource, totalResource);
slotStatusSyncer.reportSlotStatus(taskExecutorConnection.getInstanceID(), slotReport1);
+ assertThat(resourceTracker.getAcquiredResources(jobId))
+ .contains(ResourceRequirement.create(resource, 2));
assertThat(
- resourceTracker.getAcquiredResources(jobId),
- contains(ResourceRequirement.create(resource, 2)));
- assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(taskExecutorConnection.getInstanceID())
- .get()
- .getAvailableResource(),
- equalTo(ResourceProfile.fromResources(3, 12)));
- assertTrue(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1).isPresent());
- assertTrue(taskManagerTracker.getAllocatedOrPendingSlot(allocationId2).isPresent());
+ taskManagerTracker.getRegisteredTaskManager(
+ taskExecutorConnection.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAvailableResource())
+ .isEqualTo(ResourceProfile.fromResources(3, 12)));
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1)).isPresent();
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId2)).isPresent();
slotStatusSyncer.allocateSlot(
taskExecutorConnection.getInstanceID(), jobId, "address", resource);
+ assertThat(resourceTracker.getAcquiredResources(jobId))
+ .contains(ResourceRequirement.create(resource, 3));
assertThat(
- resourceTracker.getAcquiredResources(jobId),
- contains(ResourceRequirement.create(resource, 3)));
- assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(taskExecutorConnection.getInstanceID())
- .get()
- .getAvailableResource(),
- equalTo(ResourceProfile.fromResources(2, 8)));
+ taskManagerTracker.getRegisteredTaskManager(
+ taskExecutorConnection.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAvailableResource())
+ .isEqualTo(ResourceProfile.fromResources(2, 8)));
final AllocationID allocationId3 =
taskManagerTracker.getRegisteredTaskManager(taskExecutorConnection.getInstanceID())
.get().getAllocatedSlots().keySet().stream()
@@ -281,23 +267,21 @@ public class DefaultSlotStatusSyncerTest extends TestLogger {
// allocationId1 should still be allocated; allocationId2 should be freed; allocationId3
// should continue to be in a pending state;
slotStatusSyncer.reportSlotStatus(taskExecutorConnection.getInstanceID(), slotReport2);
+ assertThat(resourceTracker.getAcquiredResources(jobId))
+ .contains(ResourceRequirement.create(resource, 2));
assertThat(
- resourceTracker.getAcquiredResources(jobId),
- contains(ResourceRequirement.create(resource, 2)));
- assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(taskExecutorConnection.getInstanceID())
- .get()
- .getAvailableResource(),
- equalTo(ResourceProfile.fromResources(3, 12)));
- assertTrue(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1).isPresent());
- assertFalse(taskManagerTracker.getAllocatedOrPendingSlot(allocationId2).isPresent());
- assertTrue(taskManagerTracker.getAllocatedOrPendingSlot(allocationId3).isPresent());
- assertThat(
- taskManagerTracker.getAllocatedOrPendingSlot(allocationId1).get().getState(),
- is(SlotState.ALLOCATED));
- assertThat(
- taskManagerTracker.getAllocatedOrPendingSlot(allocationId3).get().getState(),
- is(SlotState.PENDING));
+ taskManagerTracker.getRegisteredTaskManager(
+ taskExecutorConnection.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAvailableResource())
+ .isEqualTo(ResourceProfile.fromResources(3, 12)));
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId2)).isNotPresent();
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1))
+ .hasValueSatisfying(
+ slot -> assertThat(slot.getState()).isEqualTo(SlotState.ALLOCATED));
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId3))
+ .hasValueSatisfying(
+ slot -> assertThat(slot.getState()).isEqualTo(SlotState.PENDING));
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTrackerTest.java
index b4e4e9adb76..d316387cd30 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTrackerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTrackerTest.java
@@ -25,63 +25,62 @@ import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.resourcemanager.registration.TaskExecutorConnection;
import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
import org.apache.flink.runtime.util.ResourceCounter;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.Map;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.core.Is.is;
-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;
/** Tests for the {@link FineGrainedTaskManagerTracker}. */
-public class FineGrainedTaskManagerTrackerTest extends TestLogger {
+class FineGrainedTaskManagerTrackerTest {
private static final TaskExecutorConnection TASK_EXECUTOR_CONNECTION =
new TaskExecutorConnection(
ResourceID.generate(),
new TestingTaskExecutorGatewayBuilder().createTestingTaskExecutorGateway());
@Test
- public void testInitState() {
+ void testInitState() {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
- assertThat(taskManagerTracker.getPendingTaskManagers(), is(empty()));
- assertThat(taskManagerTracker.getRegisteredTaskManagers(), is(empty()));
+ assertThat(taskManagerTracker.getPendingTaskManagers()).isEmpty();
+ assertThat(taskManagerTracker.getRegisteredTaskManagers()).isEmpty();
}
@Test
- public void testAddAndRemoveTaskManager() {
+ void testAddAndRemoveTaskManager() {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
// Add task manager
taskManagerTracker.addTaskManager(
TASK_EXECUTOR_CONNECTION, ResourceProfile.ANY, ResourceProfile.ANY);
- assertThat(taskManagerTracker.getRegisteredTaskManagers().size(), is(1));
- assertTrue(
- taskManagerTracker
- .getRegisteredTaskManager(TASK_EXECUTOR_CONNECTION.getInstanceID())
- .isPresent());
+ assertThat(taskManagerTracker.getRegisteredTaskManagers()).hasSize(1);
+ assertThat(
+ taskManagerTracker.getRegisteredTaskManager(
+ TASK_EXECUTOR_CONNECTION.getInstanceID()))
+ .isPresent();
// Remove task manager
taskManagerTracker.removeTaskManager(TASK_EXECUTOR_CONNECTION.getInstanceID());
- assertThat(taskManagerTracker.getRegisteredTaskManagers().size(), is(0));
+ assertThat(taskManagerTracker.getRegisteredTaskManagers()).isEmpty();
}
- @Test(expected = NullPointerException.class)
- public void testRemoveUnknownTaskManager() {
- final FineGrainedTaskManagerTracker taskManagerTracker =
- new FineGrainedTaskManagerTracker();
-
- taskManagerTracker.removeTaskManager(new InstanceID());
+ @Test
+ void testRemoveUnknownTaskManager() {
+ assertThatThrownBy(
+ () -> {
+ final FineGrainedTaskManagerTracker taskManagerTracker =
+ new FineGrainedTaskManagerTracker();
+ taskManagerTracker.removeTaskManager(new InstanceID());
+ })
+ .isInstanceOf(NullPointerException.class);
}
@Test
- public void testAddAndRemovePendingTaskManager() {
+ void testAddAndRemovePendingTaskManager() {
final PendingTaskManager pendingTaskManager =
new PendingTaskManager(ResourceProfile.ANY, 1);
final FineGrainedTaskManagerTracker taskManagerTracker =
@@ -96,45 +95,46 @@ public class FineGrainedTaskManagerTrackerTest extends TestLogger {
Collections.singletonMap(
pendingTaskManager.getPendingTaskManagerId(),
Collections.singletonMap(jobId, resourceCounter)));
- assertThat(taskManagerTracker.getPendingTaskManagers().size(), is(1));
+ assertThat(taskManagerTracker.getPendingTaskManagers()).hasSize(1);
assertThat(
- taskManagerTracker
- .getPendingTaskManagersByTotalAndDefaultSlotResourceProfile(
- ResourceProfile.ANY, ResourceProfile.ANY)
- .size(),
- is(1));
+ taskManagerTracker
+ .getPendingTaskManagersByTotalAndDefaultSlotResourceProfile(
+ ResourceProfile.ANY, ResourceProfile.ANY))
+ .hasSize(1);
// Remove pending task manager
final Map<JobID, ResourceCounter> records =
taskManagerTracker.removePendingTaskManager(
pendingTaskManager.getPendingTaskManagerId());
- assertThat(taskManagerTracker.getPendingTaskManagers(), is(empty()));
+ assertThat(taskManagerTracker.getPendingTaskManagers()).isEmpty();
assertThat(
- taskManagerTracker
- .getPendingAllocationsOfPendingTaskManager(
- pendingTaskManager.getPendingTaskManagerId())
- .size(),
- is(0));
+ taskManagerTracker.getPendingAllocationsOfPendingTaskManager(
+ pendingTaskManager.getPendingTaskManagerId()))
+ .isEmpty();
assertThat(
- taskManagerTracker
- .getPendingTaskManagersByTotalAndDefaultSlotResourceProfile(
- ResourceProfile.ANY, ResourceProfile.ANY)
- .size(),
- is(0));
- assertTrue(records.containsKey(jobId));
- assertThat(records.get(jobId).getResourceCount(ResourceProfile.ANY), is(1));
+ taskManagerTracker
+ .getPendingTaskManagersByTotalAndDefaultSlotResourceProfile(
+ ResourceProfile.ANY, ResourceProfile.ANY))
+ .isEmpty();
+ assertThat(records).containsKey(jobId);
+ assertThat(records.get(jobId).getResourceCount(ResourceProfile.ANY)).isEqualTo(1);
}
- @Test(expected = NullPointerException.class)
- public void testRemoveUnknownPendingTaskManager() {
- final FineGrainedTaskManagerTracker taskManagerTracker =
- new FineGrainedTaskManagerTracker();
+ @Test
+ void testRemoveUnknownPendingTaskManager() {
+ assertThatThrownBy(
+ () -> {
+ final FineGrainedTaskManagerTracker taskManagerTracker =
+ new FineGrainedTaskManagerTracker();
- taskManagerTracker.removePendingTaskManager(PendingTaskManagerId.generate());
+ taskManagerTracker.removePendingTaskManager(
+ PendingTaskManagerId.generate());
+ })
+ .isInstanceOf(NullPointerException.class);
}
@Test
- public void testSlotAllocation() {
+ void testSlotAllocation() {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
final ResourceProfile totalResource = ResourceProfile.fromResources(10, 1000);
@@ -149,13 +149,14 @@ public class FineGrainedTaskManagerTrackerTest extends TestLogger {
TASK_EXECUTOR_CONNECTION.getInstanceID(),
ResourceProfile.fromResources(3, 200),
SlotState.PENDING);
- assertTrue(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1).isPresent());
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1)).isPresent();
assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(TASK_EXECUTOR_CONNECTION.getInstanceID())
- .get()
- .getAvailableResource(),
- is(ResourceProfile.fromResources(7, 800)));
+ taskManagerTracker.getRegisteredTaskManager(
+ TASK_EXECUTOR_CONNECTION.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAvailableResource())
+ .isEqualTo(ResourceProfile.fromResources(7, 800)));
// Notify pending slot is now allocated
taskManagerTracker.notifySlotStatus(
@@ -164,13 +165,14 @@ public class FineGrainedTaskManagerTrackerTest extends TestLogger {
TASK_EXECUTOR_CONNECTION.getInstanceID(),
ResourceProfile.fromResources(3, 200),
SlotState.ALLOCATED);
- assertTrue(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1).isPresent());
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1)).isPresent();
assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(TASK_EXECUTOR_CONNECTION.getInstanceID())
- .get()
- .getAvailableResource(),
- is(ResourceProfile.fromResources(7, 800)));
+ taskManagerTracker.getRegisteredTaskManager(
+ TASK_EXECUTOR_CONNECTION.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAvailableResource())
+ .isEqualTo(ResourceProfile.fromResources(7, 800)));
// Notify free slot is now allocated
taskManagerTracker.notifySlotStatus(
@@ -179,17 +181,18 @@ public class FineGrainedTaskManagerTrackerTest extends TestLogger {
TASK_EXECUTOR_CONNECTION.getInstanceID(),
ResourceProfile.fromResources(2, 300),
SlotState.ALLOCATED);
- assertTrue(taskManagerTracker.getAllocatedOrPendingSlot(allocationId2).isPresent());
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId2)).isPresent();
assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(TASK_EXECUTOR_CONNECTION.getInstanceID())
- .get()
- .getAvailableResource(),
- is(ResourceProfile.fromResources(5, 500)));
+ taskManagerTracker.getRegisteredTaskManager(
+ TASK_EXECUTOR_CONNECTION.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAvailableResource())
+ .isEqualTo(ResourceProfile.fromResources(5, 500)));
}
@Test
- public void testFreeSlot() {
+ void testFreeSlot() {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
final ResourceProfile totalResource = ResourceProfile.fromResources(10, 1000);
@@ -217,13 +220,14 @@ public class FineGrainedTaskManagerTrackerTest extends TestLogger {
TASK_EXECUTOR_CONNECTION.getInstanceID(),
ResourceProfile.fromResources(3, 200),
SlotState.FREE);
- assertFalse(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1).isPresent());
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId1)).isNotPresent();
assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(TASK_EXECUTOR_CONNECTION.getInstanceID())
- .get()
- .getAvailableResource(),
- is(ResourceProfile.fromResources(8, 700)));
+ taskManagerTracker.getRegisteredTaskManager(
+ TASK_EXECUTOR_CONNECTION.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAvailableResource())
+ .isEqualTo(ResourceProfile.fromResources(8, 700)));
// Free allocated slot
taskManagerTracker.notifySlotStatus(
allocationId2,
@@ -231,30 +235,35 @@ public class FineGrainedTaskManagerTrackerTest extends TestLogger {
TASK_EXECUTOR_CONNECTION.getInstanceID(),
ResourceProfile.fromResources(2, 300),
SlotState.FREE);
- assertFalse(taskManagerTracker.getAllocatedOrPendingSlot(allocationId2).isPresent());
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId2)).isNotPresent();
assertThat(
- taskManagerTracker
- .getRegisteredTaskManager(TASK_EXECUTOR_CONNECTION.getInstanceID())
- .get()
- .getAvailableResource(),
- is(totalResource));
+ taskManagerTracker.getRegisteredTaskManager(
+ TASK_EXECUTOR_CONNECTION.getInstanceID()))
+ .hasValueSatisfying(
+ taskManagerInfo ->
+ assertThat(taskManagerInfo.getAvailableResource())
+ .isEqualTo(totalResource));
}
- @Test(expected = NullPointerException.class)
- public void testFreeUnknownSlot() {
- final FineGrainedTaskManagerTracker taskManagerTracker =
- new FineGrainedTaskManagerTracker();
+ @Test
+ void testFreeUnknownSlot() {
+ assertThatThrownBy(
+ () -> {
+ final FineGrainedTaskManagerTracker taskManagerTracker =
+ new FineGrainedTaskManagerTracker();
- taskManagerTracker.notifySlotStatus(
- new AllocationID(),
- new JobID(),
- new InstanceID(),
- ResourceProfile.ANY,
- SlotState.FREE);
+ taskManagerTracker.notifySlotStatus(
+ new AllocationID(),
+ new JobID(),
+ new InstanceID(),
+ ResourceProfile.ANY,
+ SlotState.FREE);
+ })
+ .isInstanceOf(NullPointerException.class);
}
@Test
- public void testRecordPendingAllocations() {
+ void testRecordPendingAllocations() {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
final PendingTaskManager pendingTaskManager1 =
@@ -277,27 +286,24 @@ public class FineGrainedTaskManagerTrackerTest extends TestLogger {
pendingTaskManager2.getPendingTaskManagerId(),
Collections.singletonMap(jobId, resourceCounter)));
assertThat(
- taskManagerTracker
- .getPendingAllocationsOfPendingTaskManager(
- pendingTaskManager1.getPendingTaskManagerId())
- .size(),
- is(0));
- assertTrue(
- taskManagerTracker
- .getPendingAllocationsOfPendingTaskManager(
- pendingTaskManager2.getPendingTaskManagerId())
- .containsKey(jobId));
+ taskManagerTracker.getPendingAllocationsOfPendingTaskManager(
+ pendingTaskManager1.getPendingTaskManagerId()))
+ .isEmpty();
+ assertThat(
+ taskManagerTracker.getPendingAllocationsOfPendingTaskManager(
+ pendingTaskManager2.getPendingTaskManagerId()))
+ .containsKey(jobId);
assertThat(
- taskManagerTracker
- .getPendingAllocationsOfPendingTaskManager(
- pendingTaskManager2.getPendingTaskManagerId())
- .get(jobId)
- .getResourceCount(ResourceProfile.ANY),
- is(1));
+ taskManagerTracker
+ .getPendingAllocationsOfPendingTaskManager(
+ pendingTaskManager2.getPendingTaskManagerId())
+ .get(jobId)
+ .getResourceCount(ResourceProfile.ANY))
+ .isEqualTo(1);
}
@Test
- public void testGetStatistics() {
+ void testGetStatistics() {
final FineGrainedTaskManagerTracker taskManagerTracker =
new FineGrainedTaskManagerTracker();
final ResourceProfile totalResource = ResourceProfile.fromResources(10, 1000);
@@ -322,11 +328,12 @@ public class FineGrainedTaskManagerTrackerTest extends TestLogger {
taskManagerTracker.addPendingTaskManager(
new PendingTaskManager(ResourceProfile.fromResources(4, 200), 1));
- assertThat(taskManagerTracker.getFreeResource(), is(ResourceProfile.fromResources(6, 700)));
- assertThat(taskManagerTracker.getRegisteredResource(), is(totalResource));
- assertThat(taskManagerTracker.getNumberRegisteredSlots(), is(10));
- assertThat(taskManagerTracker.getNumberFreeSlots(), is(8));
- assertThat(
- taskManagerTracker.getPendingResource(), is(ResourceProfile.fromResources(4, 200)));
+ assertThat(taskManagerTracker.getFreeResource())
+ .isEqualTo(ResourceProfile.fromResources(6, 700));
+ assertThat(taskManagerTracker.getRegisteredResource()).isEqualTo(totalResource);
+ assertThat(taskManagerTracker.getNumberRegisteredSlots()).isEqualTo(10);
+ assertThat(taskManagerTracker.getNumberFreeSlots()).isEqualTo(8);
+ assertThat(taskManagerTracker.getPendingResource())
+ .isEqualTo(ResourceProfile.fromResources(4, 200));
}
}