KarmaGYZ commented on a change in pull request #11615: URL: https://github.com/apache/flink/pull/11615#discussion_r415641552
########## 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: I think it probably makes sense to add ``` private boolean isMaxSlotNumExceededAfterAdding(int numNewSlot); ``` WDYT? ---------------------------------------------------------------- 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: us...@infra.apache.org