BACH-1100: Eclipse / SVN weirdness

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java

@@ -16,7 +16,9 @@
 
 package org.springframework.batch.core.step.item;
 
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -25,6 +27,7 @@ import org.springframework.batch.classify.Classifier;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NonSkippableProcessException;
+import org.springframework.batch.core.step.skip.SkipListenerFailedException;
 import org.springframework.batch.core.step.skip.SkipPolicy;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemWriter;
@@ -48,18 +51,60 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 	private boolean buffering = true;
 
+	private KeyGenerator keyGenerator;
+
+	private ChunkMonitor chunkMonitor = new ChunkMonitor();
+
+	/**
+	 * The {@link KeyGenerator} to use to identify failed items across rollback.
+	 * Not used in the case of the {@link #setBuffering(boolean) buffering flag}
+	 * being true (the default).
+	 * 
+	 * @param keyGenerator the {@link KeyGenerator} to set
+	 */
+	public void setKeyGenerator(KeyGenerator keyGenerator) {
+		this.keyGenerator = keyGenerator;
+	}
+
+	/**
+	 * @param SkipPolicy the {@link SkipPolicy} for item processing
+	 */
 	public void setProcessSkipPolicy(SkipPolicy SkipPolicy) {
 		this.itemProcessSkipPolicy = SkipPolicy;
 	}
 
+	/**
+	 * @param SkipPolicy the {@link SkipPolicy} for item writing
+	 */
 	public void setWriteSkipPolicy(SkipPolicy SkipPolicy) {
 		this.itemWriteSkipPolicy = SkipPolicy;
 	}
 
+	/**
+	 * A classifier that can distinguish between exceptions that cause rollback
+	 * (return true) or not (return false).
+	 * 
+	 * @param rollbackClassifier
+	 */
 	public void setRollbackClassifier(Classifier<Throwable, Boolean> rollbackClassifier) {
 		this.rollbackClassifier = rollbackClassifier;
 	}
 
+	/**
+	 * @param chunkMonitor
+	 */
+	public void setChunkMonitor(ChunkMonitor chunkMonitor) {
+		this.chunkMonitor = chunkMonitor;
+	}
+
+	/**
+	 * A flag to indicate that items have been buffered and therefore will
+	 * always come back as a chunk after a rollback. Otherwise things are more
+	 * complicated because after a rollback the new chunk might or moght not
+	 * contain items from the previous failed chunk.
+	 * 
+	 * @param buffering
+	 */
 	public void setBuffering(boolean buffering) {
 		this.buffering = buffering;
 	}
@@ -133,9 +178,8 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 			};
 
-			// TODO: is it OK to use the item as a key for the retry state?
-			O output = batchRetryTemplate.execute(retryCallback, recoveryCallback, new DefaultRetryState(item,
-					rollbackClassifier));
+			O output = batchRetryTemplate.execute(retryCallback, recoveryCallback, new DefaultRetryState(
+					getInputKey(item), rollbackClassifier));
 			if (output != null) {
 				outputs.add(output);
 			}
@@ -152,99 +196,143 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 		RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
 			public Object doWithRetry(RetryContext context) throws Exception {
-				doWrite(outputs.getItems());
-				contribution.incrementWriteCount(outputs.size());
+
+				if (!inputs.isBusy()) {
+					chunkMonitor.setChunkSize(inputs.size());
+					doWrite(outputs.getItems());
+					contribution.incrementWriteCount(outputs.size());
+				}
+				else {
+					scan(contribution, inputs, outputs, chunkMonitor);
+				}
 				return null;
+
 			}
 		};
 
