azagrebin commented on a change in pull request #11615:
URL: https://github.com/apache/flink/pull/11615#discussion_r415743176
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerImpl.java
##########
@@ -808,16 +850,31 @@ private void
fulfillPendingSlotRequestWithPendingTaskManagerSlot(PendingSlotRequ
return Optional.empty();
}
- private boolean isFulfillableByRegisteredSlots(ResourceProfile
resourceProfile) {
+ private boolean isFulfillableByRegisteredOrPendingSlots(ResourceProfile
resourceProfile) {
for (TaskManagerSlot slot : slots.values()) {
if
(slot.getResourceProfile().isMatching(resourceProfile)) {
return true;
}
}
+
+ for (PendingTaskManagerSlot slot : pendingSlots.values()) {
+ if
(slot.getResourceProfile().isMatching(resourceProfile)) {
+ return true;
+ }
+ }
+
return false;
}
private Optional<PendingTaskManagerSlot>
allocateResource(ResourceProfile requestedSlotResourceProfile) {
+ final int numRegisteredSlots = getNumberRegisteredSlots();
+ final int numPendingSlots = getNumberPendingTaskManagerSlots();
+ if (numPendingSlots + numRegisteredSlots + numSlotsPerWorker >
maxSlotNum) {
+ LOG.warn("Could not allocate {} more slots. The number
of registered and pending slots is {}, while the maximum is {}.",
+ numSlotsPerWorker, numPendingSlots +
numRegisteredSlots, maxSlotNum);
+ return Optional.empty();
+ }
Review comment:
do you mean to add just `isMaxSlotNumExceededAfterAdding` without
deduplication with `isMaxSlotNumExceededAfterRegistration`? because
`isMaxSlotNumExceededAfterRegistration` has 2 checks against 2 different
`numNewSlot`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
