From c8166df78828d44540a354f24db5ecf7a61bc872 Mon Sep 17 00:00:00 2001 From: dsyer Date: Sat, 2 Feb 2008 14:25:07 +0000 Subject: [PATCH] IN PROGRESS - issue BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering) http://jira.springframework.org/browse/BATCH-7 Re-order transaction boundary and interrupt check. --- .../step/simple/SimpleStepExecutor.java | 29 +++++++++++-------- 1 file changed, 17 insertions(+), 12 deletions(-) diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index 576da8855..e2982223a 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -177,7 +177,7 @@ public class SimpleStepExecutor { ExitStatus status = ExitStatus.FAILED; - if (streamManager==null) { + if (streamManager == null) { streamManager = new SimpleStreamManager(transactionManager); } @@ -193,7 +193,7 @@ public class SimpleStepExecutor { if (saveStreamContext && isRestart) { stepContext.setInitialStreamContext(stepInstance.getStreamContext()); } - + try { stepExecution.setStartTime(new Date(System.currentTimeMillis())); @@ -227,15 +227,20 @@ public class SimpleStepExecutor { StreamContext statistics = stepContext.getStreamContext(); contribution.setStreamContext(new GenericStreamContext(statistics.getProperties())); contribution.incrementCommitCount(); - // Apply the contribution to the step - // only if chunk was successful - stepExecution.apply(contribution); - if (saveStreamContext) { - stepInstance.setStreamContext(stepContext.getStreamContext()); - jobRepository.update(stepInstance); + synchronized (stepExecution) { + // Apply the contribution to the step + // only if chunk was successful + stepExecution.apply(contribution); + + if (saveStreamContext) { + stepInstance.setStreamContext(stepContext.getStreamContext()); + jobRepository.update(stepInstance); + } + jobRepository.saveOrUpdate(stepExecution); } - jobRepository.saveOrUpdate(stepExecution); + + streamManager.commit(transaction); } catch (Throwable t) { @@ -246,7 +251,9 @@ public class SimpleStepExecutor { * commit (e.g. Hibernate flush) so this catch block * comes outside the transaction. */ - stepExecution.rollback(); + synchronized (stepExecution) { + stepExecution.rollback(); + } streamManager.rollback(transaction); if (t instanceof RuntimeException) { throw (RuntimeException) t; @@ -261,8 +268,6 @@ public class SimpleStepExecutor { // caller interruptionPolicy.checkInterrupted(context); - streamManager.commit(transaction); - return result; }