(flink) 02/02: [FLINK-34434][slotmanager] Complete the returnedFuture when slot removed before allocation success

[guoyangze]

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

guoyangze pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 21cfe998f4fb21afe24ceb8b6f4fef180e89b9e9
Author: Yangze Guo <karma...@gmail.com>
AuthorDate: Mon Feb 19 10:35:45 2024 +0800

    [FLINK-34434][slotmanager] Complete the returnedFuture when slot removed 
before allocation success
---
 .../resourcemanager/slotmanager/DefaultSlotStatusSyncer.java   |  2 +-
 .../slotmanager/DefaultSlotStatusSyncerTest.java               | 10 ++++++++++
 2 files changed, 11 insertions(+), 1 deletion(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java
index 0d687df9ad4..794db8125c2 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java
@@ -152,7 +152,7 @@ public class DefaultSlotStatusSyncer implements 
SlotStatusSyncer {
                                 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/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 4c3d2d5a723..ebce5c72b65 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
@@ -119,6 +119,16 @@ class DefaultSlotStatusSyncerTest {
                                 
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 =