import org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.ProportionalCapacityPreemptionPolicy; //导入依赖的package包/类
@Test(timeout = 10000)
public void testRMStarts() throws IOException {
  Configuration conf = new YarnConfiguration();
  conf.setBoolean(YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, true);
  conf.set(YarnConfiguration.RM_SCHEDULER_MONITOR_POLICIES,
      ProportionalCapacityPreemptionPolicy.class.getCanonicalName());

  RMStorageFactory.setConfiguration(conf);
  YarnAPIStorageFactory.setConfiguration(conf);
  DBUtility.InitializeDB();
  
  ResourceManager rm = new MockRM();
  try {
    rm.init(conf);
  } catch (Exception e) {
    fail("ResourceManager does not start when " +
        YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS + " is set to true");
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.ProportionalCapacityPreemptionPolicy; //导入依赖的package包/类
@Test(timeout = 10000)
public void testRMStarts() {
  Configuration conf = new YarnConfiguration();
  conf.setBoolean(YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, true);
  conf.set(YarnConfiguration.RM_SCHEDULER_MONITOR_POLICIES,
      ProportionalCapacityPreemptionPolicy.class.getCanonicalName());

  ResourceManager rm = new ResourceManager();
  try {
    rm.init(conf);
  } catch (Exception e) {
    fail("ResourceManager does not start when " +
        YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS + " is set to true");
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.ProportionalCapacityPreemptionPolicy; //导入依赖的package包/类
@Test(timeout = 10000)
public void testRMStarts() {
  Configuration conf = new YarnConfiguration();
  conf.setBoolean(YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, true);
  conf.set(YarnConfiguration.RM_SCHEDULER_MONITOR_POLICIES,
      ProportionalCapacityPreemptionPolicy.class.getCanonicalName());

  ResourceManager rm = new MockRM();
  try {
    rm.init(conf);
  } catch (Exception e) {
    fail("ResourceManager does not start when " +
        YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS + " is set to true");
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.ProportionalCapacityPreemptionPolicy; //导入依赖的package包/类
@Before
void setUp() throws Exception {
  conf = new YarnConfiguration();
  conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
      ResourceScheduler.class);
  conf.setBoolean(YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS, true);
  conf.setClass(YarnConfiguration.RM_SCHEDULER_MONITOR_POLICIES,
      ProportionalCapacityPreemptionPolicy.class, SchedulingEditPolicy.class);
  conf = TestUtils.getConfigurationWithMultipleQueues(this.conf);

  // Set preemption related configurations
  conf.setInt(CapacitySchedulerConfiguration.PREEMPTION_WAIT_TIME_BEFORE_KILL,
      0);
  conf.setFloat(CapacitySchedulerConfiguration.TOTAL_PREEMPTION_PER_ROUND,
      1.0f);
  conf.setFloat(
      CapacitySchedulerConfiguration.PREEMPTION_NATURAL_TERMINATION_FACTOR,
      1.0f);
  conf.setLong(CapacitySchedulerConfiguration.PREEMPTION_MONITORING_INTERVAL,
      60000L);
  RMStorageFactory.setConfiguration(conf);
  YarnAPIStorageFactory.setConfiguration(conf);
  DBUtility.InitializeDB();
  mgr = new NullRMNodeLabelsManager();
  mgr.init(this.conf);
  clock = mock(Clock.class);
  when(clock.getTime()).thenReturn(0L);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.ProportionalCapacityPreemptionPolicy; //导入依赖的package包/类
@Ignore
@Test (timeout = 60000)
public void testPreemptionPolicyCleanupKillableContainersWhenNoPreemptionNeeded()
    throws Exception {
  /**
   * Test case:
   * <pre>
   *             Root
   *            /  |  \
   *           a   b   c
   *          10   20  70
   * </pre>
   * Submit applications to two queues, one uses more than the other, so
   * preemption will happen.
   *
   * Check:
   * 1) Containers will be marked to killable
   * 2) Cancel resource request
   * 3) Killable containers will be cancelled from policy and scheduler
   */
  MockRM rm1 = new MockRM(conf);
  rm1.getRMContext().setNodeLabelManager(mgr);
  rm1.start();
  MockNM nm1 = rm1.registerNode("h1:1234", 8 * GB);
  CapacityScheduler cs = (CapacityScheduler) rm1.getResourceScheduler();
  RMNode rmNode1 = rm1.getRMContext().getRMNodes().get(nm1.getNodeId());

  // launch an app to queue, AM container should be launched in nm1
  RMApp app1 = rm1.submitApp(1 * GB, "app", "user", null, "a");
  MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);

  am1.allocate("*", 1 * GB, 6, new ArrayList<ContainerId>());

  // Do allocation 6 times for node1
  for (int i = 0; i < 6; i++) {
    cs.handle(new NodeUpdateSchedulerEvent(rmNode1));
  }

  // App1 should have 7 containers now, and no available resource for cluster
  FiCaSchedulerApp schedulerApp1 = cs.getApplicationAttempt(
      am1.getApplicationAttemptId());
  Assert.assertEquals(7, schedulerApp1.getLiveContainers().size());

  // Submit app2 to queue-c and asks for a 1G container for AM
  RMApp app2 = rm1.submitApp(1 * GB, "app", "user", null, "c");
  MockAM am2 = MockRM.launchAndRegisterAM(app2, rm1, nm1);

  // NM1 has available resource = 0G
  Assert.assertEquals(0 * GB, cs.getNode(nm1.getNodeId())
      .getAvailableResource().getMemorySize());
  am2.allocate("*", 3 * GB, 1, new ArrayList<ContainerId>());

  // Get edit policy and do one update
  ProportionalCapacityPreemptionPolicy editPolicy =
      (ProportionalCapacityPreemptionPolicy) getSchedulingEditPolicy(rm1);

  // Call edit schedule twice, and check if 3 container from app1 marked
  // to be "killable"
  editPolicy.editSchedule();
  editPolicy.editSchedule();

  PreemptionManager pm = cs.getPreemptionManager();
  waitKillableContainersSize(pm, "a", RMNodeLabelsManager.NO_LABEL, 3);

  // Change reqeust from 3G to 2G, now we can preempt one less container. (3->2)
  am2.allocate("*", 2 * GB, 1, new ArrayList<ContainerId>());
  editPolicy.editSchedule();
  Assert.assertEquals(0, editPolicy.getToPreemptContainers().size());
  waitKillableContainersSize(pm, "a", RMNodeLabelsManager.NO_LABEL, 2);

  // Call editSchedule once more to make sure still nothing happens
  editPolicy.editSchedule();
  Assert.assertEquals(0, editPolicy.getToPreemptContainers().size());
  waitKillableContainersSize(pm, "a", RMNodeLabelsManager.NO_LABEL, 2);
}