@Test public void executeFlow_failed_finalize() throws Exception { ProfileBuilder prf = new ProfileBuilder(folder.getRoot()); prf.setTracker(FinalizerFailed.class); ExecutionTask task = prf.task(); try { task.executeFlow("batch", "testing", "flow"); fail(); } catch (IOException e) { } List<Record> results = SerialExecutionTracker.get(prf.trackingId); verifyPhaseOrder(results); assertThat(phase(results, "testing", ExecutionPhase.SETUP).size(), is(2)); assertThat(phase(results, "testing", ExecutionPhase.INITIALIZE).size(), is(1)); assertThat(phase(results, "testing", ExecutionPhase.IMPORT).size(), is(2)); assertThat(phase(results, "testing", ExecutionPhase.PROLOGUE).size(), is(1)); assertThat(phase(results, "testing", ExecutionPhase.MAIN).size(), is(4)); assertThat(phase(results, "testing", ExecutionPhase.EPILOGUE).size(), is(1)); assertThat(phase(results, "testing", ExecutionPhase.EXPORT).size(), is(2)); assertThat(phase(results, "testing", ExecutionPhase.FINALIZE).size(), is(1)); assertThat(phase(results, "testing", ExecutionPhase.CLEANUP).size(), is(0)); }
public void executeFlow( String batchId, String flowId, String executionId) throws InterruptedException, IOException { if (batchId == null) { throw new IllegalArgumentException("batchId must not be null"); } if (flowId == null) { throw new IllegalArgumentException("flowId must not be null"); } if (executionId == null) { throw new IllegalArgumentException("executionId must not be null"); } FlowScript flow = script.findFlow(flowId); if (flow == null) { throw new IllegalArgumentException(MessageFormat.format( "Flow is undefined: batchId={0}, flowId={1}, executionId={2}", batchId, flowId, executionId)); } try (ExecutionLock lock = acquireExecutionLock(batchId)) { lock.beginFlow(flowId, executionId); executeFlow(batchId, flow, executionId); lock.endFlow(flowId, executionId); } }
ExecutionTask { public void executeFlow( String batchId, String flowId, String executionId) throws InterruptedException, IOException { if (batchId == null) { throw new IllegalArgumentException("batchId must not be null"); } if (flowId == null) { throw new IllegalArgumentException("flowId must not be null"); } if (executionId == null) { throw new IllegalArgumentException("executionId must not be null"); } FlowScript flow = script.findFlow(flowId); if (flow == null) { throw new IllegalArgumentException(MessageFormat.format( "Flow is undefined: batchId={0}, flowId={1}, executionId={2}", batchId, flowId, executionId)); } try (ExecutionLock lock = acquireExecutionLock(batchId)) { lock.beginFlow(flowId, executionId); executeFlow(batchId, flow, executionId); lock.endFlow(flowId, executionId); } } }
ExecutionTask { public void executeFlow( String batchId, String flowId, String executionId) throws InterruptedException, IOException { if (batchId == null) { throw new IllegalArgumentException("batchId must not be null"); } if (flowId == null) { throw new IllegalArgumentException("flowId must not be null"); } if (executionId == null) { throw new IllegalArgumentException("executionId must not be null"); } FlowScript flow = script.findFlow(flowId); if (flow == null) { throw new IllegalArgumentException(MessageFormat.format( "Flow is undefined: batchId={0}, flowId={1}, executionId={2}", batchId, flowId, executionId)); } try (ExecutionLock lock = acquireExecutionLock(batchId)) { lock.beginFlow(flowId, executionId); executeFlow(batchId, flow, executionId); lock.endFlow(flowId, executionId); } } ExecutionTask(
            ExecutionMonitorProvider monitors,
            ExecutionLockProvider locks,
            JobScheduler scheduler,
            HadoopScriptHandler hadoopHandler,
            Map<String, CommandScriptHandler> commandHandlers,
            BatchScript script,
            Map<String, String> batchArguments,
            Map<String, Blob> extensions); }
ExecutionTask { public void executeFlow( String batchId, String flowId, String executionId) throws InterruptedException, IOException { if (batchId == null) { throw new IllegalArgumentException("batchId must not be null"); } if (flowId == null) { throw new IllegalArgumentException("flowId must not be null"); } if (executionId == null) { throw new IllegalArgumentException("executionId must not be null"); } FlowScript flow = script.findFlow(flowId); if (flow == null) { throw new IllegalArgumentException(MessageFormat.format( "Flow is undefined: batchId={0}, flowId={1}, executionId={2}", batchId, flowId, executionId)); } try (ExecutionLock lock = acquireExecutionLock(batchId)) { lock.beginFlow(flowId, executionId); executeFlow(batchId, flow, executionId); lock.endFlow(flowId, executionId); } } ExecutionTask(
            ExecutionMonitorProvider monitors,
            ExecutionLockProvider locks,
            JobScheduler scheduler,
            HadoopScriptHandler hadoopHandler,
            Map<String, CommandScriptHandler> commandHandlers,
            BatchScript script,
            Map<String, String> batchArguments,
            Map<String, Blob> extensions); static ExecutionTask load(
            YaessProfile profile,
            Properties script,
            Map<String, String> batchArguments); static ExecutionTask load(
            YaessProfile profile,
            Properties script,
            Map<String, String> batchArguments,
            Map<String, String> yaessArguments); static ExecutionTask load(
            YaessProfile profile,
            Properties script,
            Map<String, String> batchArguments,
            Map<String, String> yaessArguments,
            Map<String, Blob> extensions); void executeBatch(String batchId); void executeFlow(
            String batchId,
            String flowId,
            String executionId); void executePhase(
            String batchId,
            String flowId,
            String executionId,
            ExecutionPhase phase); void executePhase(ExecutionContext context); }
ExecutionTask { public void executeFlow( String batchId, String flowId, String executionId) throws InterruptedException, IOException { if (batchId == null) { throw new IllegalArgumentException("batchId must not be null"); } if (flowId == null) { throw new IllegalArgumentException("flowId must not be null"); } if (executionId == null) { throw new IllegalArgumentException("executionId must not be null"); } FlowScript flow = script.findFlow(flowId); if (flow == null) { throw new IllegalArgumentException(MessageFormat.format( "Flow is undefined: batchId={0}, flowId={1}, executionId={2}", batchId, flowId, executionId)); } try (ExecutionLock lock = acquireExecutionLock(batchId)) { lock.beginFlow(flowId, executionId); executeFlow(batchId, flow, executionId); lock.endFlow(flowId, executionId); } } ExecutionTask(
            ExecutionMonitorProvider monitors,
            ExecutionLockProvider locks,
            JobScheduler scheduler,
            HadoopScriptHandler hadoopHandler,
            Map<String, CommandScriptHandler> commandHandlers,
            BatchScript script,
            Map<String, String> batchArguments,
            Map<String, Blob> extensions); static ExecutionTask load(
            YaessProfile profile,
            Properties script,
            Map<String, String> batchArguments); static ExecutionTask load(
            YaessProfile profile,
            Properties script,
            Map<String, String> batchArguments,
            Map<String, String> yaessArguments); static ExecutionTask load(
            YaessProfile profile,
            Properties script,
            Map<String, String> batchArguments,
            Map<String, String> yaessArguments,
            Map<String, Blob> extensions); void executeBatch(String batchId); void executeFlow(
            String batchId,
            String flowId,
            String executionId); void executePhase(
            String batchId,
            String flowId,
            String executionId,
            ExecutionPhase phase); void executePhase(ExecutionContext context);  }