You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2019/07/10 09:36:16 UTC
[flink] 15/16: [FLINK-12763][runtime] Remove eagerly rejected
pending slot requests from SlotManager
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit caf89b18ad351964af4bf6999dfb21ee3887fc55
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Tue Jul 9 19:24:17 2019 +0200
[FLINK-12763][runtime] Remove eagerly rejected pending slot requests from SlotManager
---
.../apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java | 1 +
.../resourcemanager/slotmanager/SlotManagerFailUnfulfillableTest.java | 2 +-
2 files changed, 2 insertions(+), 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java
index 977c3f4..bea588b 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java
@@ -488,6 +488,7 @@ public class SlotManager implements AutoCloseable {
continue;
}
if (!isFulfillableByRegisteredSlots(pendingSlotRequest.getResourceProfile())) {
+ slotRequestIterator.remove();
resourceActions.notifyAllocationFailure(
pendingSlotRequest.getJobId(),
pendingSlotRequest.getAllocationId(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerFailUnfulfillableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerFailUnfulfillableTest.java
index 73a081f..cda07fe 100755
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerFailUnfulfillableTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerFailUnfulfillableTest.java
@@ -84,7 +84,7 @@ public class SlotManagerFailUnfulfillableTest extends TestLogger {
// assert
assertEquals(1, allocationFailures.size());
assertEquals(request.getAllocationId(), allocationFailures.get(0));
-// assertEquals(0, slotManager.getNumberPendingSlotRequests()); // BUG, to be fixed in follow-up commit
+ assertEquals(0, slotManager.getNumberPendingSlotRequests());
}
@Test