-		RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
+		if (!buffering) {
 
-			public Object recover(RetryContext context) throws Exception {
+			RecoveryCallback<Object> batchRecoveryCallback = new RecoveryCallback<Object>() {
 
-				Exception le = (Exception) context.getLastThrowable();
-				if (outputs.size() > 1 && !rollbackClassifier.classify(le)) {
-					throw new RetryException("Invalid retry state during write caused by "
-							+ "exception that does not classify for rollback: ", le);
-				}
+				public Object recover(RetryContext context) throws Exception {
 
-				boolean singleton = outputs.size() == 1;
-
-				Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
-				for (Chunk<O>.ChunkIterator outputIterator = outputs.iterator(); outputIterator.hasNext();) {
-
-					inputIterator.next();
-					O item = outputIterator.next();
-					if (singleton) {
-						checkSkipPolicy(inputIterator, outputIterator, le, contribution);
-						return null;
+					Exception e = (Exception) context.getLastThrowable();
+					if (outputs.size() > 1 && !rollbackClassifier.classify(e)) {
+						throw new RetryException("Invalid retry state during write caused by "
+								+ "exception that does not classify for rollback: ", e);
 					}
 
-					try {
-						writeItems(Collections.singletonList(item));
-					}
-					catch (Exception e) {
+					Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
+					for (Chunk<O>.ChunkIterator outputIterator = outputs.iterator(); outputIterator.hasNext();) {
+
+						inputIterator.next();
+						outputIterator.next();
+
 						checkSkipPolicy(inputIterator, outputIterator, e, contribution);
-						if (rollbackClassifier.classify(e)) {
-							throw e;
-						}
-						else {
+						if (!rollbackClassifier.classify(e)) {
 							throw new RetryException(
 									"Invalid retry state during recovery caused by exception that does not classify for rollback: ",
 									e);
 						}
+
 					}
 
-				}
-
-				doAfterWrite(outputs.getItems());
-				contribution.incrementWriteCount(outputs.size());
-				return null;
-
-			}
-
-		};
-
-		RecoveryCallback<Object> batchRecoveryCallback = new RecoveryCallback<Object>() {
-
-			public Object recover(RetryContext context) throws Exception {
-
-				Exception e = (Exception) context.getLastThrowable();
-				if (outputs.size() > 1 && !rollbackClassifier.classify(e)) {
-					throw new RetryException("Invalid retry state during write caused by "
-							+ "exception that does not classify for rollback: ", e);
-				}
-
-				Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
-				for (Chunk<O>.ChunkIterator outputIterator = outputs.iterator(); outputIterator.hasNext();) {
-
-					inputIterator.next();
-					outputIterator.next();
-
-					checkSkipPolicy(inputIterator, outputIterator, e, contribution);
-					if (!rollbackClassifier.classify(e)) {
-						throw new RetryException(
-								"Invalid retry state during recovery caused by exception that does not classify for rollback: ",
-								e);
-					}
+					return null;
 
 				}
 
-				return null;
+			};
 
-			}
+			batchRetryTemplate.execute(retryCallback, batchRecoveryCallback, BatchRetryTemplate.createState(
+					getInputKeys(inputs), rollbackClassifier));
 
-		};
-
-		if (!buffering) {
-			batchRetryTemplate.execute(retryCallback, batchRecoveryCallback, BatchRetryTemplate.createState(inputs
-					.getItems(), rollbackClassifier));
 		}
 		else {
+
+			RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
+
+				public Object recover(RetryContext context) throws Exception {
+
+					Exception le = (Exception) context.getLastThrowable();
+					if (outputs.size() > 1 && !rollbackClassifier.classify(le)) {
+						throw new RetryException("Invalid retry state during write caused by "
+								+ "exception that does not classify for rollback: ", le);
+					}
+
+					boolean singleton = outputs.size() == 1;
+
+					if (singleton && !inputs.isBusy()) {
+						Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
+						Chunk<O>.ChunkIterator outputIterator = outputs.iterator();
+						checkSkipPolicy(inputIterator, outputIterator, le, contribution);
+						return null;
+					}
+
+					inputs.setBusy(true);
+					scan(contribution, inputs, outputs, chunkMonitor);
+					return null;
+
+				}
+
+			};
+
 			batchRetryTemplate.execute(retryCallback, recoveryCallback, new DefaultRetryState(inputs,
 					rollbackClassifier));
+
+		}
+		
+		callSkipListeners(inputs, outputs);
+
+	}
+
+	private void callSkipListeners(final Chunk<I> inputs, final Chunk<O> outputs) {
+
+		for (SkipWrapper<I> wrapper : inputs.getSkips()) {
+			I item = wrapper.getItem();
+			if (item == null) {
+				continue;
+			}
+			Exception e = wrapper.getException();
+			try {
+				getListener().onSkipInProcess(item, e);
+			}
+			catch (RuntimeException ex) {
+				throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
+			}
 		}
 
+		for (SkipWrapper<O> wrapper : outputs.getSkips()) {
+			Exception e = wrapper.getException();
+			try {
+				getListener().onSkipInWrite(wrapper.getItem(), e);
+			}
+			catch (RuntimeException ex) {
+				throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
+			}
+		}
+		
+		// Clear skips if we are possibly going to process this chunk again	
+		outputs.clearSkips();
+		inputs.clearSkips();
+
+	}
+
+	private Object getInputKey(I item) {
+		if (keyGenerator == null) {
+			return item;
+		}
+		return keyGenerator.getKey(item);
+	}
+
+	private List<?> getInputKeys(final Chunk<I> inputs) {
+		if (keyGenerator == null) {
+			return inputs.getItems();
+		}
+		List<Object> keys = new ArrayList<Object>();
+		for (I item : inputs.getItems()) {
+			keys.add(keyGenerator.getKey(item));
+		}
+		return keys;
 	}
 
 	private void checkSkipPolicy(Chunk<I>.ChunkIterator inputIterator, Chunk<O>.ChunkIterator outputIterator,
@@ -260,4 +348,43 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 		}
 	}
 
+	private void scan(final StepContribution contribution, final Chunk<I> inputs, final Chunk<O> outputs, ChunkMonitor chunkMonitor)
+			throws Exception {
+
+		if (outputs.isEmpty()) {
+			inputs.setBusy(false);
+			return;
+		}
+
+		Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
+		Chunk<O>.ChunkIterator outputIterator = outputs.iterator();
+
+		List<O> items = Collections.singletonList(outputIterator.next());
+		try {
+			writeItems(items);
+		}
+		catch (Exception e) {
+			checkSkipPolicy(inputIterator, outputIterator, e, contribution);
+			if (rollbackClassifier.classify(e)) {
+				throw e;
+			}
+			else {
+				throw new RetryException(
+						"Invalid retry state during recovery caused by exception that does not classify for rollback: ",
+						e);
+			}
+		}
+		// If successful we are going to return and allow
+		// the driver to commit...
+		doAfterWrite(items);
+		contribution.incrementWriteCount(1);
+		inputIterator.remove();
+		outputIterator.remove();
+		chunkMonitor.incrementOffset();
+		if (outputs.isEmpty()) {
+			inputs.setBusy(false);
+			chunkMonitor.resetOffset();
+		}
+	}
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java

@@ -19,6 +19,7 @@ package org.springframework.batch.core.step.item;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NonSkippableReadException;
+import org.springframework.batch.core.step.skip.SkipListenerFailedException;
 import org.springframework.batch.core.step.skip.SkipPolicy;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.repeat.RepeatOperations;
@@ -31,6 +32,10 @@ public class FaultTolerantChunkProvider<I> extends SimpleChunkProvider<I> {
 		super(itemReader, repeatOperations);
 	}
 
+	/**
+	 * The policy that determines whether exceptions can be skipped on read.
+	 * @param SkipPolicy
+	 */
 	public void setSkipPolicy(SkipPolicy SkipPolicy) {
 		this.skipPolicy = SkipPolicy;
 	}
@@ -58,4 +63,16 @@ public class FaultTolerantChunkProvider<I> extends SimpleChunkProvider<I> {
 		}
 	}
 
