BATCHADM-49: add some javadocs + look at thread safety in chunk writer

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkHandler.java

@@ -16,8 +16,29 @@
 
 package org.springframework.batch.integration.chunk;
 
+
+/**
+ * Interface for a remote worker in the Remote Chunking pattern. A request comes from a master process containing some
+ * items to be processed. Once the items are done with a response needs to be generated containing a summary of the
+ * result.
+ * 
+ * @author Dave Syer
+ * 
+ * @param <T> the type of the items to be processed (it is recommended to use a Memento like a primary key)
+ */
 public interface ChunkHandler<T> {
 
+	/**
+	 * Handle the chunk, processing all the items and returning a response summarising the result. If the result is a
+	 * failure then the response should say so. The handler only throws an exception if it needs to roll back a
+	 * transaction and knows that the request will be re-delivered (if not to the same handler then to one processing
+	 * the same Step).
+	 * 
+	 * @param chunk a request containing the chunk to process
+	 * @return a response summarising the result
+	 * 
+	 * @throws Exception if the handler needs to roll back a transaction and have the chunk re-delivered
+	 */
 	ChunkResponse handleChunk(ChunkRequest<T> chunk) throws Exception;
 
 }

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java

@@ -21,6 +21,7 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -58,11 +59,9 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	private int maxWaitTimeouts = DEFAULT_MAX_WAIT_TIMEOUTS;
 
 	/**
-	 * The maximum number of times to wait at the end of a step for a non-null
-	 * result from the remote workers. This is a multiplier on the receive
-	 * timeout set separately on the gateway. The ideal value is a compromise
-	 * between allowing slow workers time to finish, and responsiveness if there
-	 * is a dead worker. Defaults to 40.
+	 * The maximum number of times to wait at the end of a step for a non-null result from the remote workers. This is a
+	 * multiplier on the receive timeout set separately on the gateway. The ideal value is a compromise between allowing
+	 * slow workers time to finish, and responsiveness if there is a dead worker. Defaults to 40.
 	 * 
 	 * @param maxWaitTimeouts the maximum number of wait timeouts
 	 */
@@ -71,8 +70,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	}
 
 	/**
-	 * Public setter for the throttle limit. This limits the number of pending
-	 * requests for chunk processing to avoid overwhelming the receivers.
+	 * Public setter for the throttle limit. This limits the number of pending requests for chunk processing to avoid
+	 * overwhelming the receivers.
 	 * @param throttleLimit the throttle limit to set
 	 */
 	public void setThrottleLimit(long throttleLimit) {
@@ -96,7 +95,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 			ChunkRequest<T> request = new ChunkRequest<T>(items, localState.getJobId(), localState
 					.createStepContribution());
 			messagingGateway.send(request);
-			localState.expected++;
+			localState.expected.incrementAndGet();
 
 		}
 
