zentol commented on a change in pull request #17059:
URL: https://github.com/apache/flink/pull/17059#discussion_r698580248
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclarativeSlotPool.java
##########
@@ -354,30 +369,49 @@ public void registerNewSlotsListener(NewSlotsListener
newSlotsListener) {
@Override
public ResourceCounter releaseSlots(ResourceID owner, Exception cause) {
- final Collection<AllocatedSlot> removedSlots =
slotPool.removeSlots(owner);
+ final AllocatedSlotPool.AllocatedSlotsAndReservationStatus
removedSlots =
+ slotPool.removeSlots(owner);
- ResourceCounter previouslyFulfilledRequirements =
getFulfilledRequirements(removedSlots);
-
- releasePayload(removedSlots, cause);
- releaseSlots(removedSlots, cause);
+ final Collection<AllocatedSlot> slotsToFree = new ArrayList<>();
+ for (AllocatedSlot removedSlot : removedSlots.getAllocatedSlots()) {
+ if (!removedSlots.wasFree(removedSlot)) {
+ slotsToFree.add(removedSlot);
+ }
+ }
- return previouslyFulfilledRequirements;
+ return freeAndReleaseSlots(slotsToFree,
removedSlots.getAllocatedSlots(), cause);
}
@Override
public ResourceCounter releaseSlot(AllocationID allocationId, Exception
cause) {
Review comment:
The behavior of these 2 methods should be identical, so some common
parts were extracted.
--
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]