+	@Override
+	public void postProcess(StepContribution contribution, Chunk<I> chunk) {
+		for (Exception e : chunk.getErrors()) {
+			try {
+				getListener().onSkipInRead(e);
+			}
+			catch (RuntimeException ex) {
+				throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
+			}
+		}
+	}
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java

@@ -29,6 +29,9 @@ import org.springframework.batch.core.step.skip.SkipLimitExceededException;
 import org.springframework.batch.core.step.skip.SkipListenerFailedException;
 import org.springframework.batch.core.step.skip.SkipPolicy;
 import org.springframework.batch.core.step.tasklet.TaskletStep;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.batch.item.support.CompositeItemStream;
 import org.springframework.batch.repeat.RepeatOperations;
 import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.retry.RetryException;
@@ -40,6 +43,8 @@ import org.springframework.batch.retry.policy.MapRetryContextCache;
 import org.springframework.batch.retry.policy.NeverRetryPolicy;
 import org.springframework.batch.retry.policy.RetryContextCache;
 import org.springframework.batch.retry.policy.SimpleRetryPolicy;
+import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
 
 /**
  * Factory bean for step that provides options for configuring skip behaviour.
@@ -88,6 +93,22 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	private RetryContextCache retryContextCache;
 
+	private KeyGenerator keyGenerator;
+
+	private ChunkMonitor chunkMonitor = new ChunkMonitor();
+
+	/**
+	 * The {@link KeyGenerator} to use to identify failed items across rollback.
+	 * Not used in the case of the
+	 * {@link #setIsReaderTransactionalQueue(boolean) transactional queue flag}
+	 * being false (the default).
+	 * 
+	 * @param keyGenerator the {@link KeyGenerator} to set
+	 */
+	public void setKeyGenerator(KeyGenerator keyGenerator) {
+		this.keyGenerator = keyGenerator;
+	}
+
 	/**
 	 * Setter for the retry policy. If this is specified the other retry
 	 * properties are ignored (retryLimit, backOffPolicy,
@@ -207,27 +228,61 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 		super.applyConfiguration(step);
 
 	}
-	
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	protected void registerStreams(TaskletStep step, ItemStream[] streams) {
+		CompositeItemStream composite = new CompositeItemStream();
+		boolean streamIsReader = false;
+		for (final ItemStream stream : streams) {
+			if (stream instanceof ItemReader) {
+				streamIsReader = true;
+				composite.register(stream);
+			}
+			else {
+				step.registerStream(stream);
+			}
+		}
+		TaskExecutor taskExecutor = getTaskExecutor();
+		// In cases where multiple nested item readers are registered,
+		// they all want to get the open() and close() callbacks.
+		if (streamIsReader) {
+			chunkMonitor.setItemStream(composite);
+			step.registerStream(chunkMonitor);
+			boolean concurrent = taskExecutor != null && !(taskExecutor instanceof SyncTaskExecutor);
+			if (!concurrent) {
+				chunkMonitor.setItemReader(getItemReader());
+			} else {
+				logger.warn("Synchronous TaskExecutor detected (" + taskExecutor.getClass()
+						+ ") with ItemStream reader.  This is probably an error, "
+						+ "and may lead to incorrect restart data being stored.");
+			}
+		}
+	}
+
 	/**
 	 * @return {@link ChunkProvider} configured for fault-tolerance.
 	 */
 	@Override
