@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!")); }
@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); }
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); } }
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); }  }
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); }
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; }