@Test public void dependencies() throws Exception { BasicJobflowInfo jobflow = new BasicJobflowInfo(context.getFlowId()); MockTaskInfo a = new MockTaskInfo("A"); MockTaskInfo b = new MockTaskInfo("B"); MockTaskInfo c = new MockTaskInfo("C"); a.addBlocker(b); b.addBlocker(c); jobflow.addTask(TaskInfo.Phase.MAIN, a); jobflow.addTask(TaskInfo.Phase.MAIN, b); jobflow.addTask(TaskInfo.Phase.MAIN, c); MockExecutor e0 = new MockExecutor(t -> true); JobflowExecutor executor = new BasicJobflowExecutor(e0); executor.execute(context, jobflow); assertThat(e0.getValues(), contains("C", "B", "A")); }
@Override public void execute(TaskExecutionContext context, JobflowInfo jobflow) throws IOException, InterruptedException { LOG.info("start jobflow: {} - {}", context.getBatchId(), context.getFlowId()); boolean finalized = false; try { for (TaskInfo.Phase phase : EnumSet.complementOf(BODY)) { executePhase(context, jobflow, phase); } finalized = true; executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception e) { if (finalized == false && jobflow.getTasks(TaskInfo.Phase.FINALIZE).isEmpty() == false) { LOG.error(MessageFormat.format( "batch \"{0}\" was failed. we try to execute finalize phase.", context.getBatchId()), e); try { executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception nested) { LOG.warn(MessageFormat.format( "error occurred while running finalize of batch \"{0}\".", context.getBatchId()), nested); e.addSuppressed(nested); } } throw e; } try { executePhase(context, jobflow, TaskInfo.Phase.CLEANUP); } catch (Exception e) { LOG.warn(MessageFormat.format( "error occurred while runnning cleanup of batch \"{0}\".", context.getBatchId()), e); } LOG.info("finish jobflow: {} - {}", context.getBatchId(), context.getFlowId()); }
BasicJobflowExecutor implements JobflowExecutor { @Override public void execute(TaskExecutionContext context, JobflowInfo jobflow) throws IOException, InterruptedException { LOG.info("start jobflow: {} - {}", context.getBatchId(), context.getFlowId()); boolean finalized = false; try { for (TaskInfo.Phase phase : EnumSet.complementOf(BODY)) { executePhase(context, jobflow, phase); } finalized = true; executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception e) { if (finalized == false && jobflow.getTasks(TaskInfo.Phase.FINALIZE).isEmpty() == false) { LOG.error(MessageFormat.format( "batch \"{0}\" was failed. we try to execute finalize phase.", context.getBatchId()), e); try { executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception nested) { LOG.warn(MessageFormat.format( "error occurred while running finalize of batch \"{0}\".", context.getBatchId()), nested); e.addSuppressed(nested); } } throw e; } try { executePhase(context, jobflow, TaskInfo.Phase.CLEANUP); } catch (Exception e) { LOG.warn(MessageFormat.format( "error occurred while runnning cleanup of batch \"{0}\".", context.getBatchId()), e); } LOG.info("finish jobflow: {} - {}", context.getBatchId(), context.getFlowId()); } }
BasicJobflowExecutor implements JobflowExecutor { @Override public void execute(TaskExecutionContext context, JobflowInfo jobflow) throws IOException, InterruptedException { LOG.info("start jobflow: {} - {}", context.getBatchId(), context.getFlowId()); boolean finalized = false; try { for (TaskInfo.Phase phase : EnumSet.complementOf(BODY)) { executePhase(context, jobflow, phase); } finalized = true; executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception e) { if (finalized == false && jobflow.getTasks(TaskInfo.Phase.FINALIZE).isEmpty() == false) { LOG.error(MessageFormat.format( "batch \"{0}\" was failed. we try to execute finalize phase.", context.getBatchId()), e); try { executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception nested) { LOG.warn(MessageFormat.format( "error occurred while running finalize of batch \"{0}\".", context.getBatchId()), nested); e.addSuppressed(nested); } } throw e; } try { executePhase(context, jobflow, TaskInfo.Phase.CLEANUP); } catch (Exception e) { LOG.warn(MessageFormat.format( "error occurred while runnning cleanup of batch \"{0}\".", context.getBatchId()), e); } LOG.info("finish jobflow: {} - {}", context.getBatchId(), context.getFlowId()); } BasicJobflowExecutor(Collection<? extends TaskExecutor> taskExecutors);  BasicJobflowExecutor(TaskExecutor... taskExecutors); }
BasicJobflowExecutor implements JobflowExecutor { @Override public void execute(TaskExecutionContext context, JobflowInfo jobflow) throws IOException, InterruptedException { LOG.info("start jobflow: {} - {}", context.getBatchId(), context.getFlowId()); boolean finalized = false; try { for (TaskInfo.Phase phase : EnumSet.complementOf(BODY)) { executePhase(context, jobflow, phase); } finalized = true; executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception e) { if (finalized == false && jobflow.getTasks(TaskInfo.Phase.FINALIZE).isEmpty() == false) { LOG.error(MessageFormat.format( "batch \"{0}\" was failed. we try to execute finalize phase.", context.getBatchId()), e); try { executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception nested) { LOG.warn(MessageFormat.format( "error occurred while running finalize of batch \"{0}\".", context.getBatchId()), nested); e.addSuppressed(nested); } } throw e; } try { executePhase(context, jobflow, TaskInfo.Phase.CLEANUP); } catch (Exception e) { LOG.warn(MessageFormat.format( "error occurred while runnning cleanup of batch \"{0}\".", context.getBatchId()), e); } LOG.info("finish jobflow: {} - {}", context.getBatchId(), context.getFlowId()); } BasicJobflowExecutor(Collection<? extends TaskExecutor> taskExecutors);  BasicJobflowExecutor(TaskExecutor... taskExecutors); @Override void execute(TaskExecutionContext context, JobflowInfo jobflow); }
BasicJobflowExecutor implements JobflowExecutor { @Override public void execute(TaskExecutionContext context, JobflowInfo jobflow) throws IOException, InterruptedException { LOG.info("start jobflow: {} - {}", context.getBatchId(), context.getFlowId()); boolean finalized = false; try { for (TaskInfo.Phase phase : EnumSet.complementOf(BODY)) { executePhase(context, jobflow, phase); } finalized = true; executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception e) { if (finalized == false && jobflow.getTasks(TaskInfo.Phase.FINALIZE).isEmpty() == false) { LOG.error(MessageFormat.format( "batch \"{0}\" was failed. we try to execute finalize phase.", context.getBatchId()), e); try { executePhase(context, jobflow, TaskInfo.Phase.FINALIZE); } catch (Exception nested) { LOG.warn(MessageFormat.format( "error occurred while running finalize of batch \"{0}\".", context.getBatchId()), nested); e.addSuppressed(nested); } } throw e; } try { executePhase(context, jobflow, TaskInfo.Phase.CLEANUP); } catch (Exception e) { LOG.warn(MessageFormat.format( "error occurred while runnning cleanup of batch \"{0}\".", context.getBatchId()), e); } LOG.info("finish jobflow: {} - {}", context.getBatchId(), context.getFlowId()); } BasicJobflowExecutor(Collection<? extends TaskExecutor> taskExecutors);  BasicJobflowExecutor(TaskExecutor... taskExecutors); @Override void execute(TaskExecutionContext context, JobflowInfo jobflow);  }