From f29aa1e9fd0cbd2891ae188abe5d5f5e160f3555 Mon Sep 17 00:00:00 2001 From: dsyer Date: Thu, 14 Aug 2008 12:35:05 +0000 Subject: [PATCH] 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 --- .../batch/core/step/item/StepHandlerStep.java | 48 ++++++++++++------- .../item/StepExecutorInterruptionTests.java | 30 ++++++++++++ 2 files changed, 60 insertions(+), 18 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandlerStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandlerStep.java index af0317609..8469d86e5 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandlerStep.java +++ b/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).
* * 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}.
* @@ -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); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepExecutorInterruptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepExecutorInterruptionTests.java index f2d542deb..6c60221ae 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepExecutorInterruptionTests.java +++ b/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(new AbstractItemReader() { + 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 */