class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, InitializingBean {

	private ItemProcessor<I, O> delegate;

	private TaskExecutor taskExecutor = new SyncTaskExecutor();

	/**
	 * Check mandatory properties (the {@link #setDelegate(ItemProcessor)}).
	 *
	 * @see InitializingBean#afterPropertiesSet()
	 */
	@Override
	public void afterPropertiesSet() throws Exception {
		Assert.state(delegate != null, "The delegate must be set.");
	}

	/**
	 * The {@link ItemProcessor} to use to delegate processing to in a background thread.
	 * @param delegate the {@link ItemProcessor} to use as a delegate
	 */
	public void setDelegate(ItemProcessor<I, O> delegate) {
		this.delegate = delegate;
	}

	/**
	 * The {@link TaskExecutor} to use to allow the item processing to proceed in the
	 * background. Defaults to a {@link SyncTaskExecutor} so no threads are created unless
	 * this is overridden.
	 * @param taskExecutor a {@link TaskExecutor}
	 */
	public void setTaskExecutor(TaskExecutor taskExecutor) {
		this.taskExecutor = taskExecutor;
	}

	/**
	 * Transform the input by delegating to the provided item processor. The return value
	 * is wrapped in a {@link Future} so that clients can unpack it later.
	 *
	 * @see ItemProcessor#process(Object)
	 */
	@Override
	@Nullable
	public Future<O> process(final I item) throws Exception {<FILL_FUNCTION_BODY>}

	/**
	 * @return the current step execution if there is one
	 */
	private StepExecution getStepExecution() {
		StepContext context = StepSynchronizationManager.getContext();
		if (context == null) {
			return null;
		}
		return context.getStepExecution();
	}

}

		final StepExecution stepExecution = getStepExecution();
		FutureTask<O> task = new FutureTask<>(() -> {
			if (stepExecution != null) {
				StepSynchronizationManager.register(stepExecution);
			}
			try {
				return delegate.process(item);
			}
			finally {
				if (stepExecution != null) {
					StepSynchronizationManager.close();
				}
			}
		});
		taskExecutor.execute(task);
		return task;
	
<no_super_class>