@Test(timeout = 10000) public void notEnoughHosts() throws Exception { int numHosts = 10; int maxContainersPerHost = 5; int numContainers = numHosts * maxContainersPerHost + 1; ContainerAllocator containerAllocator = setup(numHosts, maxContainersPerHost, numContainers); mThrown.expect(RuntimeException.class); mThrown.expectMessage( ExceptionMessage.YARN_NOT_ENOUGH_HOSTS.getMessage(numContainers, CONTAINER_NAME, numHosts)); containerAllocator.allocateContainers(); }
public List<Container> allocateContainers() throws Exception { for (int attempt = 0; attempt < MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS; attempt++) { LOG.debug("Attempt {} of {} to allocate containers", attempt, MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS); int numContainersToRequest = mTargetNumContainers - mAllocatedContainerHosts.size(); LOG.debug("Requesting {} containers", numContainersToRequest); mOutstandingContainerRequestsLatch = new CountDownLatch(numContainersToRequest); requestContainers(); mOutstandingContainerRequestsLatch.await(); if (mAllocatedContainerHosts.size() == mTargetNumContainers) { break; } } if (mAllocatedContainers.size() != mTargetNumContainers) { throw new RuntimeException(String.format("Failed to allocate %d %s containers", mTargetNumContainers, mContainerName)); } return mAllocatedContainers; }
ContainerAllocator { public List<Container> allocateContainers() throws Exception { for (int attempt = 0; attempt < MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS; attempt++) { LOG.debug("Attempt {} of {} to allocate containers", attempt, MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS); int numContainersToRequest = mTargetNumContainers - mAllocatedContainerHosts.size(); LOG.debug("Requesting {} containers", numContainersToRequest); mOutstandingContainerRequestsLatch = new CountDownLatch(numContainersToRequest); requestContainers(); mOutstandingContainerRequestsLatch.await(); if (mAllocatedContainerHosts.size() == mTargetNumContainers) { break; } } if (mAllocatedContainers.size() != mTargetNumContainers) { throw new RuntimeException(String.format("Failed to allocate %d %s containers", mTargetNumContainers, mContainerName)); } return mAllocatedContainers; } }
ContainerAllocator { public List<Container> allocateContainers() throws Exception { for (int attempt = 0; attempt < MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS; attempt++) { LOG.debug("Attempt {} of {} to allocate containers", attempt, MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS); int numContainersToRequest = mTargetNumContainers - mAllocatedContainerHosts.size(); LOG.debug("Requesting {} containers", numContainersToRequest); mOutstandingContainerRequestsLatch = new CountDownLatch(numContainersToRequest); requestContainers(); mOutstandingContainerRequestsLatch.await(); if (mAllocatedContainerHosts.size() == mTargetNumContainers) { break; } } if (mAllocatedContainers.size() != mTargetNumContainers) { throw new RuntimeException(String.format("Failed to allocate %d %s containers", mTargetNumContainers, mContainerName)); } return mAllocatedContainers; } ContainerAllocator(String containerName, int targetNumContainers, int maxContainersPerHost,
      Resource resource, YarnClient yarnClient, AMRMClientAsync<ContainerRequest> rmClient);  ContainerAllocator(String containerName, int targetNumContainers, int maxContainersPerHost,
      Resource resource, YarnClient yarnClient,
      AMRMClientAsync<ContainerRequest> rmClient, String preferredHost); }
ContainerAllocator { public List<Container> allocateContainers() throws Exception { for (int attempt = 0; attempt < MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS; attempt++) { LOG.debug("Attempt {} of {} to allocate containers", attempt, MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS); int numContainersToRequest = mTargetNumContainers - mAllocatedContainerHosts.size(); LOG.debug("Requesting {} containers", numContainersToRequest); mOutstandingContainerRequestsLatch = new CountDownLatch(numContainersToRequest); requestContainers(); mOutstandingContainerRequestsLatch.await(); if (mAllocatedContainerHosts.size() == mTargetNumContainers) { break; } } if (mAllocatedContainers.size() != mTargetNumContainers) { throw new RuntimeException(String.format("Failed to allocate %d %s containers", mTargetNumContainers, mContainerName)); } return mAllocatedContainers; } ContainerAllocator(String containerName, int targetNumContainers, int maxContainersPerHost,
      Resource resource, YarnClient yarnClient, AMRMClientAsync<ContainerRequest> rmClient);  ContainerAllocator(String containerName, int targetNumContainers, int maxContainersPerHost,
      Resource resource, YarnClient yarnClient,
      AMRMClientAsync<ContainerRequest> rmClient, String preferredHost); List<Container> allocateContainers(); synchronized void allocateContainer(Container container); }
ContainerAllocator { public List<Container> allocateContainers() throws Exception { for (int attempt = 0; attempt < MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS; attempt++) { LOG.debug("Attempt {} of {} to allocate containers", attempt, MAX_WORKER_CONTAINER_REQUEST_ATTEMPTS); int numContainersToRequest = mTargetNumContainers - mAllocatedContainerHosts.size(); LOG.debug("Requesting {} containers", numContainersToRequest); mOutstandingContainerRequestsLatch = new CountDownLatch(numContainersToRequest); requestContainers(); mOutstandingContainerRequestsLatch.await(); if (mAllocatedContainerHosts.size() == mTargetNumContainers) { break; } } if (mAllocatedContainers.size() != mTargetNumContainers) { throw new RuntimeException(String.format("Failed to allocate %d %s containers", mTargetNumContainers, mContainerName)); } return mAllocatedContainers; } ContainerAllocator(String containerName, int targetNumContainers, int maxContainersPerHost,
      Resource resource, YarnClient yarnClient, AMRMClientAsync<ContainerRequest> rmClient);  ContainerAllocator(String containerName, int targetNumContainers, int maxContainersPerHost,
      Resource resource, YarnClient yarnClient,
      AMRMClientAsync<ContainerRequest> rmClient, String preferredHost); List<Container> allocateContainers(); synchronized void allocateContainer(Container container);  }