IN PROGRESS - BATCH-929: Deferrable Constraints cause unrecoverable errors

use skip policy for commit failure (rather than just classifier) to avoid infinite loops

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

@@ -48,6 +48,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	private int skipLimit = 0;
 
+	private Collection<Class<? extends Throwable>> nonFatalCommitExceptionClasses = new HashSet<Class<? extends Throwable>>();
+
 	private Collection<Class<? extends Throwable>> skippableExceptionClasses = new HashSet<Class<? extends Throwable>>();
 
 	private Collection<Class<? extends Throwable>> fatalExceptionClasses = new HashSet<Class<? extends Throwable>>();
@@ -77,6 +79,17 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 		this.isReaderTransactionalQueue = isReaderTransactionalQueue;
 	}
 
+	/**
+	 * Exceptions that will not cause UNKNOWN status of the step if they occur
+	 * on commit. The assumption is that failed commit with one of the supplied
+	 * exceptions is equivalent to rollback (and therefore need not cause
+	 * failure if skippable).
+	 * @param nonFatalCommitExceptionClasses
+	 */
+	public void setNonFatalCommitExceptionClasses(Collection<Class<? extends Throwable>> nonFatalCommitExceptionClasses) {
+		this.nonFatalCommitExceptionClasses = nonFatalCommitExceptionClasses;
+	}
+
 	/**
 	 * Setter for the retry policy. If this is specified the other retry
 	 * properties are ignored (retryLimit, backOffPolicy,
@@ -279,7 +292,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 				step.setTasklet(tasklet);
 			}
-
+			ItemSkipPolicy commitSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit,
+					nonFatalCommitExceptionClasses, fatalExceptionClasses);
+			step.setCommitSkipPolicy(commitSkipPolicy);
 		}
 
 	}

spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java

@@ -31,6 +31,8 @@ import org.springframework.batch.core.scope.StepContextRepeatCallback;
 import org.springframework.batch.core.step.AbstractStep;
 import org.springframework.batch.core.step.StepInterruptionPolicy;
 import org.springframework.batch.core.step.ThreadStepInterruptionPolicy;
+import org.springframework.batch.core.step.skip.ItemSkipPolicy;
+import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
@@ -40,7 +42,6 @@ import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.RepeatOperations;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.batch.repeat.support.RepeatTemplate;
-import org.springframework.batch.support.Classifier;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
@@ -90,13 +91,7 @@ public class TaskletStep extends AbstractStep {
 
 	private Semaphore semaphore = new Semaphore(1);
 
-	private Classifier<Exception, Boolean> nonFatalCommitExceptions = new Classifier<Exception, Boolean>() {
-
-		public Boolean classify(Exception classifiable) {
-			return false;
-		}
-
-	};
+	private ItemSkipPolicy commitSkipPolicy = new NeverSkipItemSkipPolicy();
 
 	/**
 	 * Default constructor.
@@ -113,11 +108,10 @@ public class TaskletStep extends AbstractStep {
 	}
 
 	/**
-	 * @param nonFatalCommitExceptions classifies whether commit exception is
-	 * fatal or not.
+	 * Skip policy applying to exception thrown on tx commit.
 	 */
-	public void setNonFatalCommitExceptions(Classifier<Exception, Boolean> nonFatalCommitExceptions) {
-		this.nonFatalCommitExceptions = nonFatalCommitExceptions;
+	public void setCommitSkipPolicy(ItemSkipPolicy commitSkipPolicy) {
+		this.commitSkipPolicy = commitSkipPolicy;
 	}
 
 	/**
@@ -296,7 +290,7 @@ public class TaskletStep extends AbstractStep {
 						transactionManager.commit(transaction);
 					}
 					catch (Exception e) {
-						if (nonFatalCommitExceptions.classify(e)) {
+						if (commitSkipPolicy.shouldSkip(e, stepExecution.getSkipCount())) {
 							rollbackExecutionContext(stepExecution);
 							throw new CommitException("non-fatal commit failure", e);
 						}

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java

@@ -525,6 +525,10 @@ public class FaultTolerantStepFactoryBeanTests {
 
 	}
 
+	@Test
+	public void testNonFatalCommitFailure() throws Exception {
+		//TODO
+	}
 	private static class SkipProcessorStub implements ItemProcessor<String, String> {
 
 		private final Collection<String> failures;

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java

@@ -3,8 +3,13 @@
  */
 package org.springframework.batch.core.step.item;
 
-import static org.junit.Assert.*;
-import static org.springframework.batch.core.BatchStatus.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.springframework.batch.core.BatchStatus.COMPLETED;
+import static org.springframework.batch.core.BatchStatus.FAILED;
+import static org.springframework.batch.core.BatchStatus.STOPPED;
+import static org.springframework.batch.core.BatchStatus.UNKNOWN;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -20,6 +25,7 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep
 import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.JobRestartException;
+import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy;
 import org.springframework.batch.core.step.tasklet.Tasklet;
 import org.springframework.batch.core.step.tasklet.TaskletStep;
 import org.springframework.batch.item.ExecutionContext;
@@ -27,7 +33,6 @@ import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
 import org.springframework.batch.item.ItemStreamSupport;
 import org.springframework.batch.repeat.RepeatStatus;
-import org.springframework.batch.support.Classifier;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 import org.springframework.core.AttributeAccessor;
 import org.springframework.transaction.TransactionException;
@@ -215,12 +220,8 @@ public class TaskletStepExceptionTests {
 		taskletStep.registerStream(stream);
 
 		final RuntimeException commitException = new RuntimeException();
-		taskletStep.setNonFatalCommitExceptions(new Classifier<Exception, Boolean>() {
-
-			public Boolean classify(Exception classifiable) {
-				return true;
-			}
-		});
+		taskletStep.setCommitSkipPolicy(new AlwaysSkipItemSkipPolicy());
+		
 		taskletStep.setTransactionManager(new ResourcelessTransactionManager() {
 			@Override
 			protected void doCommit(DefaultTransactionStatus status) throws TransactionException {