SchedulerApplicationAttempt implements SchedulableEntity { @Private public boolean hasPendingResourceRequest(String nodePartition, SchedulingMode schedulingMode) { if (schedulingMode == SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY) { nodePartition = RMNodeLabelsManager.NO_LABEL; } Resource pending = attemptResourceUsage.getPending(nodePartition); if (StringUtils.equals(nodePartition, RMNodeLabelsManager.NO_LABEL)) { pending = Resources.subtractNonNegative(pending, Resources .createResource(unconfirmedAllocatedMem.get(), unconfirmedAllocatedVcores.get())); } return !Resources.isNone(pending); } SchedulerApplicationAttempt(ApplicationAttemptId applicationAttemptId, 
      String user, Queue queue, AbstractUsersManager abstractUsersManager,
      RMContext rmContext); void setOpportunisticContainerContext(
      OpportunisticContainerContext oppContext); OpportunisticContainerContext getOpportunisticContainerContext(); Collection<RMContainer> getLiveContainers(); AppSchedulingInfo getAppSchedulingInfo(); ContainerUpdateContext getUpdateContext(); boolean isPending(); ApplicationAttemptId getApplicationAttemptId(); ApplicationId getApplicationId(); String getUser(); Set<ContainerId> getPendingRelease(); long getNewContainerId(); Collection<SchedulerRequestKey> getSchedulerKeys(); PendingAsk getPendingAsk(
      SchedulerRequestKey schedulerKey, String resourceName); int getOutstandingAsksCount(SchedulerRequestKey schedulerKey); int getOutstandingAsksCount(SchedulerRequestKey schedulerKey,
      String resourceName); String getQueueName(); Resource getAMResource(); Resource getAMResource(String label); void setAMResource(Resource amResource); void setAMResource(String label, Resource amResource); boolean isAmRunning(); void setAmRunning(boolean bool); boolean getUnmanagedAM(); RMContainer getRMContainer(ContainerId id); void addRMContainer(
      ContainerId id, RMContainer rmContainer); void removeRMContainer(ContainerId containerId); int getReReservations(SchedulerRequestKey schedulerKey); @Stable @Private Resource getCurrentReservation(); Queue getQueue(); boolean updateResourceRequests(
      List<ResourceRequest> requests); boolean updateSchedulingRequests(
      List<SchedulingRequest> requests); void recoverResourceRequestsForContainer(
      ContainerRequest containerRequest); void stop(RMAppAttemptState rmAppAttemptFinalState); boolean isStopped(); List<RMContainer> getReservedContainers(); boolean reserveIncreasedContainer(SchedulerNode node,
      SchedulerRequestKey schedulerKey, RMContainer rmContainer,
      Resource reservedResource); RMContainer reserve(SchedulerNode node,
      SchedulerRequestKey schedulerKey, RMContainer rmContainer,
      Container container); void setHeadroom(Resource globalLimit); Resource getHeadroom(); int getNumReservedContainers(
      SchedulerRequestKey schedulerKey); @SuppressWarnings("unchecked") void containerLaunchedOnNode(ContainerId containerId,
      NodeId nodeId); void showRequests(); Resource getCurrentConsumption(); void updateNMTokens(Collection<Container> containers); List<Container> pullPreviousAttemptContainers(); List<Container> pullNewlyAllocatedContainers(); synchronized void addToNewlyDemotedContainers(ContainerId containerId,
      RMContainer rmContainer); synchronized void addToNewlyDecreasedContainers(
      ContainerId containerId, RMContainer rmContainer); List<Container> pullNewlyPromotedContainers(); List<Container> pullNewlyDemotedContainers(); List<Container> pullNewlyIncreasedContainers(); List<Container> pullNewlyDecreasedContainers(); List<UpdateContainerError> pullUpdateContainerErrors(); List<NMToken> pullUpdatedNMTokens(); boolean isWaitingForAMContainer(); void updateBlacklist(List<String> blacklistAdditions,
      List<String> blacklistRemovals); boolean isPlaceBlacklisted(String resourceName); int addMissedNonPartitionedRequestSchedulingOpportunity(
      SchedulerRequestKey schedulerKey); void resetMissedNonPartitionedRequestSchedulingOpportunity(
      SchedulerRequestKey schedulerKey); void addSchedulingOpportunity(
      SchedulerRequestKey schedulerKey); void subtractSchedulingOpportunity(
      SchedulerRequestKey schedulerKey); int getSchedulingOpportunities(
      SchedulerRequestKey schedulerKey); void resetSchedulingOpportunities(
      SchedulerRequestKey schedulerKey); void resetSchedulingOpportunities(SchedulerRequestKey schedulerKey,
      long currentTimeMs); ApplicationResourceUsageReport getResourceUsageReport(); @VisibleForTesting Map<ContainerId, RMContainer> getLiveContainersMap(); Map<SchedulerRequestKey, Long> getLastScheduledContainer(); void transferStateFromPreviousAttempt(
      SchedulerApplicationAttempt appAttempt); void move(Queue newQueue); void recoverContainer(SchedulerNode node,
      RMContainer rmContainer); void incNumAllocatedContainers(NodeType containerType,
      NodeType requestType); void setApplicationHeadroomForMetrics(Resource headroom); void recordContainerRequestTime(long value); void recordContainerAllocationTime(long value); Set<String> getBlacklistedNodes(); @Private boolean hasPendingResourceRequest(String nodePartition,
      SchedulingMode schedulingMode); @VisibleForTesting ResourceUsage getAppAttemptResourceUsage(); @Override Priority getPriority(); void setPriority(Priority appPriority); @Override String getId(); @Override int compareInputOrderTo(SchedulableEntity other); @Override ResourceUsage getSchedulingResourceUsage(); void setAppAMNodePartitionName(String partitionName); String getAppAMNodePartitionName(); void updateAMContainerDiagnostics(AMState state,
      String diagnosticMessage); ReentrantReadWriteLock.WriteLock getWriteLock(); @Override boolean isRecovering(); AppPlacementAllocator<N> getAppPlacementAllocator(
      SchedulerRequestKey schedulerRequestKey); void incUnconfirmedRes(Resource res); void decUnconfirmedRes(Resource res); @Override int hashCode(); @Override boolean equals(Object o); Map<String, String> getApplicationSchedulingEnvs();  }
