### Question:
InProcessStageConfigurator extends StageConfigurator { @Override public void configure(Job job) throws IOException, InterruptedException { if (job.getConfiguration().getBoolean(KEY_FORCE, false)) { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "force enabled in-process execution: {0}", job.getJobName())); } install(job); return; } long limit = job.getConfiguration().getLong(KEY_LIMIT, -1L); if (limit < 0L) { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "in-process execution is disabled: {0}", job.getJobName())); } return; } if (hasCustomJobRunner(job)) { if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "custom job runner is already activated: {0}", job.getJobName())); } return; } long estimated = getEstimatedJobSize(job); if (LOG.isDebugEnabled()) { LOG.debug(MessageFormat.format( "estimated input data size for in-process execution: " + "job={0}, limit={1}, estimated={2}", job.getJobName(), limit, estimated)); } if (estimated < 0L || estimated > limit) { return; } if (LOG.isInfoEnabled()) { LOG.info(MessageFormat.format( "enable in-process execution: job={0}, limit={1}, estimated={2}", job.getJobName(), limit, estimated)); } install(job); }  @Override void configure(Job job); static final String KEY_LIMIT; static final String KEY_FORCE; }### Answer:
@Test public void simple() throws Exception { Job job = newJob(); Configuration conf = job.getConfiguration(); conf.setLong(KEY_LIMIT, 100); StageInputFormat.setSplitCombinerClass(job, IdentitySplitCombiner.class); new Mock(100).configure(job); assertThat(conf.get(StageConstants.PROP_JOB_RUNNER), is(notNullValue())); assertThat(StageResourceDriver.getAccessMode(job), is(StageResourceDriver.AccessMode.DIRECT)); assertThat(StageInputFormat.getSplitCombinerClass(job), is((Object) ExtremeSplitCombiner.class)); }
@Test public void disabled() throws Exception { Job job = newJob(); Configuration conf = job.getConfiguration(); Assume.assumeThat(conf.get(KEY_LIMIT), is(nullValue())); StageInputFormat.setSplitCombinerClass(job, IdentitySplitCombiner.class); new Mock(100).configure(job); assertThat(conf.get(StageConstants.PROP_JOB_RUNNER), is(nullValue())); assertThat(StageResourceDriver.getAccessMode(job), is(not(StageResourceDriver.AccessMode.DIRECT))); assertThat(StageInputFormat.getSplitCombinerClass(job), is(not((Object) ExtremeSplitCombiner.class))); }
@Test public void large() throws Exception { Job job = newJob(); Configuration conf = job.getConfiguration(); conf.setLong(KEY_LIMIT, 100); new Mock(101).configure(job); assertThat(conf.get(StageConstants.PROP_JOB_RUNNER), is(nullValue())); }
@Test public void activate_properties() throws Exception { Job job = newJob(); Configuration conf = job.getConfiguration(); conf.setLong(KEY_LIMIT, 100); conf.set(KEY_PREFIX_REPLACE + "com.example.testing", "YES!"); new Mock(100).configure(job); assertThat(conf.get("com.example.testing"), is("YES!")); }
@Test public void activate_properties_skip() throws Exception { Job job = newJob(); Configuration conf = job.getConfiguration(); conf.setLong(KEY_LIMIT, 100); conf.set(KEY_PREFIX_REPLACE + "com.example.testing", "YES!"); new Mock(1000).configure(job); assertThat(conf.get("com.example.testing"), is(not("YES!"))); }