@@ -142,8 +141,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 
 	public void open(ExecutionContext executionContext) throws ItemStreamException {
 		if (executionContext.containsKey(EXPECTED)) {
-			localState.expected = executionContext.getLong(EXPECTED);
-			localState.actual = executionContext.getLong(ACTUAL);
+			localState.open(executionContext.getInt(EXPECTED), executionContext.getInt(ACTUAL));
 			if (!waitForResults()) {
 				throw new ItemStreamException("Timed out waiting for back log on open");
 			}
@@ -151,17 +149,16 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	}
 
 	public void update(ExecutionContext executionContext) throws ItemStreamException {
-		executionContext.putLong(EXPECTED, localState.expected);
-		executionContext.putLong(ACTUAL, localState.actual);
+		executionContext.putInt(EXPECTED, localState.expected.intValue());
+		executionContext.putInt(ACTUAL, localState.actual.intValue());
 	}
 
 	public Collection<StepContribution> getStepContributions() {
-		return localState.pollContributions();
+		return localState.pollStepContributions();
 	}
 
 	/**
-	 * Wait until all the results that are in the pipeline come back to the
-	 * reply channel.
+	 * Wait until all the results that are in the pipeline come back to the reply channel.
 	 * 
 	 * @return true if successfully received a result, false if timed out
 	 */
@@ -169,6 +166,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 		int count = 0;
 		int maxCount = maxWaitTimeouts;
 		Throwable failure = null;
+		logger.error("Waiting for " + localState.getExpecting() + " results");
 		while (localState.getExpecting() > 0 && count++ < maxCount) {
 			try {
 				getNextResult();
@@ -186,27 +184,25 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	}
 
 	/**
-	 * Get the next result if it is available (within the timeout specified in
-	 * the gateway), otherwise do nothing.
+	 * Get the next result if it is available (within the timeout specified in the gateway), otherwise do nothing.
 	 * 
-	 * @throws AsynchronousFailureException If there is a response and it
-	 * contains a failed chunk response.
+	 * @throws AsynchronousFailureException If there is a response and it contains a failed chunk response.
 	 * 
-	 * @throws IllegalStateException if the result contains the wrong job
-	 * instance id (maybe we are sharing a channel and we shouldn't be)
+	 * @throws IllegalStateException if the result contains the wrong job instance id (maybe we are sharing a channel
+	 * and we shouldn't be)
 	 */
 	private void getNextResult() throws AsynchronousFailureException {
 		ChunkResponse payload = (ChunkResponse) messagingGateway.receive();
 		if (payload != null) {
 			if (logger.isDebugEnabled()) {
-				logger.debug("Found result: "+payload);
+				logger.debug("Found result: " + payload);
 			}
 			Long jobInstanceId = payload.getJobId();
 			Assert.state(jobInstanceId != null, "Message did not contain job instance id.");
 			Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id ["
 					+ jobInstanceId + "] should have been [" + localState.getJobId() + "].");
-			localState.actual++;
-			localState.pushContribution(payload.getStepContribution());
+			localState.actual.incrementAndGet();
+			localState.pushStepContribution(payload.getStepContribution());
 			if (!payload.isSuccessful()) {
 				throw new AsynchronousFailureException("Failure or interrupt detected in handler: "
 						+ payload.getMessage());
@@ -215,8 +211,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	}
 
 	/**
-	 * Re-throws the original throwable if it is unchecked, wraps checked
-	 * exceptions into {@link AsynchronousFailureException}.
+	 * Re-throws the original throwable if it is unchecked, wraps checked exceptions into
+	 * {@link AsynchronousFailureException}.
 	 */
 	private static AsynchronousFailureException wrapIfNecessary(Throwable throwable) {
 		if (throwable instanceof Error) {
@@ -231,30 +227,40 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	}
 
 	private static class LocalState {
-		private long actual;
 
-		private long expected;
+		private AtomicInteger actual = new AtomicInteger();
+
+		private AtomicInteger expected = new AtomicInteger();
 
 		private StepExecution stepExecution;
 
 		private Queue<StepContribution> contributions = new LinkedBlockingQueue<StepContribution>();
 
-		public long getExpecting() {
-			return expected - actual;
+		public int getExpecting() {
+			return expected.get() - actual.get();
 		}
 
-		public Collection<StepContribution> pollContributions() {
+		public void open(int expectedValue, int actualValue) {
+			actual.set(actualValue);
+			expected.set(expectedValue);
+		}
+
+		public Collection<StepContribution> pollStepContributions() {
 			Collection<StepContribution> set = new ArrayList<StepContribution>();
-			StepContribution item = contributions.poll();
-			while (item != null) {
-				set.add(item);
-				item = contributions.poll();
+			synchronized (contributions) {
+				StepContribution item = contributions.poll();
+				while (item != null) {
+					set.add(item);
+					item = contributions.poll();
+				}
 			}
 			return set;
 		}
 
-		public void pushContribution(StepContribution stepContribution) {
-			contributions.add(stepContribution);
+		public void pushStepContribution(StepContribution stepContribution) {
+			synchronized (contributions) {
+				contributions.add(stepContribution);
+			}
 		}
 
 		public StepContribution createStepContribution() {
@@ -270,7 +276,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 		}
 
 		public void reset() {
-			expected = actual = 0;
+			expected.set(0);
+			actual.set(0);
 		}
 	}
 

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java

@@ -32,20 +32,26 @@ import org.springframework.integration.annotation.MessageEndpoint;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.util.Assert;
 
+/**
+ * A {@link ChunkHandler} based on a {@link ChunkProcessor}. Knows how to distinguish between a processor that is fault
+ * tolerant, and one that is not. If the processor is fault tolerant then exceptions can be propagated on the assumption
+ * that there will be a roll back and the request will be re-delivered.
+ * 
+ * @author Dave Syer
+ * 
+ * @param <S> the type of the items in the chunk to be handled
+ */
 @MessageEndpoint
-public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>,
-		InitializingBean {
+public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, InitializingBean {
 
-	private static final Log logger = LogFactory
-			.getLog(ChunkProcessorChunkHandler.class);
+	private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class);
 
 	private ChunkProcessor<S> chunkProcessor;
 
 	/*
 	 * (non-Javadoc)
 	 * 
-	 * @see
-	 * org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
 	 */
 	public void afterPropertiesSet() throws Exception {
 		Assert.notNull(chunkProcessor, "A ChunkProcessor must be provided");
@@ -54,8 +60,7 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>,
 	/**
 	 * Public setter for the {@link ChunkProcessor}.
 	 * 
-	 * @param chunkProcessor
-	 *            the chunkProcessor to set
+	 * @param chunkProcessor the chunkProcessor to set
 	 */
 	public void setChunkProcessor(ChunkProcessor<S> chunkProcessor) {
 		this.chunkProcessor = chunkProcessor;
@@ -66,8 +71,7 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>,
 	 * @see ChunkHandler#handleChunk(ChunkRequest)
 	 */
 	@ServiceActivator
-	public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest)
-			throws Exception {
+	public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) throws Exception {
 
 		logger.debug("Handling chunk: " + chunkRequest);
 
@@ -76,47 +80,48 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>,
 		Throwable failure = process(chunkRequest, stepContribution);
 		if (failure != null) {
 			logger.debug("Failed chunk", failure);
-			return new ChunkResponse(false, chunkRequest.getJobId(),
-					stepContribution, failure.getClass().getName() + ": "
-							+ failure.getMessage());
+			return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, failure.getClass().getName()
+					+ ": " + failure.getMessage());
 		}
 
 		logger.debug("Completed chunk handling with " + stepContribution);
-		return new ChunkResponse(true, chunkRequest.getJobId(),
-				stepContribution);
+		return new ChunkResponse(true, chunkRequest.getJobId(), stepContribution);
 
 	}
 
 	/**
-	 * @param chunkRequest
-	 *            the current request
-	 * @param stepContribution
-	 *            the step contribution to update
-	 * @throws Exception
-	 *             if there is a fatal exception
+	 * @param chunkRequest the current request
+	 * @param stepContribution the step contribution to update
+	 * @throws Exception if there is a fatal exception
 	 */
-	private Throwable process(ChunkRequest<S> chunkRequest,
-			StepContribution stepContribution) throws Exception {
+	private Throwable process(ChunkRequest<S> chunkRequest, StepContribution stepContribution) throws Exception {
 
 		Chunk<S> chunk = new Chunk<S>(chunkRequest.getItems());
 		Throwable failure = null;
 		try {
 			chunkProcessor.process(stepContribution, chunk);
-		} catch (SkipLimitExceededException e) {
+		}
+		catch (SkipLimitExceededException e) {
 			failure = e;
-		} catch (NonSkippableReadException e) {
+		}
+		catch (NonSkippableReadException e) {
 			failure = e;
-		} catch (SkipListenerFailedException e) {
+		}
+		catch (SkipListenerFailedException e) {
 			failure = e;
-		} catch (RetryException e) {
+		}
+		catch (RetryException e) {
 			failure = e;
-		} catch (JobInterruptedException e) {
+		}
+		catch (JobInterruptedException e) {
 			failure = e;
-		} catch (Exception e) {
+		}
+		catch (Exception e) {
 			if (chunkProcessor instanceof FaultTolerantChunkProcessor<?, ?>) {
 				// try again...
 				throw e;
-			} else {
+			}
+			else {
 				failure = e;
 			}
 		}
@@ -124,4 +129,5 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>,
 		return failure;
 
 	}
+
 }

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java

@@ -21,10 +21,19 @@ import java.util.Collection;
 
 import org.springframework.batch.core.StepContribution;
 
+/**
+ * Encapsulation of a chunk of items to be processed remotely as part of a step execution.
+ * 
+ * @author Dave Syer
+ * 
+ * @param <T> the type of the items to process
+ */
 public class ChunkRequest<T> implements Serializable {
 
 	private final Long jobId;
+
 	private final Collection<? extends T> items;
+
 	private final StepContribution stepContribution;
 
 	public ChunkRequest(Collection<? extends T> items, Long jobId, StepContribution stepContribution) {
@@ -40,20 +49,21 @@ public class ChunkRequest<T> implements Serializable {
 	public Collection<? extends T> getItems() {
 		return items;
 	}
-	
+
 	/**
 	 * @return the {@link StepContribution} for this chunk
 	 */
 	public StepContribution getStepContribution() {
 		return stepContribution;
 	}
-	
+
 	/**
 	 * @see java.lang.Object#toString()
 	 */
 	@Override
 	public String toString() {
-		return getClass().getSimpleName()+": jobId="+jobId+", contribution="+stepContribution+", item count="+items.size();
+		return getClass().getSimpleName() + ": jobId=" + jobId + ", contribution=" + stepContribution + ", item count="
+				+ items.size();
 	}
 
 }

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java

@@ -20,11 +20,20 @@ import java.io.Serializable;
 
 import org.springframework.batch.core.StepContribution;
 
+/**
+ * Encapsulates a response to processing a chunk of items, summarising the result as a {@link StepContribution}.
+ * 
+ * @author Dave Syer
+ * 
+ */
 public class ChunkResponse implements Serializable {
 
 	private final StepContribution stepContribution;
+
 	private final Long jobId;
+
 	private final boolean status;
+
 	private final String message;
 
 	public ChunkResponse(Long jobId, StepContribution stepContribution) {
@@ -63,7 +72,8 @@ public class ChunkResponse implements Serializable {
 	 */
 	@Override
 	public String toString() {
-		return getClass().getSimpleName()+": jobId="+jobId+", stepContribution="+stepContribution+", successful="+status;
+		return getClass().getSimpleName() + ": jobId=" + jobId + ", stepContribution=" + stepContribution
+				+ ", successful=" + status;
 	}
 
 }

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkHandlerFactoryBean.java

@@ -36,13 +36,12 @@ import org.springframework.util.Assert;
 import org.springframework.util.ReflectionUtils;
 
 /**
- * Convenient factory bean for a chunk handler that also converts an existing
- * chunk-oriented step into a remote chunk master. The idea is to lift the
- * existing chunk processor out of a step that works locally, and replace it
- * with a chunk writer that is already configured to write chunks into a message
- * channel. The existing step hands its business chunk processing responsibility
- * over to the handler produced by the factory, which then needs to be set up as
- * a remote worker on the other end of the channel the chunks are being sent to.
+ * Convenient factory bean for a chunk handler that also converts an existing chunk-oriented step into a remote chunk
+ * master. The idea is to lift the existing chunk processor out of a Step that works locally, and replace it with a one
+ * that writes chunks into a message channel. The existing step hands its business chunk processing responsibility over
+ * to the handler produced by the factory, which then needs to be set up as a worker on the other end of the channel the
+ * chunks are being sent to. Once this chunk handler is installed the application is playing the role of both the master
+ * and the slave listeners in the Remote Chunking pattern for the Step in question.
  * 
  * @author Dave Syer
  * 
@@ -58,6 +57,8 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
 	private StepContributionSource stepContributionSource;
 
 	/**
+	 * The local step that is to be converted to a remote chunk master.
+	 * 
 	 * @param step the step to set
 	 */
 	public void setStep(TaskletStep step) {
@@ -65,6 +66,9 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
 	}
 
 	/**
+	 * The item writer to be injected into the step. Its responsibility is to send chunks of items to remote workers.
+	 * Usually in practice it will be a {@link ChunkMessageChannelItemWriter}.
+	 * 
 	 * @param chunkWriter the chunk writer to set
 	 */
 	public void setChunkWriter(ItemWriter<T> chunkWriter) {
@@ -72,21 +76,39 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
 	}
 
 	/**
-	 * @param stepContributionSource the step contribution source to set
-	 * (defaults to the chunk writer)
+	 * A source of {@link StepContribution} instances coming back from remote workers.
+	 * 
+	 * @param stepContributionSource the step contribution source to set (defaults to the chunk writer)
 	 */
 	public void setStepContributionSource(StepContributionSource stepContributionSource) {
 		this.stepContributionSource = stepContributionSource;
 	}
 
+	/**
+	 * The type of object created by this factory. Returns {@link ChunkHandler} class.
+	 * 
+	 * @see FactoryBean#getObjectType()
+	 */
 	public Class<?> getObjectType() {
 		return ChunkHandler.class;
 	}
 
+	/**
+	 * Optimization for the bean facctory (always returns true).
+	 * 
+	 * @see FactoryBean#isSingleton()
+	 */
 	public boolean isSingleton() {
 		return true;
 	}
 
+	/**
+	 * Builds a {@link ChunkHandler} from the {@link ChunkProcessor} extracted from the {@link #setStep(TaskletStep)
+	 * step} provided. Also modifies the step to send chunks to the chunk handler via the
+	 * {@link #setChunkWriter(ItemWriter) chunk writer}.
+	 * 
+	 * @see FactoryBean#getObject()
+	 */
 	public ChunkHandler<T> getObject() throws Exception {
 
 		if (stepContributionSource == null) {
@@ -117,7 +139,8 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
 		ChunkProcessorChunkHandler<T> handler = new ChunkProcessorChunkHandler<T>();
 		setNonBuffering(chunkProcessor);
 		handler.setChunkProcessor(chunkProcessor);
-		// TODO: create step context for the processor in case it has scope="step" dependencies
+		// TODO: create step context for the processor in case it has
+		// scope="step" dependencies
 		handler.afterPropertiesSet();
 
 		return handler;
@@ -125,17 +148,22 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
 	}
 
 	/**
-	 * @param chunkProcessor
+	 * Overrides the buffering settings in the chunk processor if it is fault tolerant.
+	 * @param chunkProcessor the chunk processor that is going to be used in the workers
 	 */
 	private void setNonBuffering(ChunkProcessor<T> chunkProcessor) {
 		if (chunkProcessor instanceof FaultTolerantChunkProcessor<?, ?>) {
-			((FaultTolerantChunkProcessor<?, ?>)chunkProcessor).setBuffering(false);
+			((FaultTolerantChunkProcessor<?, ?>) chunkProcessor).setBuffering(false);
 		}
 	}
 
 	/**
-	 * @param tasklet
-	 * @param chunkWriter
+	 * Replace the chunk processor in the tasklet provided with one that can act as a master in the Remote Chunking
+	 * pattern.
+	 * 
+	 * @param tasklet a ChunkOrientedTasklet
+	 * @param chunkWriter an ItemWriter that can send the chunks to remote workers
+	 * @param stepContributionSource a StepContributionSource used to gather results from the workers
 	 */
 	private void replaceChunkProcessor(ChunkOrientedTasklet<?> tasklet, ItemWriter<T> chunkWriter,
 			final StepContributionSource stepContributionSource) {
@@ -152,10 +180,13 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
 	}
 
 	/**
-	 * @param contribution
-	 * @param chunkWriter
+	 * Update a StepContribution with all the data from a StepContributionSource. The filter and write conuts plus the
+	 * exit status will be updated to reflect the data in the source.
+	 * 
+	 * @param contribution the current contribution
+	 * @param stepContributionSource a source of StepContributions
 	 */
-	private void updateStepContribution(StepContribution contribution, StepContributionSource stepContributionSource) {
+	protected void updateStepContribution(StepContribution contribution, StepContributionSource stepContributionSource) {
 		for (StepContribution result : stepContributionSource.getStepContributions()) {
 			contribution.incrementFilterCount(result.getFilterCount());
 			contribution.incrementWriteCount(result.getWriteCount());
@@ -170,8 +201,9 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
 	}
 
 	/**
-	 * @param chunkProcessor
-	 * @return
+	 * Pull out an item writer from a ChunkProcessor
+	 * @param chunkProcessor a ChunkProcessor
+	 * @return its ItemWriter
 	 */
 	@SuppressWarnings("unchecked")
 	private ItemWriter<T> getItemWriter(ChunkProcessor<T> chunkProcessor) {
@@ -179,8 +211,9 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
 	}
 
 	/**
-	 * @param tasklet
-	 * @return
+	 * Pull the ChunkProcessor out of a tasklet.
+	 * @param tasklet a ChunkOrientedTasklet
+	 * @return the ChunkProcessor
 	 */
 	@SuppressWarnings("unchecked")
 	private ChunkProcessor<T> getChunkProcessor(ChunkOrientedTasklet<?> tasklet) {
@@ -188,11 +221,12 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
 	}
 
 	/**
-	 * @param bean
-	 * @return
+	 * Pull a Tasklet out of a step.
+	 * @param step a TaskletStep
+	 * @return the Tasklet
 	 */
-	private Tasklet getTasklet(TaskletStep bean) {
-		return (Tasklet) getField(bean, "tasklet");
+	private Tasklet getTasklet(TaskletStep step) {
+		return (Tasklet) getField(step, "tasklet");
 	}
 
 	private static Object getField(Object target, String name) {

spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/StepContributionSource.java

@@ -19,13 +19,23 @@ package org.springframework.batch.integration.chunk;
 import java.util.Collection;
 
 import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.StepExecution;
 
 /**
+ * A source of {@link StepContribution} instances that can be aggregated and used to update an ongoing
+ * {@link StepExecution}.
+ * 
  * @author Dave Syer
- *
+ * 
  */
 public interface StepContributionSource {
 
+	/**
+	 * Get the currently available contributions and drain the source. The next call would return an empty collection,
+	 * unless new contributions have arrived.
+	 * 
+	 * @return a collection of {@link StepContribution} instances
+	 */
 	Collection<StepContribution> getStepContributions();
 
 }