@Test public void dependencies() throws Exception { BasicJobflowInfo a = new BasicJobflowInfo("A"); BasicJobflowInfo b = new BasicJobflowInfo("B"); BasicJobflowInfo c = new BasicJobflowInfo("C"); a.addBlocker(b); b.addBlocker(c); BasicBatchInfo batch = new BasicBatchInfo("b"); batch.addElement(a); batch.addElement(b); batch.addElement(c); List<String> flowIds = new ArrayList<>(); List<String> execIds = new ArrayList<>(); BatchExecutor executor = new BasicBatchExecutor((ctxt, jobflow) -> { flowIds.add(ctxt.getFlowId()); execIds.add(ctxt.getExecutionId()); }); executor.execute(context, batch); assertThat(flowIds, contains("C", "B", "A")); assertThat(execIds.stream().distinct().count(), is(3L)); }
@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);  }