private void initiateStandbyAwareAllocation(String activeContainerID, String resourceID,
      ContainerAllocator containerAllocator) {

    String standbyHost = this.selectStandbyHost(activeContainerID, resourceID);

    // if the standbyHost returned is anyhost, we proceed with the request directly
    if (standbyHost.equals(ResourceRequestState.ANY_HOST)) {
      log.info("No standby container found for active container {}, making a resource-request for placing {} on {}, active's resourceID: {}",
          activeContainerID, activeContainerID, ResourceRequestState.ANY_HOST, resourceID);
      samzaApplicationState.failoversToAnyHost.incrementAndGet();
      containerAllocator.requestResource(activeContainerID, ResourceRequestState.ANY_HOST);

    } else {

      // Check if there is a running standby-container on that host that needs to be stopped
      List<String> standbySamzaContainerIds = this.standbyContainerConstraints.get(activeContainerID);

      Map<String, SamzaResource> runningStandbyContainersOnHost = new HashMap<>();
      this.samzaApplicationState.runningProcessors.forEach((samzaContainerId, samzaResource) -> {
        if (standbySamzaContainerIds.contains(samzaContainerId) && samzaResource.getHost().equals(standbyHost)) {
          runningStandbyContainersOnHost.put(samzaContainerId, samzaResource);
        }
      });

      if (runningStandbyContainersOnHost.isEmpty()) {
        // if there are no running standby-containers on the standbyHost, we proceed to directly make a resource request

        log.info("No running standby container to stop on host {}, making a resource-request for placing {} on {}, active's resourceID: {}",
            standbyHost, activeContainerID, standbyHost, resourceID);
        FailoverMetadata failoverMetadata = this.registerActiveContainerFailure(activeContainerID, resourceID);

        // record the resource request, before issuing it to avoid race with allocation-thread
        SamzaResourceRequest resourceRequestForActive =
                containerAllocator.getResourceRequest(activeContainerID, standbyHost);
        failoverMetadata.recordResourceRequest(resourceRequestForActive);
        containerAllocator.issueResourceRequest(resourceRequestForActive);
        samzaApplicationState.failoversToStandby.incrementAndGet();
      } else {
        // if there is a running standby-container on the standbyHost, we issue a stop (the stopComplete callback completes the remainder of the flow)
        FailoverMetadata failoverMetadata = this.registerActiveContainerFailure(activeContainerID, resourceID);

        runningStandbyContainersOnHost.forEach((standbyContainerID, standbyResource) -> {
          log.info("Initiating failover and stopping standby container, found standbyContainer {} = resource {}, "
                  + "for active container {}", runningStandbyContainersOnHost.keySet(),
              runningStandbyContainersOnHost.values(), activeContainerID);
          failoverMetadata.updateStandbyContainer(standbyResource.getContainerId(), standbyResource.getHost());
          samzaApplicationState.failoversToStandby.incrementAndGet();
          this.clusterResourceManager.stopStreamProcessor(standbyResource);
        });

        // if multiple standbys are on the same host, we are in an invalid state, so we fail the deploy and retry
        if (runningStandbyContainersOnHost.size() > 1) {
          throw new SamzaException(
              "Invalid State. Multiple standby containers found running on one host:" + runningStandbyContainersOnHost);
        }
      }
    }
  }
Method to handle standby-aware allocation for an active container.
 We try to find a standby host for the active container, and issue a stop on any standby-containers running on it,
 request resource to place the active on the standby's host, and one to place the standby elsewhere.
 When requesting for resources,
 NOTE: When rack awareness is turned on, we always pass the <code>hostToAvoid</> parameter as null for the {@link #requestResource} method used here
 because the hostname of the previous active processor that died does not exist in the running or pending container list anymore.
 However, different racks will always be guaranteed through {@link #checkStandbyConstraintsAndRunStreamProcessor}.
@param activeContainerID the samzaContainerID of the active-container
@param resourceID  the samza-resource-ID of the container when it failed (used to index failover-state)
samza-core/src/main/java/org/apache/samza/clustermanager/StandbyContainerManager.java