class TaskExecutorPartitionHandler extends AbstractPartitionHandler implements StepHolder, InitializingBean {

	private TaskExecutor taskExecutor = new SyncTaskExecutor();

	private Step step;

	@Override
	public void afterPropertiesSet() throws Exception {
		Assert.state(step != null, "A Step must be provided.");
	}

	/**
	 * Setter for the {@link TaskExecutor} that is used to farm out step executions to
	 * multiple threads.
	 * @param taskExecutor a {@link TaskExecutor}
	 */
	public void setTaskExecutor(TaskExecutor taskExecutor) {
		this.taskExecutor = taskExecutor;
	}

	/**
	 * Setter for the {@link Step} that will be used to execute the partitioned
	 * {@link StepExecution}. This is a regular Spring Batch step, with all the business
	 * logic required to complete an execution based on the input parameters in its
	 * {@link StepExecution} context.
	 * @param step the {@link Step} instance to use to execute business logic
	 */
	public void setStep(Step step) {
		this.step = step;
	}

	/**
	 * The step instance that will be executed in parallel by this handler.
	 * @return the step instance that will be used
	 * @see StepHolder#getStep()
	 */
	@Override
	public Step getStep() {
		return this.step;
	}

	@Override
	protected Set<StepExecution> doHandle(StepExecution managerStepExecution,
			Set<StepExecution> partitionStepExecutions) throws Exception {<FILL_FUNCTION_BODY>}

	/**
	 * Creates the task executing the given step in the context of the given execution.
	 * @param step the step to execute
	 * @param stepExecution the given execution
	 * @return the task executing the given step
	 */
	protected FutureTask<StepExecution> createTask(final Step step, final StepExecution stepExecution) {
		return new FutureTask<>(() -> {
			step.execute(stepExecution);
			return stepExecution;
		});
	}

}

		Assert.notNull(step, "A Step must be provided.");
		final Set<Future<StepExecution>> tasks = new HashSet<>(getGridSize());
		final Set<StepExecution> result = new HashSet<>();

		for (final StepExecution stepExecution : partitionStepExecutions) {
			final FutureTask<StepExecution> task = createTask(step, stepExecution);

			try {
				taskExecutor.execute(task);
				tasks.add(task);
			}
			catch (TaskRejectedException e) {
				// couldn't execute one of the tasks
				ExitStatus exitStatus = ExitStatus.FAILED
					.addExitDescription("TaskExecutor rejected the task for this step.");
				/*
				 * Set the status in case the caller is tracking it through the
				 * JobExecution.
				 */
				stepExecution.setStatus(BatchStatus.FAILED);
				stepExecution.setExitStatus(exitStatus);
				result.add(stepExecution);
			}
		}

		for (Future<StepExecution> task : tasks) {
			result.add(task.get());
		}

		return result;
	
<methods>public non-sealed void <init>() ,public int getGridSize() ,public Collection<org.springframework.batch.core.StepExecution> handle(org.springframework.batch.core.partition.StepExecutionSplitter, org.springframework.batch.core.StepExecution) throws java.lang.Exception,public void setGridSize(int) <variables>protected int gridSize