@Test public void testHasPendingResourceRequest() throws Exception { ApplicationAttemptId attemptId = createAppAttemptId(0, 0); Queue queue = createQueue("test", null); RMContext rmContext = mock(RMContext.class); when(rmContext.getEpoch()).thenReturn(3L); SchedulerApplicationAttempt app = new SchedulerApplicationAttempt( attemptId, "user", queue, queue.getAbstractUsersManager(), rmContext); Priority priority = Priority.newInstance(1); List<ResourceRequest> requests = new ArrayList<>(2); Resource unit = Resource.newInstance(1L, 1); requests.add(ResourceRequest.newInstance(priority, ResourceRequest.ANY, unit, 1, false, "label1")); requests.add(ResourceRequest.newInstance(priority, ResourceRequest.ANY, unit, 1, false, "")); requests.get(0).setAllocationRequestId(0L); requests.get(1).setAllocationRequestId(1L); app.updateResourceRequests(requests); assertTrue("Reported no pending resource requests for no label when " + "resource requests for no label are pending (exclusive partitions)", app.hasPendingResourceRequest("", SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY)); assertTrue("Reported no pending resource requests for label with pending " + "resource requests (exclusive partitions)", app.hasPendingResourceRequest("label1", SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY)); assertFalse("Reported pending resource requests for label with no pending " + "resource requests (exclusive partitions)", app.hasPendingResourceRequest("label2", SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY)); assertTrue("Reported no pending resource requests for no label when " + "resource requests for no label are pending (relaxed partitions)", app.hasPendingResourceRequest("", SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY)); assertTrue("Reported no pending resource requests for label with pending " + "resource requests (relaxed partitions)", app.hasPendingResourceRequest("label1", SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY)); assertTrue("Reported no pending resource requests for label with no " + "pending resource requests (relaxed partitions)", app.hasPendingResourceRequest("label2", SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY)); }