-	protected FaultTolerantChunkProvider<T> configureChunkProvider() {
-		
+	protected SimpleChunkProvider<T> configureChunkProvider() {
+
 		SkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, skippableExceptionClasses,
 				fatalExceptionClasses);
 		FaultTolerantChunkProvider<T> chunkProvider = new FaultTolerantChunkProvider<T>(getItemReader(),
 				getChunkOperations());
 		chunkProvider.setSkipPolicy(readSkipPolicy);
-		
+
 		return chunkProvider;
+
 	}
 
 	/**
 	 * @return {@link ChunkProcessor} configured for fault-tolerance.
 	 */
 	@Override
-	protected FaultTolerantChunkProcessor<T, S> configureChunkProcessor() {
+	protected SimpleChunkProcessor<T, S> configureChunkProcessor() {
 
 		SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, skippableExceptionClasses,
 				fatalExceptionClasses);
@@ -246,10 +301,12 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 		chunkProcessor.setWriteSkipPolicy(writeSkipPolicy);
 		chunkProcessor.setProcessSkipPolicy(writeSkipPolicy);
 		chunkProcessor.setRollbackClassifier(rollbackClassifier);
+		chunkProcessor.setKeyGenerator(keyGenerator);
+		chunkProcessor.setChunkMonitor(chunkMonitor);
 
 		return chunkProcessor;
-	}
 
