You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/02/24 18:17:49 UTC
[7/9] flink git commit: [hotfix] Cancel pending slot request when
SlotPool is suspended
[hotfix] Cancel pending slot request when SlotPool is suspended
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/529b5128
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/529b5128
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/529b5128
Branch: refs/heads/master
Commit: 529b5128e9f6a7b8f368d6c22b90885667579352
Parents: 4fc3bcf
Author: Till Rohrmann <tr...@apache.org>
Authored: Thu Feb 22 15:37:37 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sat Feb 24 15:04:56 2018 +0100
----------------------------------------------------------------------
.../apache/flink/runtime/jobmaster/slotpool/SlotPool.java | 8 ++++++++
1 file changed, 8 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/529b5128/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
index ea816b5..409f8f7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
@@ -227,6 +227,14 @@ public class SlotPool extends RpcEndpoint implements SlotPoolGateway, AllocatedS
validateRunsInMainThread();
+ // cancel all pending allocations --> we can request these slots
+ // again after we regained the leadership
+ Set<AllocationID> allocationIds = pendingRequests.keySetB();
+
+ for (AllocationID allocationId : allocationIds) {
+ resourceManagerGateway.cancelSlotRequest(allocationId);
+ }
+
// suspend this RPC endpoint
stop();