@Test
public void testSubClusterExpiry() throws Exception {

  // Tests how the headroom info are used to split based on the capacity
  // each RM claims to give us.
  // Configure policy to be 100% headroom based
  getPolicyInfo().setHeadroomAlpha(1.0f);
  YarnConfiguration conf = new YarnConfiguration();
  // Set expiry to 500ms
  conf.setLong(YarnConfiguration.FEDERATION_AMRMPROXY_SUBCLUSTER_TIMEOUT,
      500);
  initializePolicy(conf);
  List<ResourceRequest> resourceRequests = createSimpleRequest();
  prepPolicyWithHeadroom(true);
  // For first time, no sub-cluster expired
  Set<SubClusterId> expiredSCList = new HashSet<>();
  Map<SubClusterId, List<ResourceRequest>> response =
      ((FederationAMRMProxyPolicy) getPolicy())
          .splitResourceRequests(resourceRequests, expiredSCList);
  // pretty print requests
  prettyPrintRequests(response);
  validateSplit(response, resourceRequests);
  /*
   * based on headroom, we expect 75 containers to got to subcluster0 (60) and
   * subcluster2 (15) according to the advertised headroom (40 and 10), no
   * containers for sublcuster1 as it advertise zero headroom, and 25 to
   * subcluster5 which has unknown headroom, and so it gets 1/4th of the load
   */
  checkExpectedAllocation(response, "subcluster0", 1, 60);
  checkExpectedAllocation(response, "subcluster1", 1, -1);
  checkExpectedAllocation(response, "subcluster2", 1, 15);
  checkExpectedAllocation(response, "subcluster5", 1, 25);
  checkTotalContainerAllocation(response, 100);
  Thread.sleep(800);
  // For the second time, sc0 and sc5 expired
  expiredSCList.add(SubClusterId.newInstance("subcluster0"));
  expiredSCList.add(SubClusterId.newInstance("subcluster5"));
  response = ((FederationAMRMProxyPolicy) getPolicy())
      .splitResourceRequests(resourceRequests, expiredSCList);
  // pretty print requests
  prettyPrintRequests(response);
  validateSplit(response, resourceRequests);
  checkExpectedAllocation(response, "subcluster0", 1, -1);
  checkExpectedAllocation(response, "subcluster1", 1, -1);
  checkExpectedAllocation(response, "subcluster2", 1, 100);
  checkExpectedAllocation(response, "subcluster5", 1, -1);
  checkTotalContainerAllocation(response, 100);
}