REOPENED - issue BATCH-511: read failures cause rollback

Add SkipLimitExceededException to the list of fatal exceptions for the step exception handler

spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java

@@ -15,7 +15,6 @@
  */
 package org.springframework.batch.core;
 
-
 /**
  * Represents a contribution to a {@link StepExecution}, buffering changes
  * until they can be applied at a chunk boundary.
@@ -84,12 +83,31 @@ public class StepContribution {
 	 * <code>StepContribution</code> (not including skips accumulated in the
 	 * parent {@link StepExecution}.
 	 */
-	public int getContributionSkipCount() {
+	public int getSkipCount() {
 		return skipCount;
 	}
 
+	/**
+	 * Increment the skip count
+	 */
 	public void incrementSkipCount() {
 		skipCount++;
 	}
 
+	/**
+	 * Increment the skip count by a non-trivial amount
+	 * @param delta
+	 */
+	public void incrementSkipCount(int delta) {
+		skipCount += delta;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Object#toString()
+	 */
+	public String toString() {
+		return "[StepContribution: items="+itemCount+", commits=" + commitCount + ", skips=" + skipCount + "]";
+	}
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java

@@ -307,7 +307,7 @@ public class StepExecution extends Entity {
 		// TODO: this should not be necessary - the step decides
 		// executionContext = contribution.getExecutionContext();
 		commitCount += contribution.getCommitCount();
-		skipCount += contribution.getContributionSkipCount();
+		skipCount += contribution.getSkipCount();
 	}
 
 	/**

spring-batch-core/src/main/java/org/springframework/batch/core/listener/SkipListenerSupport.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2006-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.core.listener;
+
+import org.springframework.batch.core.SkipListener;
+
+/**
+ * Basic no-op implementations of all {@link SkipListener} implementations.
+ * 
+ * @author Dave Syer
+ *
+ */
+public class SkipListenerSupport implements SkipListener {
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.SkipListener#onSkipInRead(java.lang.Throwable)
+	 */
+	public void onSkipInRead(Throwable t) {
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.SkipListener#onSkipInWrite(java.lang.Object, java.lang.Throwable)
+	 */
+	public void onSkipInWrite(Object item, Throwable t) {
+	}
+
+
+}

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

@@ -330,12 +330,12 @@ public class ItemOrientedStep extends AbstractStep {
 
 					}
 					catch (Error e) {
-						stepExecution.incrementSkipCountBy(contribution.getContributionSkipCount());
+						stepExecution.incrementSkipCountBy(contribution.getSkipCount());
 						processRollback(stepExecution, fatalException, transaction);
 						throw e;
 					}
 					catch (Exception e) {
-						stepExecution.incrementSkipCountBy(contribution.getContributionSkipCount());
+						stepExecution.incrementSkipCountBy(contribution.getSkipCount());
 						processRollback(stepExecution, fatalException, transaction);
 						throw e;
 					}

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

@@ -1,9 +1,12 @@
 package org.springframework.batch.core.step.item;
 
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
 
 import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
+import org.springframework.batch.core.step.skip.SkipLimitExceededException;
 import org.springframework.batch.item.ItemKeyGenerator;
 import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler;
 
@@ -92,14 +95,24 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 			 * to absorb exceptions at the step level because the failed items
 			 * will never re-appear after a rollback.
 			 */
-			itemHandler.setItemSkipPolicy(new LimitCheckingItemSkipPolicy(skipLimit, Arrays
-					.asList(skippableExceptionClasses), Arrays.asList(fatalExceptionClasses)));
-			SimpleLimitExceptionHandler exceptionHandler = new SimpleLimitExceptionHandler();
-			exceptionHandler.setLimit(skipLimit);
+			List fatalExceptionList = new ArrayList(Arrays.asList(fatalExceptionClasses));
+			if (!fatalExceptionList.contains(SkipLimitExceededException.class)) {
+				fatalExceptionList.add(SkipLimitExceededException.class);
+			}
+			fatalExceptionClasses = (Class[]) fatalExceptionList.toArray(new Class[0]);
+			
+			LimitCheckingItemSkipPolicy skipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, Arrays
+					.asList(skippableExceptionClasses), fatalExceptionList);
+			itemHandler.setItemSkipPolicy(skipPolicy);
+			SimpleLimitExceptionHandler exceptionHandler = new SimpleLimitExceptionHandler(skipLimit);
 			exceptionHandler.setExceptionClasses(skippableExceptionClasses);
 			exceptionHandler.setFatalExceptionClasses(fatalExceptionClasses);
+
+			getStepOperations().setExceptionHandler(exceptionHandler);
+
+			// for subclass to pick up limit and exception classes
 			setExceptionHandler(exceptionHandler);
-			getStepOperations().setExceptionHandler(getExceptionHandler());
+
 			itemHandler.setItemKeyGenerator(itemKeyGenerator);
 
 			BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();