@Test public void simple() throws Exception { BasicBatchInfo batch = new BasicBatchInfo("b"); BasicJobflowInfo jobflow = new BasicJobflowInfo("f"); batch.addElement(jobflow); Map<String, String> arguments = Collections.singletonMap("testing", "OK"); AtomicBoolean worked = new AtomicBoolean(false); BatchExecutor executor = new BasicBatchExecutor((c, j) -> { assertThat(worked.get(), is(false)); assertThat(c.getBatchId(), is(batch.getId())); assertThat(c.getFlowId(), is(jobflow.getId())); assertThat(c.getExecutionId(), is("testing")); assertThat(c.getBatchArguments(), is(arguments)); assertThat(j, is(jobflow)); worked.set(true); }, () -> "testing"); executor.execute(context, batch, arguments); assertThat(worked.get(), is(true)); }
@Override public void execute( ExecutionContext context, BatchInfo batch, Map<String, String> arguments) throws IOException, InterruptedException { LOG.info("start batch: {} ({})", batch.getId(), arguments); if (validateParameters) { batch.findAttribute(ParameterListAttribute.class) .ifPresent(it -> validateParameters(it, arguments)); } if (LOG.isDebugEnabled()) { LOG.debug("starting jobflow graph: {} ({} jobflows)", batch.getId(), batch.getElements().size()); } int count = 0; for (JobflowInfo jobflow : Util.sort(batch.getElements())) { if (LOG.isDebugEnabled()) { LOG.debug("preparing jobflow: {} ({}/{})", jobflow.getId(), ++count, batch.getElements().size()); } executeJobflow(context, batch, jobflow, arguments); } LOG.info("finish batch: {} ({})", batch.getId(), arguments); }
BasicBatchExecutor implements BatchExecutor { @Override public void execute( ExecutionContext context, BatchInfo batch, Map<String, String> arguments) throws IOException, InterruptedException { LOG.info("start batch: {} ({})", batch.getId(), arguments); if (validateParameters) { batch.findAttribute(ParameterListAttribute.class) .ifPresent(it -> validateParameters(it, arguments)); } if (LOG.isDebugEnabled()) { LOG.debug("starting jobflow graph: {} ({} jobflows)", batch.getId(), batch.getElements().size()); } int count = 0; for (JobflowInfo jobflow : Util.sort(batch.getElements())) { if (LOG.isDebugEnabled()) { LOG.debug("preparing jobflow: {} ({}/{})", jobflow.getId(), ++count, batch.getElements().size()); } executeJobflow(context, batch, jobflow, arguments); } LOG.info("finish batch: {} ({})", batch.getId(), arguments); } }
BasicBatchExecutor implements BatchExecutor { @Override public void execute( ExecutionContext context, BatchInfo batch, Map<String, String> arguments) throws IOException, InterruptedException { LOG.info("start batch: {} ({})", batch.getId(), arguments); if (validateParameters) { batch.findAttribute(ParameterListAttribute.class) .ifPresent(it -> validateParameters(it, arguments)); } if (LOG.isDebugEnabled()) { LOG.debug("starting jobflow graph: {} ({} jobflows)", batch.getId(), batch.getElements().size()); } int count = 0; for (JobflowInfo jobflow : Util.sort(batch.getElements())) { if (LOG.isDebugEnabled()) { LOG.debug("preparing jobflow: {} ({}/{})", jobflow.getId(), ++count, batch.getElements().size()); } executeJobflow(context, batch, jobflow, arguments); } LOG.info("finish batch: {} ({})", batch.getId(), arguments); } BasicBatchExecutor(JobflowExecutor jobflowExecutor, Function<JobflowInfo, String> executionIds);  BasicBatchExecutor(JobflowExecutor jobflowExecutor, Supplier<String> executionIds);  BasicBatchExecutor(JobflowExecutor jobflowExecutor); }
BasicBatchExecutor implements BatchExecutor { @Override public void execute( ExecutionContext context, BatchInfo batch, Map<String, String> arguments) throws IOException, InterruptedException { LOG.info("start batch: {} ({})", batch.getId(), arguments); if (validateParameters) { batch.findAttribute(ParameterListAttribute.class) .ifPresent(it -> validateParameters(it, arguments)); } if (LOG.isDebugEnabled()) { LOG.debug("starting jobflow graph: {} ({} jobflows)", batch.getId(), batch.getElements().size()); } int count = 0; for (JobflowInfo jobflow : Util.sort(batch.getElements())) { if (LOG.isDebugEnabled()) { LOG.debug("preparing jobflow: {} ({}/{})", jobflow.getId(), ++count, batch.getElements().size()); } executeJobflow(context, batch, jobflow, arguments); } LOG.info("finish batch: {} ({})", batch.getId(), arguments); } BasicBatchExecutor(JobflowExecutor jobflowExecutor, Function<JobflowInfo, String> executionIds);  BasicBatchExecutor(JobflowExecutor jobflowExecutor, Supplier<String> executionIds);  BasicBatchExecutor(JobflowExecutor jobflowExecutor); BasicBatchExecutor withValidateParameters(boolean enable); @Override void execute(
            ExecutionContext context,
            BatchInfo batch, Map<String, String> arguments); }
BasicBatchExecutor implements BatchExecutor { @Override public void execute( ExecutionContext context, BatchInfo batch, Map<String, String> arguments) throws IOException, InterruptedException { LOG.info("start batch: {} ({})", batch.getId(), arguments); if (validateParameters) { batch.findAttribute(ParameterListAttribute.class) .ifPresent(it -> validateParameters(it, arguments)); } if (LOG.isDebugEnabled()) { LOG.debug("starting jobflow graph: {} ({} jobflows)", batch.getId(), batch.getElements().size()); } int count = 0; for (JobflowInfo jobflow : Util.sort(batch.getElements())) { if (LOG.isDebugEnabled()) { LOG.debug("preparing jobflow: {} ({}/{})", jobflow.getId(), ++count, batch.getElements().size()); } executeJobflow(context, batch, jobflow, arguments); } LOG.info("finish batch: {} ({})", batch.getId(), arguments); } BasicBatchExecutor(JobflowExecutor jobflowExecutor, Function<JobflowInfo, String> executionIds);  BasicBatchExecutor(JobflowExecutor jobflowExecutor, Supplier<String> executionIds);  BasicBatchExecutor(JobflowExecutor jobflowExecutor); BasicBatchExecutor withValidateParameters(boolean enable); @Override void execute(
            ExecutionContext context,
            BatchInfo batch, Map<String, String> arguments);  }