From b811f1412253c9f0c113c302d16b583f95e3a1d1 Mon Sep 17 00:00:00 2001 From: dsyer Date: Tue, 23 Jun 2009 15:28:02 +0000 Subject: [PATCH] RESOLVED - issue BATCH-1304: Filter counter not incremented whenever there's a skip Fixed - required new protected methods in SimpleChunkProcessor --- .../item/FaultTolerantChunkProcessor.java | 91 ++++++++++++++++-- .../core/step/item/SimpleChunkProcessor.java | 95 ++++++++++++++----- .../FaultTolerantChunkProcessorTests.java | 7 +- 3 files changed, 159 insertions(+), 34 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java index f06772054..6d738e7cc 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java @@ -121,14 +121,69 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor inputs) { + @SuppressWarnings("unchecked") + UserData data = (UserData) inputs.getUserData(); + if (data == null) { + data = new UserData(inputs.size()); + inputs.setUserData(data); + data.setOutputs(new Chunk()); + } + } + + @Override + protected int getFilterCount(Chunk inputs, Chunk outputs) { + @SuppressWarnings("unchecked") + UserData data = (UserData) inputs.getUserData(); + return data.size() - outputs.size() - inputs.getSkips().size(); + } + + @Override + protected boolean isComplete(Chunk inputs) { + + /* + * Need to remember the write skips across transactions, otherwise they + * keep coming back. Since we register skips with the inputs they will + * not be processed again but the output skips need to be saved for + * registration later with the listeners. The inputs are going to be the + * same for all transactions processing the same chunk, but the outputs + * are not, so we stash them in user data on the inputs. + */ + + @SuppressWarnings("unchecked") + UserData data = (UserData) inputs.getUserData(); + Chunk previous = data.getOutputs(); + + return inputs.isEmpty() && previous.getSkips().isEmpty(); + + } + + @Override + protected Chunk getAdjustedOutputs(Chunk inputs, Chunk outputs) { + + @SuppressWarnings("unchecked") + UserData data = (UserData) inputs.getUserData(); + Chunk previous = data.getOutputs(); + + Chunk next = new Chunk(outputs.getItems(), previous.getSkips()); + next.setBusy(previous.isBusy()); + + // Remember for next time if there are skips accumulating + data.setOutputs(next); + + return next; + + } + @Override protected Chunk transform(final StepContribution contribution, Chunk inputs) throws Exception { Chunk outputs = new Chunk(); - Object userData = inputs.getUserData(); @SuppressWarnings("unchecked") - final Chunk cache = (userData instanceof Chunk) ? (Chunk) userData : null; - final Chunk.ChunkIterator cacheIterator = (cache != null) ? cache.iterator() : null; + UserData data = (UserData) inputs.getUserData(); + Chunk cache = data.getOutputs(); + final Chunk.ChunkIterator cacheIterator = cache.isEmpty() ? null : cache.iterator(); final AtomicInteger count = new AtomicInteger(0); for (final Chunk.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) { @@ -141,7 +196,7 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor 1) { /* * If there is a cached chunk then we must be @@ -284,7 +339,7 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor inputs, final Chunk outputs, ChunkMonitor chunkMonitor) throws Exception { - logger.debug("Scanning for failed item on write: "+inputs); + logger.debug("Scanning for failed item on write: " + inputs); if (outputs.isEmpty()) { inputs.setBusy(false); return; @@ -393,4 +448,28 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor { + + private final int size; + + private Chunk outputs; + + public UserData(int size) { + this.size = size; + } + + public int size() { + return size; + } + + public Chunk getOutputs() { + return outputs; + } + + public void setOutputs(Chunk outputs) { + this.outputs = outputs; + } + + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java index 56fa6e964..d01cb0a91 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java @@ -27,8 +27,9 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; /** - * Simple implementation of the {@link ChunkProcessor} interface that handles basic - * item writing and processing. Any exceptions encountered will be rethrown. + * Simple implementation of the {@link ChunkProcessor} interface that handles + * basic item writing and processing. Any exceptions encountered will be + * rethrown. * * @see ChunkOrientedTasklet */ @@ -157,41 +158,83 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi public final void process(StepContribution contribution, Chunk inputs) throws Exception { - /* - * Need to remember the write skips across transactions, otherwise they - * keep coming back. Since we register skips with the inputs they will - * not be processed again but the output skips need to be saved for - * registration later with the listeners. The inputs are going to be the - * same for all transactions processing the same chunk, but the outputs - * are not, so we stash them in user data on the inputs. - */ - - @SuppressWarnings("unchecked") - Chunk skips = (Chunk) inputs.getUserData(); - if (skips == null) { - skips = new Chunk(); - } + // Allow temporary state to be stored in the user data field + initializeUserData(inputs); // If there is no input we don't have to do anything more - if (inputs.isEmpty() && skips.getSkips().isEmpty()) { + if (isComplete(inputs)) { return; } - int inputsSize = inputs.size(); - + // Make the transformation, calling remove() on the inputs iterator if + // any items are filtered. Might throw exception and cause rollback. Chunk outputs = transform(contribution, inputs); - contribution.incrementFilterCount(inputsSize - outputs.size() - inputs.getSkips().size()); + // Adjust the filter count based on available data + contribution.incrementFilterCount(getFilterCount(inputs, outputs)); - boolean busy = skips.isBusy(); - outputs = new Chunk(outputs.getItems(), skips.getSkips()); - outputs.setBusy(busy); + // Adjust the outputs if necessary for housekeeping purposes, and then + // write them out... + write(contribution, inputs, getAdjustedOutputs(inputs, outputs)); - // Remember for next time if there are skips accumulating - inputs.setUserData(outputs); + } - write(contribution, inputs, outputs); + /** + * Extension point for subclasses to allow them to memorise the contents of + * the inputs, in case they are needed for accounting purposes later. The + * default implementation sets up some user data to remember the original + * size of the inputs. If this method is overridden then some or all of + * {@link #isComplete(Chunk)}, {@link #getFilterCount(Chunk, Chunk)} and + * {@link #getAdjustedOutputs(Chunk, Chunk)} might also need to be, to + * ensure that the user data is handled consistently. + * + * @param inputs the inputs for the process + */ + protected void initializeUserData(Chunk inputs) { + inputs.setUserData(inputs.size()); + } + /** + * Extension point for subclasses to calculate the filter count. Defaults to + * the difference between input size and output size. + * + * @param inputs the inputs after transformation + * @param outputs the outputs after transformation + * + * @return the difference in sizes + * + * @see #initializeUserData(Chunk) + */ + protected int getFilterCount(Chunk inputs, Chunk outputs) { + return (Integer) inputs.getUserData() - outputs.size(); + } + + /** + * Extension point for subclasses that want to store additional data in the + * inputs. Default just checks if inputs are empty. + * + * @param inputs the input chunk + * @return true if it is empty + * + * @see #initializeUserData(Chunk) + */ + protected boolean isComplete(Chunk inputs) { + return inputs.isEmpty(); + } + + /** + * Extension point for subclasses that want to adjust the outputs based on + * additional saved data in the inputs. Default implementation just returns + * the outputs unchanged. + * + * @param inputs the inputs for the transformation + * @param outputs the result of the transformation + * @return the outputs unchanged + * + * @see #initializeUserData(Chunk) + */ + protected Chunk getAdjustedOutputs(Chunk inputs, Chunk outputs) { + return outputs; } /** diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java index c8ba0bc7a..001785826 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java @@ -77,10 +77,13 @@ public class FaultTolerantChunkProcessorTests { if (item.equals("1")) { throw new RuntimeException("Skippable"); } + if (item.equals("3")) { + return null; + } return item; } }); - Chunk inputs = new Chunk(Arrays.asList("1", "2")); + Chunk inputs = new Chunk(Arrays.asList("3", "1", "2")); try { processor.process(contribution, inputs); fail("Expected Exception"); @@ -90,7 +93,7 @@ public class FaultTolerantChunkProcessorTests { processor.process(contribution, inputs); assertEquals(1, list.size()); assertEquals(1, contribution.getSkipCount()); - assertEquals(0, contribution.getFilterCount()); + assertEquals(1, contribution.getFilterCount()); } @Test