@Override
  public void handle (TaskExecution aTaskExecution) {
    SimpleTaskExecution mtask = SimpleTaskExecution.of(aTaskExecution);
    mtask.setStatus(TaskStatus.COMPLETED);
    taskExecutionRepo.merge(mtask);
    
    if(aTaskExecution.getOutput() != null && aTaskExecution.getName() != null) {
      Context context = contextRepository.peek(aTaskExecution.getParentId()+"/"+aTaskExecution.getInteger("branch"));
      MapContext newContext = new MapContext(context.asMap());
      newContext.put(aTaskExecution.getName(), aTaskExecution.getOutput());
      contextRepository.push(aTaskExecution.getParentId()+"/"+aTaskExecution.getInteger("branch"), newContext);
    }
    
    TaskExecution fork = taskExecutionRepo.findOne(aTaskExecution.getParentId());
    List<List> list = fork.getList("branches", List.class);
    List<Map<String,Object>> branch = list.get(aTaskExecution.getInteger("branch"));
    if(aTaskExecution.getTaskNumber() < branch.size()) {
      Map<String,Object> task = branch.get(aTaskExecution.getTaskNumber());
      SimpleTaskExecution execution = SimpleTaskExecution.of(task);
      execution.setId(UUIDGenerator.generate());
      execution.setStatus(TaskStatus.CREATED);
      execution.setCreateTime(new Date());
      execution.set("branch", aTaskExecution.getInteger("branch"));
      execution.setTaskNumber(aTaskExecution.getTaskNumber()+1);
      execution.setJobId(aTaskExecution.getJobId());
      execution.setParentId(aTaskExecution.getParentId());
      execution.setPriority(aTaskExecution.getPriority());
      Context context = contextRepository.peek(aTaskExecution.getParentId()+"/"+aTaskExecution.getInteger("branch"));
      contextRepository.push(execution.getId(), context);
      TaskExecution evaluatedExecution = taskEvaluator.evaluate(execution, context);
      taskExecutionRepo.create(evaluatedExecution);
      taskDispatcher.dispatch(evaluatedExecution);
    }
    else {
      long branchesLeft = counterRepository.decrement(aTaskExecution.getParentId());
      if(branchesLeft == 0) {
        SimpleTaskExecution forkTask = SimpleTaskExecution.of(taskExecutionRepo.findOne(aTaskExecution.getParentId()));
        forkTask.setEndTime(new Date ());
        forkTask.setExecutionTime(forkTask.getEndTime().getTime()-forkTask.getStartTime().getTime());
        taskCompletionHandler.handle(forkTask);
      }
    }
  }
Handles {@link TaskExecution} completions which are the child 
execution tasks of a parent <code>fork</code> {@link TaskExecution}.

This handler will either execute the next task in the branch, or if 
arrived at the last task, it will complete the branch and check if
all branches are now complete. If so, it will complete the overall 
<code>fork</code> task.

@author Arik Cohen
@since May 11, 2017
src/main/java/com/creactiviti/piper/core/ForkTaskCompletionHandler.java