@Test(expected = YarnRuntimeException.class) public void testCreateDirsWithFileSystemNotBecomingAvailBeforeTimeout() throws Exception { dfsCluster.getFileSystem().setSafeMode( HdfsConstants.SafeModeAction.SAFEMODE_ENTER); Assert.assertTrue(dfsCluster.getFileSystem().isInSafeMode()); final ControlledClock clock = new ControlledClock(new SystemClock()); clock.setTime(1); new Thread() { @Override public void run() { try { Thread.sleep(500); clock.setTime(3000); } catch (Exception ex) { Assert.fail(ex.toString()); } } }.start(); testCreateHistoryDirs(dfsCluster.getConfiguration(0), clock); }
public void verifySlotMillis(int mapMemMb, int reduceMemMb, int minContainerSize) throws Exception { Clock actualClock = new SystemClock(); ControlledClock clock = new ControlledClock(actualClock); clock.setTime(10); MRApp app = new MRApp(1, 1, false, "testSlotMillisCounterUpdate", true, clock); Configuration conf = new Configuration(); conf.setInt(MRJobConfig.MAP_MEMORY_MB, mapMemMb); conf.setInt(MRJobConfig.REDUCE_MEMORY_MB, reduceMemMb); conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, minContainerSize); app.setClusterInfo(new ClusterInfo(Resource.newInstance(10240, 1))); Job job = app.submit(conf); app.waitForState(job, JobState.RUNNING); Map<TaskId, Task> tasks = job.getTasks(); Assert.assertEquals("Num tasks is not correct", 2, tasks.size()); Iterator<Task> taskIter = tasks.values().iterator(); Task mTask = taskIter.next(); app.waitForState(mTask, TaskState.RUNNING); Task rTask = taskIter.next(); app.waitForState(rTask, TaskState.RUNNING); Map<TaskAttemptId, TaskAttempt> mAttempts = mTask.getAttempts(); Assert.assertEquals("Num attempts is not correct", 1, mAttempts.size()); Map<TaskAttemptId, TaskAttempt> rAttempts = rTask.getAttempts(); Assert.assertEquals("Num attempts is not correct", 1, rAttempts.size()); TaskAttempt mta = mAttempts.values().iterator().next(); TaskAttempt rta = rAttempts.values().iterator().next(); app.waitForState(mta, TaskAttemptState.RUNNING); app.waitForState(rta, TaskAttemptState.RUNNING); clock.setTime(11); app.getContext() .getEventHandler() .handle(new TaskAttemptEvent(mta.getID(), TaskAttemptEventType.TA_DONE)); app.getContext() .getEventHandler() .handle(new TaskAttemptEvent(rta.getID(), TaskAttemptEventType.TA_DONE)); app.waitForState(job, JobState.SUCCEEDED); Assert.assertEquals(mta.getFinishTime(), 11); Assert.assertEquals(mta.getLaunchTime(), 10); Assert.assertEquals(rta.getFinishTime(), 11); Assert.assertEquals(rta.getLaunchTime(), 10); Assert.assertEquals((int) Math.ceil((float) mapMemMb / minContainerSize), job.getAllCounters().findCounter(JobCounter.SLOTS_MILLIS_MAPS) .getValue()); Assert.assertEquals( (int) Math.ceil((float) reduceMemMb / minContainerSize), job .getAllCounters().findCounter(JobCounter.SLOTS_MILLIS_REDUCES) .getValue()); }
public void verifyMillisCounters(int mapMemMb, int reduceMemMb, int minContainerSize) throws Exception { Clock actualClock = new SystemClock(); ControlledClock clock = new ControlledClock(actualClock); clock.setTime(10); MRApp app = new MRApp(1, 1, false, "testSlotMillisCounterUpdate", true, clock); Configuration conf = new Configuration(); conf.setInt(MRJobConfig.MAP_MEMORY_MB, mapMemMb); conf.setInt(MRJobConfig.REDUCE_MEMORY_MB, reduceMemMb); conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, minContainerSize); app.setClusterInfo(new ClusterInfo(Resource.newInstance(10240, 1))); Job job = app.submit(conf); app.waitForState(job, JobState.RUNNING); Map<TaskId, Task> tasks = job.getTasks(); Assert.assertEquals("Num tasks is not correct", 2, tasks.size()); Iterator<Task> taskIter = tasks.values().iterator(); Task mTask = taskIter.next(); app.waitForState(mTask, TaskState.RUNNING); Task rTask = taskIter.next(); app.waitForState(rTask, TaskState.RUNNING); Map<TaskAttemptId, TaskAttempt> mAttempts = mTask.getAttempts(); Assert.assertEquals("Num attempts is not correct", 1, mAttempts.size()); Map<TaskAttemptId, TaskAttempt> rAttempts = rTask.getAttempts(); Assert.assertEquals("Num attempts is not correct", 1, rAttempts.size()); TaskAttempt mta = mAttempts.values().iterator().next(); TaskAttempt rta = rAttempts.values().iterator().next(); app.waitForState(mta, TaskAttemptState.RUNNING); app.waitForState(rta, TaskAttemptState.RUNNING); clock.setTime(11); app.getContext() .getEventHandler() .handle(new TaskAttemptEvent(mta.getID(), TaskAttemptEventType.TA_DONE)); app.getContext() .getEventHandler() .handle(new TaskAttemptEvent(rta.getID(), TaskAttemptEventType.TA_DONE)); app.waitForState(job, JobState.SUCCEEDED); Assert.assertEquals(mta.getFinishTime(), 11); Assert.assertEquals(mta.getLaunchTime(), 10); Assert.assertEquals(rta.getFinishTime(), 11); Assert.assertEquals(rta.getLaunchTime(), 10); Counters counters = job.getAllCounters(); Assert.assertEquals((int) Math.ceil((float) mapMemMb / minContainerSize), counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue()); Assert.assertEquals((int) Math.ceil((float) reduceMemMb / minContainerSize), counters.findCounter(JobCounter.SLOTS_MILLIS_REDUCES).getValue()); Assert.assertEquals(1, counters.findCounter(JobCounter.MILLIS_MAPS).getValue()); Assert.assertEquals(1, counters.findCounter(JobCounter.MILLIS_REDUCES).getValue()); Assert.assertEquals(mapMemMb, counters.findCounter(JobCounter.MB_MILLIS_MAPS).getValue()); Assert.assertEquals(reduceMemMb, counters.findCounter(JobCounter.MB_MILLIS_REDUCES).getValue()); Assert.assertEquals(1, counters.findCounter(JobCounter.VCORES_MILLIS_MAPS).getValue()); Assert.assertEquals(1, counters.findCounter(JobCounter.VCORES_MILLIS_REDUCES).getValue()); }
