XComp commented on code in PR #24325:
URL: https://github.com/apache/flink/pull/24325#discussion_r1494179385
##########
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java:
##########
@@ -216,6 +216,65 @@ void testAllocateSlotFailsWithException() {
assertThat(taskManagerInfo.getAllocatedSlots()).isEmpty());
}
+ @Test
+ void testAllocationUpdatesIgnoredIfSlotRemoved() throws Exception {
+ final FineGrainedTaskManagerTracker taskManagerTracker =
+ new FineGrainedTaskManagerTracker();
+ final CompletableFuture<
+ Tuple6<
+ SlotID,
+ JobID,
+ AllocationID,
+ ResourceProfile,
+ String,
+ ResourceManagerId>>
+ requestFuture = new CompletableFuture<>();
+ final CompletableFuture<Acknowledge> responseFuture = new
CompletableFuture<>();
+ final TestingTaskExecutorGateway taskExecutorGateway =
+ new TestingTaskExecutorGatewayBuilder()
+ .setRequestSlotFunction(
+ tuple6 -> {
+ requestFuture.complete(tuple6);
+ return responseFuture;
+ })
+ .createTestingTaskExecutorGateway();
+ final TaskExecutorConnection taskExecutorConnection =
+ new TaskExecutorConnection(ResourceID.generate(),
taskExecutorGateway);
+ taskManagerTracker.addTaskManager(
+ taskExecutorConnection, ResourceProfile.ANY,
ResourceProfile.ANY);
+ final ResourceTracker resourceTracker = new DefaultResourceTracker();
+ final JobID jobId = new JobID();
+ final SlotStatusSyncer slotStatusSyncer =
+ new DefaultSlotStatusSyncer(TASK_MANAGER_REQUEST_TIMEOUT);
+ slotStatusSyncer.initialize(
+ taskManagerTracker,
+ resourceTracker,
+ ResourceManagerId.generate(),
+ EXECUTOR_RESOURCE.getExecutor());
+
+ final CompletableFuture<Void> allocatedFuture =
+ slotStatusSyncer.allocateSlot(
+ taskExecutorConnection.getInstanceID(),
+ jobId,
+ "address",
+ ResourceProfile.ANY);
+ final AllocationID allocationId = requestFuture.get().f2;
+ 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);
+ });
+
+
taskManagerTracker.removeTaskManager(taskExecutorConnection.getInstanceID());
+
assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId)).isEmpty();
+
+ responseFuture.complete(Acknowledge.get());
+ assertThat(allocatedFuture).isNotCompletedExceptionally();
Review Comment:
```suggestion
assertThatFuture(allocatedFuture).eventuallySucceeds();
```
The test is succeeding even with the production code change not being
applied. We're getting into the right code path for this test, but we're not
asserting the returned future correctly as far as I can see.
(hint: You have to statically import `FlinkAssertions.assertThatFuture` for
the above code change to work)
##########
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java:
##########
@@ -216,6 +216,65 @@ void testAllocateSlotFailsWithException() {
assertThat(taskManagerInfo.getAllocatedSlots()).isEmpty());
}
+ @Test
+ void testAllocationUpdatesIgnoredIfSlotRemoved() throws Exception {
+ final FineGrainedTaskManagerTracker taskManagerTracker =
+ new FineGrainedTaskManagerTracker();
+ final CompletableFuture<
+ Tuple6<
+ SlotID,
+ JobID,
+ AllocationID,
+ ResourceProfile,
+ String,
+ ResourceManagerId>>
+ requestFuture = new CompletableFuture<>();
Review Comment:
```suggestion
final CompletableFuture<AllocationID> requestSlotReceivedFuture =
new CompletableFuture<>();
```
nit: We're only using `AllocationID` in the test later on. Hence, there's no
need to expose the entire `requestSlot` signature through a `Tuple6`
declaration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
