private void handleNewContainer(final Container container) {

    LOG.log(Level.FINE, "allocated container: id[ {0} ]", container.getId());

    synchronized (this) {

      if (!matchContainerWithPendingRequest(container)) {
        LOG.log(Level.WARNING, "Got an extra container {0} that doesn't match, releasing...", container.getId());
        this.resourceManager.releaseAssignedContainer(container.getId());
        return;
      }

      final AMRMClient.ContainerRequest matchedRequest = this.requestsAfterSentToRM.peek();

      this.containerRequestCounter.decrement();
      this.containers.add(container);

      LOG.log(Level.FINEST, "{0} matched with {1}", new Object[] {container, matchedRequest});

      // Due to the bug YARN-314 and the workings of AMRMCClientAsync, when x-priority m-capacity zero-container
      // request and x-priority n-capacity nonzero-container request are sent together, where m > n, RM ignores
      // the latter.
      // Therefore it is necessary avoid sending zero-container request, even if it means getting extra containers.
      // It is okay to send nonzero m-capacity and n-capacity request together since bigger containers
      // can be matched.
      // TODO[JIRA REEF-42, REEF-942]: revisit this when implementing locality-strictness.
      // (i.e. a specific rack request can be ignored)
      if (this.requestsAfterSentToRM.size() > 1) {
        try {
          this.resourceManager.removeContainerRequest(matchedRequest);
        } catch (final Exception e) {
          LOG.log(Level.WARNING, "Error removing request from Async AMRM client queue: " + matchedRequest, e);
        }
      }

      this.requestsAfterSentToRM.remove();
      this.doHomogeneousRequests();

      LOG.log(Level.FINEST, "Allocated Container: memory = {0}, core number = {1}",
          new Object[] {container.getResource().getMemory(), container.getResource().getVirtualCores()});

      this.reefEventHandlers.onResourceAllocation(ResourceEventImpl.newAllocationBuilder()
          .setIdentifier(container.getId().toString())
          .setNodeId(container.getNodeId().toString())
          .setResourceMemory(container.getResource().getMemory())
          .setVirtualCores(container.getResource().getVirtualCores())
          .setRackName(rackNameFormatter.getRackName(container))
          .setRuntimeName(RuntimeIdentifier.RUNTIME_NAME)
          .build());

      this.updateRuntimeStatus();
    }
  }
private void handleNewContainer ( final Container container ) { LOG . log ( Level . FINE , "allocated container: id[ {0} ]" , container . getId ( ) ) ; synchronized ( this ) { if ( ! matchContainerWithPendingRequest ( container ) ) { LOG . log ( Level . WARNING , "Got an extra container {0} that doesn't match, releasing..." , container . getId ( ) ) ; this . resourceManager . releaseAssignedContainer ( container . getId ( ) ) ; return ; } final AMRMClient . ContainerRequest matchedRequest = this . requestsAfterSentToRM . peek ( ) ; this . containerRequestCounter . decrement ( ) ; this . containers . add ( container ) ; LOG . log ( Level . FINEST , "{0} matched with {1}" , new Object [ ] { container , matchedRequest } ) ; if ( this . requestsAfterSentToRM . size ( ) > 1 ) { try { this . resourceManager . removeContainerRequest ( matchedRequest ) ; } catch ( final Exception e ) { LOG . log ( Level . WARNING , "Error removing request from Async AMRM client queue: " + matchedRequest , e ) ; } } this . requestsAfterSentToRM . remove ( ) ; this . doHomogeneousRequests ( ) ; LOG . log ( Level . FINEST , "Allocated Container: memory = {0}, core number = {1}" , new Object [ ] { container . getResource ( ) . getMemory ( ) , container . getResource ( ) . getVirtualCores ( ) } ) ; this . reefEventHandlers . onResourceAllocation ( ResourceEventImpl . newAllocationBuilder ( ) . setIdentifier ( container . getId ( ) . toString ( ) ) . setNodeId ( container . getNodeId ( ) . toString ( ) ) . setResourceMemory ( container . getResource ( ) . getMemory ( ) ) . setVirtualCores ( container . getResource ( ) . getVirtualCores ( ) ) . setRackName ( rackNameFormatter . getRackName ( container ) ) . setRuntimeName ( RuntimeIdentifier . RUNTIME_NAME ) . build ( ) ) ; this . updateRuntimeStatus ( ) ; } }
Handles new container allocations. Calls come from YARN.
@param container newly allocated YARN container.