private static Map<String, LocationId> getProcessorLocality(Config config, LocalityManager localityManager) {
    Map<String, LocationId> containerToLocationId = new HashMap<>();
    Map<String, ProcessorLocality> existingContainerLocality = localityManager.readLocality().getProcessorLocalities();

    for (int i = 0; i < new JobConfig(config).getContainerCount(); i++) {
      String containerId = Integer.toString(i);
      LocationId locationId = Optional.ofNullable(existingContainerLocality.get(containerId))
          .map(ProcessorLocality::host)
          .filter(StringUtils::isNotEmpty)
          .map(LocationId::new)
          // To handle the case when the container count is increased between two different runs of a samza-yarn job,
          // set the locality of newly added containers to any_host.
          .orElse(new LocationId("ANY_HOST"));
      containerToLocationId.put(containerId, locationId);
    }
    return containerToLocationId;
  }
Retrieves and returns the processor locality of a samza job using provided {@see Config} and {@see LocalityManager}.
@param config provides the configurations defined by the user. Required to connect to the storage layer.
@param localityManager provides the processor to host mapping persisted to the metadata store.
@return the processor locality.
samza-core/src/main/java/org/apache/samza/coordinator/JobModelHelper.java