[hotfix][runtime] Minor cleanup in SlotManager

Avoid checking whether slots would be fulfillable unless necessary
tongcheng-elong · Jul 10, 2019 · be794ac · be794ac
1 parent caf89b1
commit be794ac
Showing 1 changed file with 7 additions and 5 deletions.
diff --git a/...ntime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java b/...ntime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java
@@ -758,13 +758,15 @@ private void internalRequestSlot(PendingSlotRequest pendingSlotRequest) throws R
  pendingTaskManagerSlotOptional = allocateResource(resourceProfile);
  }
 
- pendingTaskManagerSlotOptional.ifPresent(pendingTaskManagerSlot -> assignPendingTaskManagerSlot(pendingSlotRequest, pendingTaskManagerSlot));
- if (!pendingTaskManagerSlotOptional.isPresent()) {
+ if (pendingTaskManagerSlotOptional.isPresent()) {
+ assignPendingTaskManagerSlot(pendingSlotRequest, pendingTaskManagerSlotOptional.get());
+ }
+ else {
  // request can not be fulfilled by any free slot or pending slot that can be allocated,
  // check whether it can be fulfilled by allocated slots
- boolean fulfillable = isFulfillableByRegisteredSlots(pendingSlotRequest.getResourceProfile());
- if (!fulfillable && failUnfulfillableRequest) {
- throw new ResourceManagerException("Requested resource profile (" + pendingSlotRequest.getResourceProfile() + ") is unfulfillable.");
+ if (failUnfulfillableRequest && !isFulfillableByRegisteredSlots(pendingSlotRequest.getResourceProfile())) {
+  throw new ResourceManagerException("Requested resource profile (" +
+  pendingSlotRequest.getResourceProfile() + ") is unfulfillable.");
  }
  }
  }