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 4daa84edd..cc7c67df0 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 @@ -25,6 +25,7 @@ import org.springframework.batch.core.domain.ChunkingResult; import org.springframework.batch.core.domain.Dechunker; import org.springframework.batch.core.domain.DechunkingResult; import org.springframework.batch.core.domain.ItemFailureLog; +import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.JobInterruptedException; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; @@ -47,6 +48,7 @@ import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.RepeatListener; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.retry.RetryCallback; @@ -56,16 +58,38 @@ import org.springframework.transaction.TransactionStatus; import org.springframework.util.Assert; /** - * 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 default).
+ *

Implementation of the {@link Step} interface that deals with input and output as 'chunks'. Reading is + * delegated to a {@link Chunker} that will read in a {@link Chunk} of items for processing. The number of + * items per chunks is configurable as the chunk size. Once the chunk has been read, any errors encountered + * while reading (usually skipped unless configured not to) will be logged out via the {@link ItemFailureLog}. + * The chunk will then be 'dechunked', which in most scenarios will mean delegating to an {@link ItemWriter} + * by writing out one chunk at a time. The transaction boundary is around this process. If any errors are + * encountered, the dechunking process will error out, leaving the decision for retrying the chunk up to + * a {@link RepeatTemplate}. This template is configurable, allowing for the number of retries and how long + * to wait between retries (backoff) to be set. Once dechunking has been finished, any errors not fatal to + * the chunk (usually because the error didn't invalidate the transaction) will also be written out via + * the {@link ItemFailureLog}

* - * Clients can use interceptors in the step operations to intercept or listen to the iteration on a step-wide basis, for - * instance to get a callback when the step is complete. Those that want callbacks at the level of an individual tasks, - * can specify interceptors for the chunk operations. + *

Clients can use {@link RepeatListener}s in the step operations to intercept or listen to the iteration + * on a step-wide basis, for instance to get a callback when the step is complete. The open and close methods of + * could easily be done with AOP, however, notifications in between complete chunks (before and after) can + * be quite useful

+ * + *

Repository Usage: The {@link JobRepository} is used extensively to store metadata about the run such as + * when the {@link StepExecution} was started, or the commit count.

+ * + *

Interruption: At various times while processing, the step will check to see if it has been interrupted + * by calling the {@link StepInterruptionPolicy}. This policy could check if thread.isInterupted() is true, + * or RepeatContext.isTerminateOnly() is set. It could even be a check to see if a 'stop file' has been added + * to a particular directory. If the step should finish, a {@link JobInterruptedException} is thrown, and the + * step will clean up, set the status of the {@link StepExecution} to 'STOPPED' and rethrow.ExitStatusClassification: Any number of fatal errors could be thrown during processing. In general, the + * framework must remain fairly dumb as to what error code these exceptions should translate to. By default + * it's a fairly generic 'FATAL_EXECUTION'. However, this may be insufficient for many scenarios. If an + * enterprise scheduler is used to kick off a batch job, the exit code is the only means of communication as + * to what action must be taken. It may also be the only result that many batch operators see as well. Therefore, + * an {@link ExitStatusExceptionClassifier} may be used to classify an exception to a particular exit code.

* * @author Dave Syer * @author Lucas Ward @@ -79,6 +103,7 @@ public class ChunkedStep extends AbstractStep { private JobRepository jobRepository; + //default to simple exception classification. private ExitStatusExceptionClassifier exceptionClassifier = new SimpleExitStatusExceptionClassifier(); // default to checking current thread for interruption. @@ -89,8 +114,12 @@ public class ChunkedStep extends AbstractStep { private StreamManager streamManager; private ItemReader itemReader; + private Chunker chunker; private ItemWriter itemWriter; + private Dechunker dechunker; + + private ItemSkipPolicy itemSkipPolicy; private RetryTemplate retryTemplate = new RetryTemplate(); @@ -168,6 +197,24 @@ public class ChunkedStep extends AbstractStep { public void setItemWriter(ItemWriter itemWriter) { this.itemWriter = itemWriter; } + + public void setChunker(Chunker chunker) { + this.chunker = chunker; + } + + public void setDechunker(Dechunker dechunker) { + this.dechunker = dechunker; + } + + /** + * Set the skip policy. If set, it will be used for both reading + * and writing. + * + * @param itemSkipPolicy + */ + public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { + this.itemSkipPolicy = itemSkipPolicy; + } /** * Check mandatory properties (reader and writer). @@ -175,8 +222,24 @@ public class ChunkedStep extends AbstractStep { * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() */ public void afterPropertiesSet() throws Exception { - Assert.notNull(itemReader, "ItemReader must be provided"); - Assert.notNull(itemWriter, "ItemWriter must be provided"); + //This is currently a little bit funky, I don't want to require a chunker or + //dechunker to be wired in, since the developer should really only be wiring in a + //ItemReader and ItemWriter, a namespace should take care of the issue though. + if(chunker == null){ + chunker = new ItemChunker(itemReader); + if(itemSkipPolicy != null){ + ((ItemChunker)chunker).setItemSkipPolicy(itemSkipPolicy); + } + } + + if(dechunker == null){ + dechunker = new ItemDechunker(itemWriter); + if(itemSkipPolicy != null){ + ((ItemChunker)dechunker).setItemSkipPolicy(itemSkipPolicy); + } + } + + } /** @@ -184,7 +247,7 @@ public class ChunkedStep extends AbstractStep { * 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 execution, which - * would normally be available to the caller somehow through the step's {@link JobExecutionContext}.
+ * would normally be available to the caller somehow through the step's {@link StepContext}.
* * @throws JobInterruptedException if the step or a chunk is interrupted * @throws RuntimeException if there is an exception during a chunk execution @@ -228,9 +291,7 @@ public class ChunkedStep extends AbstractStep { // interruption. interruptionPolicy.checkInterrupted(context); - //shouldn't have to create a chunker each time, I'll refactor the interface later - Chunker chunker = new ItemChunker(itemReader, stepExecution); - ChunkingResult chunkingResult = chunker.chunk(chunkSize); + ChunkingResult chunkingResult = chunker.chunk(chunkSize, stepExecution); if(chunkingResult == null){ return ExitStatus.FINISHED; @@ -287,11 +348,11 @@ public class ChunkedStep extends AbstractStep { } /** - * Execute a bunch of identical business logic operations all within a transaction. The transaction is - * programmatically started and stopped outside this method, so subclasses that override do not need to create a - * transaction. + * Execute a bunch of identical business logic operations all within a transaction. * - * @param stepInstance the current step containing the {@link Tasklet} with the business logic. + * @param stepExecution the current execution in which to process the chunk in. + * @param chunk to be processed. + * @param stepContext the current step context. * @return true if there is more data to process. */ void processChunk(Chunk chunk, final StepExecution stepExecution, StepContext stepContext) { @@ -301,10 +362,8 @@ public class ChunkedStep extends AbstractStep { final StepContribution contribution = stepExecution.createStepContribution(); try { - - Dechunker dechunker = new ItemDechunker(itemWriter, stepExecution); - DechunkingResult chunkResult = dechunker.dechunk(chunk); + DechunkingResult chunkResult = dechunker.dechunk(chunk, stepExecution); failureLog.log(chunkResult.getExceptions()); // TODO: check that stepExecution can @@ -363,7 +422,7 @@ public class ChunkedStep extends AbstractStep { } - /** + /* * Convenience method to update the status in all relevant places. * * @param stepInstance the current step diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java index 971dd32b3..71e4bf4f2 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java @@ -17,6 +17,7 @@ package org.springframework.batch.execution.step.simple; import org.springframework.batch.core.domain.Chunk; import org.springframework.batch.core.domain.ChunkingResult; +import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.ReadFailureException; @@ -31,12 +32,13 @@ import org.springframework.batch.io.exception.ReadFailureException; public interface Chunker { /** - * Read in a chunk, given the provided chunk size. + * Read in a chunk, given the provided chunk size for the given StepExecution. * * @param chunkSize the number of items that should be read for this chunk. + * @param StepExecution the stepExecution the current chunk is being processed within. * @return the {@link Chunk} that has been read. * @throws IllegalArgumentException if chunkSize is less than zero. */ - public ChunkingResult chunk(int chunkSize) throws ReadFailureException; + public ChunkingResult chunk(int chunkSize, StepExecution stepExecution) throws ReadFailureException; } 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 419750a11..6a60251d0 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 @@ -36,22 +36,20 @@ import org.springframework.util.Assert; public class ItemChunker implements Chunker { private final ItemReader itemReader; - private final StepExecution stepExecution; - private long chunkCounter = 0; - private ItemSkipPolicy readFailurePolicy = new NeverSkipItemSkipPolicy(); + private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); - public ItemChunker(ItemReader itemReader, StepExecution stepExecution) { + public ItemChunker(ItemReader itemReader) { + Assert.notNull(itemReader, "ItemReader must not be null"); this.itemReader = itemReader; - this.stepExecution = stepExecution; } - public void setReadFailurePolicy(ItemSkipPolicy readFailurePolicy) { - this.readFailurePolicy = readFailurePolicy; + public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { + this.itemSkipPolicy = itemSkipPolicy; } - public ChunkingResult chunk(int size) throws ReadFailureException { + public ChunkingResult chunk(int size, StepExecution stepExecution) throws ReadFailureException { Assert.isTrue(size > 0, "Chunk size must be greater than 0"); int counter = 0; @@ -69,7 +67,7 @@ public class ItemChunker implements Chunker { counter++; } catch (Exception ex) { exceptions.add(ex); - if(!readFailurePolicy.shouldSkip(ex, stepExecution)){ + if(!itemSkipPolicy.shouldSkip(ex, stepExecution)){ rethrow(ex); } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java index 41d9f7f00..c658430c7 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java @@ -38,20 +38,19 @@ import org.springframework.util.Assert; public class ItemDechunker implements Dechunker { private final ItemWriter itemWriter; - private final StepExecution stepExecution; private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); - public ItemDechunker(ItemWriter itemWriter, StepExecution stepExecution) { + public ItemDechunker(ItemWriter itemWriter) { this.itemWriter = itemWriter; - this.stepExecution = stepExecution; } /* (non-Javadoc) * @see org.springframework.batch.core.domain.Dechunker#dechunk(org.springframework.batch.core.domain.Chunk) */ - public DechunkingResult dechunk(Chunk chunk) throws Exception { + public DechunkingResult dechunk(Chunk chunk, StepExecution stepExecution) throws Exception { Assert.notNull(chunk, "Chunk must not be null"); + Assert.notNull(stepExecution, "StepExecution must not be null"); List skippedItems = new ArrayList(); for(Iterator it = chunk.getItems().iterator(); it.hasNext();){ diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java index eb224909d..22f59d191 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java @@ -123,7 +123,7 @@ public class ChunkedStepTests extends TestCase { final JobExecution jobExecution = new JobExecution(jobInstance); final StepExecution stepExecution = new StepExecution(step, jobExecution); - chunkedStep.setItemReader(new ItemReader() { + chunkedStep.setChunker(new ItemChunker(new ItemReader() { int counter = 0; public Object read() throws Exception { assertEquals(step, stepExecution.getStep()); @@ -135,7 +135,7 @@ public class ChunkedStepTests extends TestCase { return null; } } - }); + })); chunkedStep.execute(stepExecution); assertEquals(2, processed.size()); @@ -212,7 +212,7 @@ public class ChunkedStepTests extends TestCase { }; - chunkedStep.setItemReader(itemReader); + chunkedStep.setChunker(new ItemChunker(itemReader)); try { chunkedStep.execute(stepExecution); @@ -233,7 +233,7 @@ public class ChunkedStepTests extends TestCase { } }; - chunkedStep.setItemWriter(itemWriter); + chunkedStep.setDechunker(new ItemDechunker(itemWriter)); try{ chunkedStep.execute(stepExecution); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java index 3ccd3483f..058a2e8ba 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java @@ -34,8 +34,8 @@ public class ItemChunkerTests extends TestCase { public void testSizeNegative() { try { MockItemReader itemReader = new MockItemReader(10); - ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - chunkReader.chunk(-1); + ItemChunker chunkReader = new ItemChunker(itemReader); + chunkReader.chunk(-1, stepExecution); fail(); } catch (IllegalArgumentException e) { } @@ -44,8 +44,8 @@ public class ItemChunkerTests extends TestCase { public void testSizeZero() { try { MockItemReader itemReader = new MockItemReader(10); - ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - chunkReader.chunk(0); + ItemChunker chunkReader = new ItemChunker(itemReader); + chunkReader.chunk(0, stepExecution); fail(); } catch (IllegalArgumentException e) { } @@ -53,25 +53,25 @@ public class ItemChunkerTests extends TestCase { public void testSizePositive() { MockItemReader itemReader = new MockItemReader(10); - ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - ChunkingResult chunkingResult = chunkReader.chunk(10); + ItemChunker chunkReader = new ItemChunker(itemReader); + ChunkingResult chunkingResult = chunkReader.chunk(10, stepExecution); assertEquals(10, chunkingResult.getChunk().getItems().size()); } public void testIncompleteChunk() { MockItemReader itemReader = new MockItemReader(5); - ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - ChunkingResult chunkingResult = chunkReader.chunk(10); + ItemChunker chunkReader = new ItemChunker(itemReader); + ChunkingResult chunkingResult = chunkReader.chunk(10, stepExecution); assertEquals(5, chunkingResult.getChunk().getItems().size()); } public void testPolicyNoContinue() { MockItemReader itemReader = new MockItemReader(1); itemReader.setFail(true); - ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(true)); + ItemChunker chunkReader = new ItemChunker(itemReader); + chunkReader.setItemSkipPolicy(new StubReadFailurePolicy(true)); try { - chunkReader.chunk(10); + chunkReader.chunk(10, stepExecution); fail(); } catch (RuntimeException e) { } @@ -80,9 +80,9 @@ public class ItemChunkerTests extends TestCase { public void testPolicyContinueWithFailure() { MockItemReader itemReader = new MockItemReader(1); itemReader.setFail(true); - ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(false)); - ChunkingResult chunkingResult = chunkReader.chunk(1); + ItemChunker chunkReader = new ItemChunker(itemReader); + chunkReader.setItemSkipPolicy(new StubReadFailurePolicy(false)); + ChunkingResult chunkingResult = chunkReader.chunk(1, stepExecution); assertEquals(1,chunkingResult.getChunk().getItems().size()); } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java index 994540b0b..efd0d10d7 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java @@ -48,7 +48,7 @@ public class ItemDechunkerTests extends TestCase { itemWriter = (ItemWriter)writerControl.getMock(); stepExecution = new StepExecution(null,null); - dechunker = new ItemDechunker(itemWriter, stepExecution); + dechunker = new ItemDechunker(itemWriter); List items = new ArrayList(); items.add("1"); items.add("2"); @@ -61,7 +61,7 @@ public class ItemDechunkerTests extends TestCase { itemWriter.write("1"); itemWriter.write("2"); writerControl.replay(); - dechunker.dechunk(chunk); + dechunker.dechunk(chunk, stepExecution); writerControl.verify(); } @@ -72,7 +72,7 @@ public class ItemDechunkerTests extends TestCase { itemWriter.write("2"); writerControl.setThrowable(new Exception()); writerControl.replay(); - DechunkingResult result = dechunker.dechunk(chunk); + DechunkingResult result = dechunker.dechunk(chunk, stepExecution); writerControl.verify(); List exceptions = result.getExceptions(); assertEquals(1, exceptions.size()); @@ -87,7 +87,7 @@ public class ItemDechunkerTests extends TestCase { writerControl.setThrowable(new NullPointerException()); writerControl.replay(); try{ - dechunker.dechunk(chunk); + dechunker.dechunk(chunk, stepExecution); fail(); } catch(NullPointerException ex){