+	}
 
 	/**
 	 * @return fully configured retry template for item processing phase.

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java

@@ -21,7 +21,6 @@ import java.util.List;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.listener.MulticasterBatchListener;
-import org.springframework.batch.core.step.skip.SkipListenerFailedException;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.beans.factory.InitializingBean;
@@ -94,6 +93,13 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
 		this.listener.register(listener);
 	}
 
+	/**
+	 * @return the listener
+	 */
+	protected MulticasterBatchListener<I, O> getListener() {
+		return listener;
+	}
+
 	/**
 	 * @param item the input item
 	 * @return the result of the processing
@@ -122,21 +128,20 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
 		try {
 			listener.beforeWrite(items);
 			writeItems(items);
-			listener.afterWrite(items);
+			doAfterWrite(items);
 		}
 		catch (Exception e) {
 			listener.onWriteError(e, items);
 			throw e;
 		}
 	}
-	
+
 	/**
 	 * Call the listener's after write method.
 	 * 
 	 * @param items
 	 */
-	protected final void doAfterWrite(List<O> items)
-	{
+	protected final void doAfterWrite(List<O> items) {
 		listener.afterWrite(items);
 	}
 
@@ -146,17 +151,6 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
 
 	public final void process(StepContribution contribution, Chunk<I> inputs) throws Exception {
 
-		// If there is no input we don't have to do anything more
-		if (inputs.isEmpty()) {
-			return;
-		}
-
-		int inputSize = inputs.size();
-		
-		Chunk<O> outputs = transform(contribution, inputs);
-		
-		contribution.incrementFilterCount(inputSize - outputs.size());
-
 		/*
 		 * Need to remember the write skips across transactions, otherwise they
 		 * keep coming back. Since we register skips with the inputs they will
@@ -172,37 +166,37 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
 			skips = new Chunk<O>();
 		}
 
+		// If there is no input we don't have to do anything more
+		if (inputs.isEmpty() && skips.getSkips().isEmpty()) {
+			return;
+		}
+
+		int inputsSize = inputs.size();
+
+		Chunk<O> outputs = transform(contribution, inputs);
+
+		contribution.incrementFilterCount(inputsSize  - outputs.size());
+
 		outputs = new Chunk<O>(outputs.getItems(), skips.getSkips());
+
+		// Remember for next time if there are skips accumulating
 		inputs.setUserData(outputs);
 
 		write(contribution, inputs, outputs);
 
-		for (SkipWrapper<I> wrapper : inputs.getSkips()) {
-			I item = wrapper.getItem();
-			if (item == null) {
-				continue;
-			}
-			Exception e = wrapper.getException();
-			try {
-				listener.onSkipInProcess(item, e);
-			}
-			catch (RuntimeException ex) {
-				throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
-			}
-		}
-
-		for (SkipWrapper<O> wrapper : outputs.getSkips()) {
-			Exception e = wrapper.getException();
-			try {
-				listener.onSkipInWrite(wrapper.getItem(), e);
-			}
-			catch (RuntimeException ex) {
-				throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
-			}
-		}
-
 	}
 
+	/**
+	 * Simple implementation delegates to the {@link #doWrite(List)} method and
+	 * increments the write count in the contribution. Subclasses can handle
+	 * more complicated scenarios, e.g.with fault tolerance. If output items are
+	 * skipped they should be removed from the inputs as well.
+	 * 
+	 * @param contribution the current step contribution
+	 * @param inputs the inputs that gave rise to the ouputs
+	 * @param outputs the outputs to write
+	 * @throws Exception if there is a problem
+	 */
 	protected void write(StepContribution contribution, Chunk<I> inputs, Chunk<O> outputs) throws Exception {
 		doWrite(outputs.getItems());
 		contribution.incrementWriteCount(outputs.size());

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java

@@ -23,7 +23,6 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.listener.MulticasterBatchListener;
-import org.springframework.batch.core.step.skip.SkipListenerFailedException;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.repeat.RepeatCallback;
 import org.springframework.batch.repeat.RepeatContext;
@@ -71,6 +70,13 @@ public class SimpleChunkProvider<I> implements ChunkProvider<I> {
 	public void registerListener(StepListener listener) {
 		this.listener.register(listener);
 	}
+	
+	/**
+	 * @return the listener
+	 */
+	protected MulticasterBatchListener<I, ?> getListener() {
+		return listener;
+	}
 
 	/**
 	 * Surrounds the read call with listener callbacks.
@@ -113,14 +119,7 @@ public class SimpleChunkProvider<I> implements ChunkProvider<I> {
 	}
 	
 	public void postProcess(StepContribution contribution, Chunk<I> chunk) {
-		for (Exception e : chunk.getErrors()) {
-			try {
-				listener.onSkipInRead(e);
-			}
-			catch (RuntimeException ex) {
-				throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
-			}
-		}		
+		// do nothing
 	}
 
 	protected I read(StepContribution contribution, Chunk<I> chunk) throws Exception {

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java

@@ -407,6 +407,14 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
 	public void setTaskExecutor(TaskExecutor taskExecutor) {
 		this.taskExecutor = taskExecutor;
 	}
+	
+	/**
+	 * Mkae the {@link TaskExecutor} available to subclasses
+	 * @return the taskExecutor to be used to execute chunks
+	 */
+	protected TaskExecutor getTaskExecutor() {
+		return taskExecutor;
+	}
 
 	/**
 	 * Public setter for the throttle limit. This limits the number of tasks
@@ -437,7 +445,7 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
 		step.setStartLimit(startLimit);
 		step.setAllowStartIfComplete(allowStartIfComplete);
 
-		step.setStreams(streams);
+		registerStreams(step, streams);
 
 		if (chunkOperations == null) {
 			RepeatTemplate repeatTemplate = new RepeatTemplate();
@@ -468,6 +476,7 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
 
 		registerItemListeners(chunkProvider, chunkProcessor);
 		registerStepListeners(step, chunkOperations);
+		registerStreams(step, itemReader, itemProcessor, itemWriter);
 
 		ChunkOrientedTasklet<T> tasklet = new ChunkOrientedTasklet<T>(chunkProvider, chunkProcessor);
 		tasklet.setBuffering(!isReaderTransactionalQueue());
@@ -476,6 +485,15 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
 
 	}
 
+	/**
+	 * Register the streams with the step.
+	 * @param step the {@link TaskletStep}
+	 * @param streams the streams to register
+	 */
+	protected void registerStreams(TaskletStep step, ItemStream[] streams) {
+		step.setStreams(streams);
+	}
+
 	/**
 	 * Extension point for creating appropriate {@link ChunkProvider}. Return
 	 * value must subclass {@link SimpleChunkProvider} due to listener
@@ -512,16 +530,21 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
 		}
 		return new SimpleCompletionPolicy(commitInterval);
 	}
+	
+	private void registerStreams(TaskletStep step, ItemReader<? extends T> itemReader, ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter) {
+		for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) {
+			if (itemHandler instanceof ItemStream) {
+				registerStreams(step, new ItemStream[] {(ItemStream) itemHandler});
+			}
+		}		
+	}
 
 	/**
 	 * Register listeners with step and chunk.
 	 */
 	private void registerStepListeners(TaskletStep step, RepeatOperations chunkOperations) {
 
-		for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) {
-			if (itemHandler instanceof ItemStream) {
-				step.registerStream((ItemStream) itemHandler);
-			}
+		for (Object itemHandler : new Object[] { getItemReader(), itemWriter, itemProcessor }) {
 			if (StepListenerFactoryBean.isListener(itemHandler)) {
 				StepListener listener = StepListenerFactoryBean.getListener(itemHandler);
 				if (listener instanceof StepExecutionListener) {