From 1b1bf33931ff3b2bdf221befacec2824df9beaec Mon Sep 17 00:00:00 2001 From: lucasward Date: Fri, 15 Feb 2008 23:30:21 +0000 Subject: [PATCH] BATCH-220: Moved transactional boundary within ChunkedStep to include both 'Chunking' and 'Dechunking' --- .../execution/step/simple/ChunkedStep.java | 29 ++++++++++--------- .../execution/step/simple/ItemChunker.java | 7 ----- 2 files changed, 15 insertions(+), 21 deletions(-) diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java index b6137beb5..b454a5468 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java @@ -330,20 +330,10 @@ public class ChunkedStep extends StepSupport implements InitializingBean { // interruption. interruptionPolicy.checkInterrupted(context); - ChunkingResult chunkingResult = chunker.chunk(chunkSize, stepExecution); - - if (chunkingResult == null) { - return ExitStatus.FINISHED; - } - - final Chunk chunk = chunkingResult.getChunk(); - failureLog.handle(chunkingResult.getExceptions()); - - retryTemplate.execute(new RetryCallback() { + ExitStatus result = (ExitStatus)retryTemplate.execute(new RetryCallback() { public Object doWithRetry(RetryContext context) throws Throwable { - processChunk(chunk, stepExecution, stepContext); - return null; + return processChunk(stepExecution, stepContext); } }); @@ -352,7 +342,7 @@ public class ChunkedStep extends StepSupport implements InitializingBean { // caller interruptionPolicy.checkInterrupted(context); - return ExitStatus.CONTINUABLE; + return result; } }); @@ -416,13 +406,22 @@ public class ChunkedStep extends StepSupport implements InitializingBean { * @param stepContext the current step context. * @return true if there is more data to process. */ - void processChunk(Chunk chunk, final StepExecution stepExecution, StepContext stepContext) { + ExitStatus processChunk(final StepExecution stepExecution, StepContext stepContext) { TransactionStatus transaction = streamManager.getTransaction(stepExecution); final StepContribution contribution = stepExecution.createStepContribution(); try { + + ChunkingResult chunkingResult = chunker.chunk(chunkSize, stepExecution); + + if (chunkingResult == null) { + return ExitStatus.FINISHED; + } + + final Chunk chunk = chunkingResult.getChunk(); + failureLog.handle(chunkingResult.getExceptions()); DechunkingResult chunkResult = dechunker.dechunk(chunk, stepExecution); failureLog.handle(chunkResult.getExceptions()); @@ -451,6 +450,8 @@ public class ChunkedStep extends StepSupport implements InitializingBean { } streamManager.commit(transaction); + + return ExitStatus.CONTINUABLE; } catch (Throwable t) { diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java index b5be9d4c8..6e3f8aa72 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java @@ -21,7 +21,6 @@ import java.util.List; import org.springframework.batch.core.domain.Chunk; import org.springframework.batch.core.domain.Chunker; import org.springframework.batch.core.domain.ChunkingResult; -import org.springframework.batch.core.domain.Entity; import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.ReadFailureException; @@ -96,10 +95,4 @@ public class ItemChunker implements Chunker { return new Long(chunkCounter++); } - /** - * No-op implementation. - */ - public void flush(Entity stepExecution) { - } - }