class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, InitializingBean {

	private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class);

	private ChunkProcessor<S> chunkProcessor;

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

	/**
	 * Public setter for the {@link ChunkProcessor}.
	 * @param chunkProcessor the chunkProcessor to set
	 */
	public void setChunkProcessor(ChunkProcessor<S> chunkProcessor) {
		this.chunkProcessor = chunkProcessor;
	}

	/**
	 *
	 * @see ChunkHandler#handleChunk(ChunkRequest)
	 */
	@Override
	@ServiceActivator
	public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) throws Exception {

		if (logger.isDebugEnabled()) {
			logger.debug("Handling chunk: " + chunkRequest);
		}

		StepContribution stepContribution = chunkRequest.getStepContribution();

		Throwable failure = process(chunkRequest, stepContribution);
		if (failure != null) {
			logger.debug("Failed chunk", failure);
			return new ChunkResponse(false, chunkRequest.getSequence(), chunkRequest.getJobId(), stepContribution,
					failure.getClass().getName() + ": " + failure.getMessage());
		}

		if (logger.isDebugEnabled()) {
			logger.debug("Completed chunk handling with " + stepContribution);
		}
		return new ChunkResponse(true, chunkRequest.getSequence(), chunkRequest.getJobId(), stepContribution);

	}

	/**
	 * @param chunkRequest the current request
	 * @param stepContribution the step contribution to update
	 * @throws Exception if there is a fatal exception
	 */
	@SuppressWarnings(value = { "unchecked", "rawtypes" })
	private Throwable process(ChunkRequest<S> chunkRequest, StepContribution stepContribution) throws Exception {<FILL_FUNCTION_BODY>}

}


		Chunk chunk = chunkRequest.getItems();
		Throwable failure = null;
		try {
			chunkProcessor.process(stepContribution, chunk);
		}
		catch (SkipLimitExceededException | NonSkippableReadException | SkipListenerFailedException | RetryException
				| JobInterruptedException e) {
			failure = e;
		}
		catch (Exception e) {
			if (chunkProcessor instanceof FaultTolerantChunkProcessor<?, ?>) {
				// try again...
				throw e;
			}
			else {
				failure = e;
			}
		}

		return failure;

	
<no_super_class>