@Override
  public void onContainersAllocated(List<Container> containers) {
    List<SamzaResource> resources = new ArrayList<SamzaResource>();
    for (Container container : containers) {
      log.info("Got allocation notification for Container ID: {} on host: {}", container.getId(),
          container.getNodeId().getHost());
      String containerId = container.getId().toString();
      String host = container.getNodeId().getHost();
      int memory = container.getResource().getMemory();
      int numCores = container.getResource().getVirtualCores();

      SamzaResource resource = new SamzaResource(numCores, memory, host, containerId);
      allocatedResources.put(resource, container);
      resources.add(resource);
    }
    clusterManagerCallback.onResourcesAvailable(resources);
  }
Callback invoked from Yarn when containers are allocated. This translates the yarn callbacks into Samza
specific ones.
@param containers the list of {@link Container} returned by Yarn.
samza-yarn3/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java