class JobFlowExecutor implements FlowExecutor {

	private final ThreadLocal<StepExecution> stepExecutionHolder = new ThreadLocal<>();

	private final JobExecution execution;

	protected ExitStatus exitStatus = ExitStatus.EXECUTING;

	private final StepHandler stepHandler;

	private final JobRepository jobRepository;

	/**
	 * @param jobRepository instance of {@link JobRepository}.
	 * @param stepHandler instance of {@link StepHandler}.
	 * @param execution instance of {@link JobExecution}.
	 */
	public JobFlowExecutor(JobRepository jobRepository, StepHandler stepHandler, JobExecution execution) {
		this.jobRepository = jobRepository;
		this.stepHandler = stepHandler;
		this.execution = execution;
		stepExecutionHolder.set(null);
	}

	@Override
	public String executeStep(Step step)
			throws JobInterruptedException, JobRestartException, StartLimitExceededException {<FILL_FUNCTION_BODY>}

	private boolean isStepRestart(Step step) {
		long count = jobRepository.getStepExecutionCount(execution.getJobInstance(), step.getName());

		return count > 0;
	}

	@Override
	public void abandonStepExecution() {
		StepExecution lastStepExecution = stepExecutionHolder.get();
		if (lastStepExecution != null && lastStepExecution.getStatus().isGreaterThan(BatchStatus.STOPPING)) {
			lastStepExecution.upgradeStatus(BatchStatus.ABANDONED);
			jobRepository.update(lastStepExecution);
		}
	}

	@Override
	public void updateJobExecutionStatus(FlowExecutionStatus status) {
		execution.setStatus(findBatchStatus(status));
		exitStatus = exitStatus.and(new ExitStatus(status.getName()));
		execution.setExitStatus(exitStatus);
	}

	@Override
	public JobExecution getJobExecution() {
		return execution;
	}

	@Override
	@Nullable
	public StepExecution getStepExecution() {
		return stepExecutionHolder.get();
	}

	@Override
	public void close(FlowExecution result) {
		stepExecutionHolder.set(null);
	}

	@Override
	public boolean isRestart() {
		if (getStepExecution() != null && getStepExecution().getStatus() == BatchStatus.ABANDONED) {
			/*
			 * This is assumed to be the last step execution and it was marked abandoned,
			 * so we are in a restart of a stopped step.
			 */
			// TODO: mark the step execution in some more definitive way?
			return true;
		}
		return execution.getStepExecutions().isEmpty();
	}

	@Override
	public void addExitStatus(String code) {
		exitStatus = exitStatus.and(new ExitStatus(code));
	}

	/**
	 * @param status {@link FlowExecutionStatus} to convert.
	 * @return A {@link BatchStatus} appropriate for the {@link FlowExecutionStatus}
	 * provided
	 */
	protected BatchStatus findBatchStatus(FlowExecutionStatus status) {
		for (BatchStatus batchStatus : BatchStatus.values()) {
			if (status.getName().startsWith(batchStatus.toString())) {
				return batchStatus;
			}
		}
		return BatchStatus.UNKNOWN;
	}

}

		boolean isRerun = isStepRestart(step);
		StepExecution stepExecution = stepHandler.handleStep(step, execution);
		stepExecutionHolder.set(stepExecution);

		if (stepExecution == null) {
			return ExitStatus.COMPLETED.getExitCode();
		}
		if (stepExecution.isTerminateOnly()) {
			throw new JobInterruptedException("Step requested termination: " + stepExecution,
					stepExecution.getStatus());
		}

		if (isRerun) {
			stepExecution.getExecutionContext().put("batch.restart", true);
		}

		return stepExecution.getExitStatus().getExitCode();
	
<no_super_class>