RESOLVED - issue BATCH-782: Synchronization issue in ItemOrientedStep if exception is throw in chunk processing

Added flag to check that lock is not released before it is acquired

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

@@ -47,11 +47,12 @@ import org.springframework.transaction.interceptor.TransactionAttribute;
  * Simple implementation of executing the step as a set of chunks, each chunk
  * surrounded by a transaction. The structure is therefore that of two nested
  * loops, with transaction boundary around the whole inner loop. The outer loop
- * is controlled by the step operations ({@link #setStepOperations(RepeatOperations)}),
- * and the inner loop by the chunk operations ({@link #setChunkOperations(RepeatOperations)}).
- * The inner loop should always be executed in a single thread, so the chunk
- * operations should not do any concurrent execution. N.B. usually that means
- * that the chunk operations should be a {@link RepeatTemplate} (which is the
+ * is controlled by the step operations (
+ * {@link #setStepOperations(RepeatOperations)}), and the inner loop by the
+ * chunk operations ({@link #setChunkOperations(RepeatOperations)}). The inner
+ * loop should always be executed in a single thread, so the chunk operations
+ * should not do any concurrent execution. N.B. usually that means that the
+ * chunk operations should be a {@link RepeatTemplate} (which is the
  * default).<br/>
  * 
  * Clients can use interceptors in the step operations to intercept or listen to
@@ -209,9 +210,9 @@ public class StepHandlerStep extends AbstractStep {
 	 * Process the step and update its context so that progress can be monitored
 	 * by the caller. The step is broken down into chunks, each one executing in
 	 * a transaction. The step and its execution and execution context are all
-	 * given an up to date {@link BatchStatus}, and the {@link JobRepository}
-	 * is used to store the result. Various reporting information are also added
-	 * to the current context (the {@link RepeatContext} governing the step
+	 * given an up to date {@link BatchStatus}, and the {@link JobRepository} is
+	 * used to store the result. Various reporting information are also added to
+	 * the current context (the {@link RepeatContext} governing the step
 	 * execution, which would normally be available to the caller somehow
 	 * through the step's {@link ExecutionContext}.<br/>
 	 * 
@@ -242,18 +243,22 @@ public class StepHandlerStep extends AbstractStep {
 
 				TransactionStatus transaction = transactionManager.getTransaction(transactionAttribute);
 
+				boolean locked = false;
+
 				try {
 
 					try {
 						exitStatus = processChunk(stepExecution, contribution);
-					} catch (Error e) {
+					}
+					catch (Error e) {
 						if (transactionAttribute.rollbackOn(e)) {
 							throw e;
 						}
-					} catch (Exception e) {
+					}
+					catch (Exception e) {
 						if (transactionAttribute.rollbackOn(e)) {
 							throw e;
-						}						
+						}
 					}
 
 					contribution.incrementCommitCount();
@@ -263,6 +268,7 @@ public class StepHandlerStep extends AbstractStep {
 					// minimum).
 					try {
 						synchronizer.lock(stepExecution);
+						locked = true;
 					}
 					catch (InterruptedException e) {
 						stepExecution.setStatus(BatchStatus.STOPPED);
@@ -275,14 +281,16 @@ public class StepHandlerStep extends AbstractStep {
 
 					try {
 						stream.update(stepExecution.getExecutionContext());
-					} catch (Error e) {
+					}
+					catch (Error e) {
 						if (transactionAttribute.rollbackOn(e)) {
 							throw e;
 						}
-					} catch (Exception e) {
+					}
+					catch (Exception e) {
 						if (transactionAttribute.rollbackOn(e)) {
 							throw e;
-						}						
+						}
 					}
 
 					try {
@@ -304,7 +312,7 @@ public class StepHandlerStep extends AbstractStep {
 						logger.error("Fatal error detected during commit.");
 						throw new FatalException("Fatal error detected during commit", e);
 					}
-					
+
 					try {
 						getJobRepository().update(stepExecution);
 					}
@@ -324,7 +332,11 @@ public class StepHandlerStep extends AbstractStep {
 					throw e;
 				}
 				finally {
-					synchronizer.release(stepExecution);
+					// only release the lock if we acquired it
+					if (locked) {
+						synchronizer.release(stepExecution);
+					}
+					locked = false;
 				}
 
 				// Check for interruption after transaction as well, so that
@@ -398,8 +410,8 @@ public class StepHandlerStep extends AbstractStep {
 		}
 		catch (Exception e) {
 			/*
-			 * If we already failed to commit, it doesn't help to do this again -
-			 * it's better to allow the CommitFailedException to propagate
+			 * If we already failed to commit, it doesn't help to do this again
+			 * - it's better to allow the CommitFailedException to propagate
 			 */
 			if (!fatalException.hasException()) {
 				fatalException.setException(e);

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

@@ -140,6 +140,36 @@ public class StepExecutorInterruptionTests extends TestCase {
 
 	}
 
+
+	public void testLockNotReleasedIfChunkFails() throws Exception {
+
+		step.setItemHandler(new SimpleStepHandler<Object>(new AbstractItemReader<Object>() {
+			public Object read() throws Exception {
+				throw new RuntimeException("Planned!");
+			}
+		}, itemWriter));
+
+		step.setSynchronizer(new StepExecutionSynchronizer() {
+			private boolean locked = false;
+			public void lock(StepExecution stepExecution) throws InterruptedException {
+				locked = true;
+			}
+			public void release(StepExecution stepExecution) {
+				assertTrue("Lock released before it is acquired", locked);
+			}
+		});
+		
+		try {
+			step.execute(stepExecution);
+			fail("Expected planned RuntimeException");
+		} catch (RuntimeException e) {
+			assertEquals("Planned!", e.getMessage());
+		}
+
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+
+	}
+
 	/**
 	 * @return
 	 */