[FLINK-34434][slotmanager] Complete the returnedFuture when slot remo…

…ved before allocation success
bbejeck · Feb 20, 2024 · 1a494bc · 1a494bc
1 parent 15af3e4
commit 1a494bc
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 1 deletion.
diff --git a/...in/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java b/...in/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java
@@ -152,7 +152,7 @@ public CompletableFuture<Void> allocateSlot(
  LOG.debug(
  "The slot {} has been removed before. Ignore the future.",
  allocationId);
- requestFuture.complete(null);
+ returnedFuture.complete(null);
  return null;
  }
  if (acknowledge != null) {

diff --git a/...ava/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java b/...ava/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncerTest.java
@@ -119,6 +119,16 @@ void testAllocateSlotFailsWithException() {
  assertThat(taskManagerInfo.getAllocatedSlots()).isEmpty());
  }
 
+ @Test
+ void testAllocationUpdatesIgnoredIfSlotRemoved() throws Exception {
+ testSlotAllocation(
+ (slotStatusSyncer, taskManagerTracker, instanceID, allocationId) -> {
+ taskManagerTracker.removeTaskManager(instanceID);
+ assertThat(taskManagerTracker.getAllocatedOrPendingSlot(allocationId))
+ .isEmpty();
+ });
+ }
+
  @Test
  void testFreeSlot() {
  final FineGrainedTaskManagerTracker taskManagerTracker =