private String getSourceHostForContainer(ContainerPlacementRequestMessage requestMessage) {
    String sourceHost = null;
    String processorId = requestMessage.getProcessorId();
    if (samzaApplicationState.runningProcessors.containsKey(processorId)) {
      SamzaResource currentResource = samzaApplicationState.runningProcessors.get(processorId);
      LOG.info("Processor ID: {} matched a running container with containerId ID: {} is running on host: {} for ContainerPlacement action: {}",
          processorId, currentResource.getContainerId(), currentResource.getHost(), requestMessage);
      sourceHost = currentResource.getHost();
    } else {
      sourceHost = Optional.ofNullable(localityManager.readLocality().getProcessorLocality(processorId))
          .map(ProcessorLocality::host)
          .orElse(null);
      LOG.info("Processor ID: {} is not running and was last seen on host: {} for ContainerPlacement action: {}",
          processorId, sourceHost, requestMessage);
    }
    return sourceHost;
  }
Gets the hostname on which container is either currently running or was last seen on if it is not running
TODO SAMZA-2480: Move logic related to onResourcesCompleted from ContainerProcessManager to ContainerManager
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerManager.java