@SuppressWarnings({"unchecked", "rawtypes"}) @Test (timeout=30000) public void testSleepMapper() throws Exception { SleepJob.SleepMapper test = new SleepJob.SleepMapper(); Configuration conf = new Configuration(); conf.setInt(JobContext.NUM_REDUCES, 2); CompressionEmulationUtil.setCompressionEmulationEnabled(conf, true); conf.setBoolean(MRJobConfig.MAP_OUTPUT_COMPRESS, true); TaskAttemptID taskId = new TaskAttemptID(); FakeRecordLLReader reader = new FakeRecordLLReader(); LoadRecordGkNullWriter writer = new LoadRecordGkNullWriter(); OutputCommitter committer = new CustomOutputCommitter(); StatusReporter reporter = new TaskAttemptContextImpl.DummyReporter(); SleepSplit split = getSleepSplit(); MapContext<LongWritable, LongWritable, GridmixKey, NullWritable> mapcontext = new MapContextImpl<LongWritable, LongWritable, GridmixKey, NullWritable>( conf, taskId, reader, writer, committer, reporter, split); Context context = new WrappedMapper<LongWritable, LongWritable, GridmixKey, NullWritable>() .getMapContext(mapcontext); long start = System.currentTimeMillis(); LOG.info("start:" + start); LongWritable key = new LongWritable(start + 2000); LongWritable value = new LongWritable(start + 2000); // should slip 2 sec test.map(key, value, context); LOG.info("finish:" + System.currentTimeMillis()); assertTrue(System.currentTimeMillis() >= (start + 2000)); test.cleanup(context); assertEquals(1, writer.getData().size()); }
@SuppressWarnings({"unchecked", "rawtypes"}) @Test (timeout=10000) public void testSleepMapper() throws Exception { SleepJob.SleepMapper test = new SleepJob.SleepMapper(); Configuration conf = new Configuration(); conf.setInt(JobContext.NUM_REDUCES, 2); CompressionEmulationUtil.setCompressionEmulationEnabled(conf, true); conf.setBoolean(MRJobConfig.MAP_OUTPUT_COMPRESS, true); TaskAttemptID taskId = new TaskAttemptID(); FakeRecordLLReader reader = new FakeRecordLLReader(); LoadRecordGkNullWriter writer = new LoadRecordGkNullWriter(); OutputCommitter committer = new CustomOutputCommitter(); StatusReporter reporter = new TaskAttemptContextImpl.DummyReporter(); SleepSplit split = getSleepSplit(); MapContext<LongWritable, LongWritable, GridmixKey, NullWritable> mapcontext = new MapContextImpl<LongWritable, LongWritable, GridmixKey, NullWritable>( conf, taskId, reader, writer, committer, reporter, split); Context context = new WrappedMapper<LongWritable, LongWritable, GridmixKey, NullWritable>() .getMapContext(mapcontext); long start = System.currentTimeMillis(); LOG.info("start:" + start); LongWritable key = new LongWritable(start + 2000); LongWritable value = new LongWritable(start + 2000); // should slip 2 sec test.map(key, value, context); LOG.info("finish:" + System.currentTimeMillis()); assertTrue(System.currentTimeMillis() >= (start + 2000)); test.cleanup(context); assertEquals(1, writer.getData().size()); }
private SleepSplit getSleepSplit() throws Exception { String[] locations = {"locOne", "loctwo"}; long[] reduceDurations = {101L, 102L}; return new SleepSplit(0, 2000L, reduceDurations, 2, locations); }