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

allow commit failure to cause execution context rollback instead of step failure with UNKNOWN status

Author: robokaso

spring-batch-core/src/main/java/org/springframework/batch/core/repository/JobRepository.java

@@ -39,11 +39,13 @@
 public interface JobRepository {
 
 	/**
-	 * Check if an instance of this job already exists with the parameters provided.
+	 * Check if an instance of this job already exists with the parameters
+	 * provided.
 	 * 
 	 * @param jobName the name of the job
 	 * @param jobParameters the parameters to match
-	 * @return true if a {@link JobInstance} already exists for this job name and job parameters
+	 * @return true if a {@link JobInstance} already exists for this job name
+	 * and job parameters
 	 */
 	boolean isJobInstanceExists(String jobName, JobParameters jobParameters);
 
@@ -68,8 +70,8 @@ public interface JobRepository {
 	 * found and was already completed successfully.
 	 * 
 	 */
-	JobExecution createJobExecution(String jobName, JobParameters jobParameters) throws JobExecutionAlreadyRunningException,
-			JobRestartException, JobInstanceAlreadyCompleteException;
+	JobExecution createJobExecution(String jobName, JobParameters jobParameters)
+			throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException;
 
 	/**
 	 * Update the {@link JobExecution}.
@@ -112,6 +114,22 @@ JobExecution createJobExecution(String jobName, JobParameters jobParameters) thr
 	 */
 	void updateExecutionContext(StepExecution stepExecution);
 
+	/**
+	 * Load the {@link ExecutionContext} of the given {@link StepExecution}.
+	 * 
+	 * @param stepExecution the {@link StepExecution} containing the
+	 * {@link ExecutionContext}.
+	 */
+	ExecutionContext getExecutionContext(StepExecution stepExecution);
+
+	/**
+	 * Load the {@link ExecutionContext} of the given {@link JobExecution}.
+	 * 
+	 * @param jobExecution the {@link JobExecution} containing the
+	 * {@link ExecutionContext}.
+	 */
+	ExecutionContext getExecutionContext(JobExecution jobExecution);
+
 	/**
 	 * @param stepName the name of the step execution that might have run.
 	 * @return the last execution of step for the given job instance.
@@ -125,7 +143,7 @@ JobExecution createJobExecution(String jobName, JobParameters jobParameters) thr
 	int getStepExecutionCount(JobInstance jobInstance, String stepName);
 
 	/**
-	 * @param jobName the name of the job that might have run 
+	 * @param jobName the name of the job that might have run
 	 * @param jobParameters parameters identifying the {@link JobInstance}
 	 * @return the last execution of job if exists, null otherwise
 	 */

spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/SimpleJobRepository.java

@@ -336,4 +336,12 @@ public JobExecution getLastJobExecution(String jobName, JobParameters jobParamet
 
 	}
 
+	public ExecutionContext getExecutionContext(StepExecution stepExecution) {
+		return ecDao.getExecutionContext(stepExecution);
+	}
+
+	public ExecutionContext getExecutionContext(JobExecution jobExecution) {
+		return ecDao.getExecutionContext(jobExecution);
+	}
+
 }

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

@@ -20,6 +20,7 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
@@ -39,6 +40,7 @@
 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;
@@ -88,6 +90,14 @@ public boolean rollbackOn(Throwable ex) {
 
 	private Semaphore semaphore = new Semaphore(1);
 
+	private Classifier<Exception, Boolean> nonFatalCommitExceptions = new Classifier<Exception, Boolean>() {
+
+		public Boolean classify(Exception classifiable) {
+			return false;
+		}
+
+	};
+
 	/**
 	 * Default constructor.
 	 */
@@ -102,6 +112,14 @@ public TaskletStep(String name) {
 		super(name);
 	}
 
+	/**
+	 * @param nonFatalCommitExceptions classifies whether commit exception is
+	 * fatal or not.
+	 */
+	public void setNonFatalCommitExceptions(Classifier<Exception, Boolean> nonFatalCommitExceptions) {
+		this.nonFatalCommitExceptions = nonFatalCommitExceptions;
+	}
+
 	/**
 	 * Public setter for the {@link PlatformTransactionManager}.
 	 * 
@@ -278,10 +296,18 @@ public RepeatStatus doInStepContext(RepeatContext repeatContext, StepContext ste
 						transactionManager.commit(transaction);
 					}
 					catch (Exception e) {
-						fatalException.setException(e);
-						stepExecution.setStatus(BatchStatus.UNKNOWN);
-						logger.error("Fatal error detected during commit.");
-						throw new FatalException("Fatal error detected during commit", e);
+						if (nonFatalCommitExceptions.classify(e)) {
+							stepExecution.setExecutionContext(getJobRepository().getExecutionContext(stepExecution));
+							JobExecution jobExecution = stepExecution.getJobExecution();
+							jobExecution.setExecutionContext(getJobRepository().getExecutionContext(jobExecution));
+							throw new CommitException("non-fatal commit failure", e);
+						}
+						else {
+							fatalException.setException(e);
+							stepExecution.setStatus(BatchStatus.UNKNOWN);
+							logger.error("Fatal error detected during commit.");
+							throw new FatalException("Fatal error detected during commit", e);
+						}
 					}
 
 					try {
@@ -300,7 +326,10 @@ public RepeatStatus doInStepContext(RepeatContext repeatContext, StepContext ste
 					throw e;
 				}
 				catch (Exception e) {
-					processRollback(stepExecution, fatalException, transaction);
+					// if commit failed, calling rollback on tx manager would cause exception
+					if (!(e instanceof CommitException)) {
+						processRollback(stepExecution, fatalException, transaction);
+					}
 					throw e;
 				}
 				finally {
@@ -376,6 +405,15 @@ public Exception getException() {
 
 	}
 
+	/**
+	 * Signals non-fatal commit failure.
+	 */
+	private static class CommitException extends RuntimeException {
+		public CommitException(String msg, Throwable cause) {
+			super(msg, cause);
+		}
+	}
+
 	protected void close(ExecutionContext ctx) throws Exception {
 		stream.close(ctx);
 	}

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

@@ -20,6 +20,7 @@
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.item.ExecutionContext;
 
 /**
  * @author Dave Syer
@@ -82,4 +83,12 @@ public JobExecution getLastJobExecution(String jobName, JobParameters jobParamet
 		return null;
 	}
 
+	public ExecutionContext getExecutionContext(StepExecution stepExecution) {
+		return null;
+	}
+
+	public ExecutionContext getExecutionContext(JobExecution jobExecution) {
+		return null;
+	}
+
 }

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

@@ -11,7 +11,6 @@
 import org.springframework.batch.core.ExitStatus;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInstance;
-import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
@@ -28,6 +27,7 @@
 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;
@@ -41,15 +41,13 @@
  */
 public class TaskletStepExceptionTests {
 
-	TaskletStep taskletStep;
+	private TaskletStep taskletStep;
 
-	StepExecution stepExecution;
+	private StepExecution stepExecution;
 
-	UpdateCountingJobRepository jobRepository;
+	private UpdateCountingJobRepository jobRepository;
 
-	static RuntimeException taskletException = new RuntimeException();
-
-	static JobInterruptedException interruptedException = new JobInterruptedException("");
+	private static RuntimeException taskletException = new RuntimeException();
 
 	@Before
 	public void init() {
@@ -134,7 +132,7 @@ public RepeatStatus execute(StepContribution contribution, AttributeAccessor att
 	}
 
 	@Test
-	/**
+	/*
 	 * Exception in afterStep is ignored (only logged).
 	 */
 	public void testAfterStepFAilure() throws Exception {
@@ -196,12 +194,61 @@ public RepeatStatus execute(StepContribution contribution, AttributeAccessor att
 		assertEquals(exception, e.getCause());
 	}
 
+	/**
+	 * If commit exception isn't fatal step shouldn't complete with UNKNOWN
+	 * status and execution context should be rolled back.
+	 */
+	@Test
+	public void testSkippableCommitError() throws Exception {
+
+		class TestItemStream extends ItemStreamSupport {
+			private boolean called = false;
+
+			@Override
+			public void update(ExecutionContext executionContext) throws ItemStreamException {
+				executionContext.put("key", "value");
+				called = true;
+			}
+
+		}
+		final TestItemStream stream = new TestItemStream();
+		taskletStep.registerStream(stream);
+
+		final RuntimeException commitException = new RuntimeException();
+		taskletStep.setNonFatalCommitExceptions(new Classifier<Exception, Boolean>() {
+
+			public Boolean classify(Exception classifiable) {
+				return true;
+			}
+		});
+		taskletStep.setTransactionManager(new ResourcelessTransactionManager() {
+			@Override
+			protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
+				throw commitException;
+			}
+		});
+
+		taskletStep.setTasklet(new Tasklet() {
+
+			public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
+				return RepeatStatus.FINISHED;
+			}
+
+		});
+
+		taskletStep.execute(stepExecution);
+		assertEquals("step won't refuse to restart", FAILED, stepExecution.getStatus());
+		assertTrue("execution context modified", stream.called);
+		assertTrue("execution context rolled back", stepExecution.getExecutionContext().isEmpty());
+	}
+
 	@Test
 	public void testUpdateError() throws Exception {
 
 		final RuntimeException exception = new RuntimeException();
 		taskletStep.setJobRepository(new UpdateCountingJobRepository() {
 			boolean firstCall = true;
+
 			@Override
 			public void update(StepExecution arg0) {
 				if (firstCall) {
@@ -211,7 +258,7 @@ public void update(StepExecution arg0) {
 				throw exception;
 			}
 		});
-		
+
 		taskletStep.execute(stepExecution);
 		assertEquals(UNKNOWN, stepExecution.getStatus());
 		assertTrue(stepExecution.getFailureExceptions().contains(taskletException));
@@ -272,9 +319,16 @@ public int getUpdateCount() {
 		}
 
 		public JobExecution getLastJobExecution(String jobName, JobParameters jobParameters) {
-			// TODO Auto-generated method stub
 			return null;
 		}
+
+		public ExecutionContext getExecutionContext(StepExecution stepExecution) {
+			return new ExecutionContext();
+		}
+
+		public ExecutionContext getExecutionContext(JobExecution jobExecution) {
+			return new ExecutionContext();
+		}
 	}
 
 }