KarmaGYZ commented on a change in pull request #14897:
URL: https://github.com/apache/flink/pull/14897#discussion_r578870795
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java
##########
@@ -600,61 +603,69 @@ private void allocateSlot(
FutureUtils.assertNoException(slotAllocationResponseProcessingFuture);
}
- private void tryFulfillRequirementsWithPendingSlots(
+ private ResourceCounter tryFulfillRequirementsWithPendingSlots(
JobID jobId,
- Map<ResourceProfile, Integer> missingResources,
+ Collection<Map.Entry<ResourceProfile, Integer>> missingResources,
ResourceCounter pendingSlots) {
- for (Map.Entry<ResourceProfile, Integer> missingResource :
missingResources.entrySet()) {
+ for (Map.Entry<ResourceProfile, Integer> missingResource :
missingResources) {
ResourceProfile profile = missingResource.getKey();
for (int i = 0; i < missingResource.getValue(); i++) {
- if (!tryFulfillWithPendingSlots(profile, pendingSlots)) {
- boolean couldAllocateWorkerAndReserveSlot =
+ final Tuple2<Boolean, ResourceCounter> matchingResult =
+ tryFulfillWithPendingSlots(profile, pendingSlots);
+ pendingSlots = matchingResult.f1;
+ if (!matchingResult.f0) {
+ final Tuple2<Boolean, ResourceCounter> allocationResult =
tryAllocateWorkerAndReserveSlot(profile,
pendingSlots);
+ pendingSlots = allocationResult.f1;
+ boolean couldAllocateWorkerAndReserveSlot =
allocationResult.f0;
if (!couldAllocateWorkerAndReserveSlot &&
sendNotEnoughResourceNotifications) {
LOG.warn("Could not fulfill resource requirements of
job {}.", jobId);
resourceActions.notifyNotEnoughResourcesAvailable(
jobId,
resourceTracker.getAcquiredResources(jobId));
- return;
+ return pendingSlots;
}
}
}
}
+ return pendingSlots;
}
- private boolean tryFulfillWithPendingSlots(
+ private Tuple2<Boolean, ResourceCounter> tryFulfillWithPendingSlots(
Review comment:
That's a good point.
----------------------------------------------------------------
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
