diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java index 2194f47f3..f375cc8e5 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java @@ -134,4 +134,12 @@ public class StepContribution { + ", writeSkips=" + writeSkipCount + "]"; } + /** + * @param contribution + */ + public void increment(StepContribution contribution) { + itemCount += contribution.getItemCount(); + readSkipCount += contribution.getReadSkipCount(); + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java index 10c6304fb..1a7e4095e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java @@ -38,7 +38,7 @@ import org.springframework.core.AttributeAccessor; * {@link ItemWriter}. * * Provides extension points by protected {@link #read(StepContribution)} and - * {@link #write(Object, StepContribution)} methods that can be overriden to + * {@link #write(List, StepContribution)} methods that can be overriden to * provide more sophisticated behavior (e.g. skipping). * * @author Dave Syer @@ -46,7 +46,9 @@ import org.springframework.core.AttributeAccessor; */ public class ItemOrientedStepHandler implements StepHandler { - private static final String ITEM_BUFFER_KEY = ItemOrientedStepHandler.class.getName() + ".ITEM_BUFFER_KEY"; + private static final String INPUT_BUFFER_KEY = "INPUT_BUFFER_KEY"; + + private static final String OUTPUT_BUFFER_KEY = "OUTPUT_BUFFER_KEY"; protected final Log logger = LogFactory.getLog(getClass()); @@ -76,7 +78,7 @@ public class ItemOrientedStepHandler implements StepHandler { /** * Get the next item from {@link #read(StepContribution)} and if not null - * pass the item to {@link #write(Object, StepContribution)}. If the + * pass the item to {@link #write(List, StepContribution)}. If the * {@link ItemProcessor} returns null, the write is omitted and another item * taken from the reader. * @@ -85,51 +87,53 @@ public class ItemOrientedStepHandler implements StepHandler { */ public ExitStatus handle(final StepContribution contribution, AttributeAccessor attributes) throws Exception { - final List> buffer = getItemBuffer(attributes); + final List> inputs = getInputBuffer(attributes); + final List> outputs = getOutputBuffer(attributes); ExitStatus result = ExitStatus.CONTINUABLE; - if (buffer.isEmpty()) { + if (inputs.isEmpty() && outputs.isEmpty()) { result = repeatOperations.iterate(new RepeatCallback() { public ExitStatus doInIteration(final RepeatContext context) throws Exception { - ReadWrapper item = read(contribution); + ItemWrapper item = read(contribution); contribution.incrementReadSkipCount(item.getSkipCount()); if (item.getItem() == null) { return ExitStatus.FINISHED; } - buffer.add(item); + inputs.add(item); return ExitStatus.CONTINUABLE; } }); + storeInputs(attributes, inputs); + } - List processed = new ArrayList(); + for (Iterator> iterator = inputs.iterator(); iterator.hasNext();) { - for (Iterator> iterator = buffer.iterator(); iterator.hasNext();) { - - ReadWrapper item = iterator.next(); + ItemWrapper item = iterator.next(); S output = null; + // TODO: processor listener + output = itemProcessor.process(item.getItem()); + // TODO: segregate read / write / filter count // (this is read count) contribution.incrementItemCount(); - // TODO: processor listener - output = itemProcessor.process(item.getItem()); // TODO: increment filter count if this is null if (output != null) { - processed.add(output); + outputs.add(new ItemWrapper(output)); } } + storeOutputsAndClearInputs(attributes, outputs, contribution); + // TODO: use ItemWriter interface properly // TODO: make sure exceptions get handled by the appropriate handler - for (S data : processed) { - write(data, contribution); - } + write(outputs, contribution); // On successful completion clear the attributes to signal that there is // no more processing @@ -140,18 +144,65 @@ public class ItemOrientedStepHandler implements StepHandler { } + /** + * @param attributes + */ + private void clearInputs(AttributeAccessor attributes) { + attributes.removeAttribute(INPUT_BUFFER_KEY); + } + + /** + * @param attributes + * @param inputs + */ + private void storeInputs(AttributeAccessor attributes, List> inputs) { + store(attributes, INPUT_BUFFER_KEY, inputs); + } + + /** + * Savepoint at end of processing and before writing. The processed items + * ready for output are stored so that if writing fails they can be picked + * up again in the next try. The inputs are finished with so we can clear + * their attribute. + * + * @param attributes + * @param outputs + */ + private void storeOutputsAndClearInputs(AttributeAccessor attributes, List> outputs, + StepContribution contribution) { + store(attributes, OUTPUT_BUFFER_KEY, outputs); + clearInputs(attributes); + } + + /** + * @param attributes + * @param inputBufferKey + * @param outputs + */ + private void store(AttributeAccessor attributes, String key, W value) { + attributes.setAttribute(key, value); + } + private void clearAll(AttributeAccessor attributes) { for (String key : attributes.attributeNames()) { attributes.removeAttribute(key); } } - private List> getItemBuffer(AttributeAccessor attributes) { - if (!attributes.hasAttribute(ITEM_BUFFER_KEY)) { - attributes.setAttribute(ITEM_BUFFER_KEY, new ArrayList>()); + private List> getInputBuffer(AttributeAccessor attributes) { + return getBuffer(attributes, INPUT_BUFFER_KEY); + } + + private List> getOutputBuffer(AttributeAccessor attributes) { + return getBuffer(attributes, OUTPUT_BUFFER_KEY); + } + + private List getBuffer(AttributeAccessor attributes, String key) { + if (!attributes.hasAttribute(key)) { + return new ArrayList(); } @SuppressWarnings("unchecked") - List> resource = (List>) attributes.getAttribute(ITEM_BUFFER_KEY); + List resource = (List) attributes.getAttribute(key); return resource; } @@ -159,8 +210,8 @@ public class ItemOrientedStepHandler implements StepHandler { * @param contribution current context * @return next item for writing */ - protected ReadWrapper read(StepContribution contribution) throws Exception { - return new ReadWrapper(doRead()); + protected ItemWrapper read(StepContribution contribution) throws Exception { + return new ItemWrapper(doRead()); } /** @@ -173,27 +224,29 @@ public class ItemOrientedStepHandler implements StepHandler { /** * - * @param item the item to write + * @param items the item to write * @param contribution current context */ - protected void write(S item, StepContribution contribution) throws Exception { - doWrite(item); + protected void write(List> items, StepContribution contribution) throws Exception { + for (ItemWrapper item : items) { + doWrite(item); + } } /** * @param item * @throws Exception */ - protected final void doWrite(S item) throws Exception { + protected final void doWrite(ItemWrapper item) throws Exception { // TODO: increment write count - itemWriter.write(Collections.singletonList(item)); + itemWriter.write(Collections.singletonList(item.getItem())); } /** * @author Dave Syer * */ - static protected class ReadWrapper { + static protected class ItemWrapper { final private T item; @@ -202,7 +255,7 @@ public class ItemOrientedStepHandler implements StepHandler { /** * @param item */ - public ReadWrapper(T item) { + public ItemWrapper(T item) { this(item, 0); } @@ -210,7 +263,7 @@ public class ItemOrientedStepHandler implements StepHandler { * @param item * @param skipCount */ - public ReadWrapper(T item, int skipCount) { + public ItemWrapper(T item, int skipCount) { this.item = item; this.skipCount = skipCount; } @@ -230,6 +283,16 @@ public class ItemOrientedStepHandler implements StepHandler { return skipCount; } + /* + * (non-Javadoc) + * + * @see java.lang.Object#toString() + */ + @Override + public String toString() { + return String.format("[%s,%d]", item, skipCount); + } + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java index 315a5b95b..5b50203d9 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java @@ -274,8 +274,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean } }); StatefulRetryStepHandler itemHandler = new StatefulRetryStepHandler(getItemReader(), - getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, itemKeyGenerator, readSkipPolicy, - writeSkipPolicy); + getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, itemKeyGenerator, + readSkipPolicy, writeSkipPolicy); itemHandler.setSkipListeners(BatchListenerFactoryHelper.getSkipListeners(getListeners())); step.setStepHandler(itemHandler); @@ -327,9 +327,9 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean * @param itemKeyGenerator */ public StatefulRetryStepHandler(ItemReader itemReader, - ItemProcessor itemProcessor, ItemWriter itemWriter, RepeatOperations chunkOperations, - RetryOperations retryTemplate, ItemKeyGenerator itemKeyGenerator, ItemSkipPolicy readSkipPolicy, - ItemSkipPolicy writeSkipPolicy) { + ItemProcessor itemProcessor, ItemWriter itemWriter, + RepeatOperations chunkOperations, RetryOperations retryTemplate, ItemKeyGenerator itemKeyGenerator, + ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy) { super(itemReader, itemProcessor, itemWriter, chunkOperations); this.retryOperations = retryTemplate; this.itemKeyGenerator = itemKeyGenerator; @@ -368,13 +368,13 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean * count * @return next item for processing */ - protected ReadWrapper read(StepContribution contribution) throws Exception { - + protected ItemWrapper read(StepContribution contribution) throws Exception { + int skipCount = 0; while (true) { try { - return new ReadWrapper(doRead(), skipCount); + return new ItemWrapper(doRead(), skipCount); } catch (Exception e) { try { @@ -410,37 +410,48 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean /** * Execute the business logic, delegating to the writer.
* - * Process the item with the {@link ItemWriter} in a stateful retry. Any + * Process the items with the {@link ItemWriter} in a stateful retry. Any * {@link SkipListener} provided is called when retry attempts are * exhausted. The listener callback (on write failure) will happen in * the next transaction automatically.
*/ @Override - protected void write(final S item, final StepContribution contribution) throws Exception { - RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() { - public Object doWithRetry(RetryContext context) throws Throwable { - doWrite(item); - return null; - } - }, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) : item); - retryCallback.setRecoveryCallback(new RecoveryCallback() { - public Object recover(RetryContext context) { - Throwable t = context.getLastThrowable(); - if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) { - contribution.incrementWriteSkipCount(); - try { - listener.onSkipInWrite(item, t); - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t); - } - } else { - throw new RetryException("Non-skippable exception in recoverer", t); + protected void write(final List> items, final StepContribution contribution) throws Exception { + + for (final ItemWrapper item : new ArrayList>(items)) { + + RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { + doWrite(item); + return null; } - return null; - } - }); - retryOperations.execute(retryCallback); + }, itemKeyGenerator != null ? itemKeyGenerator.getKey(item.getItem()) : item); + + retryCallback.setRecoveryCallback(new RecoveryCallback() { + + public Object recover(RetryContext context) { + Throwable t = context.getLastThrowable(); + if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) { + contribution.incrementWriteSkipCount(); + items.remove(item); + try { + listener.onSkipInWrite(item.getItem(), t); + } + catch (RuntimeException ex) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t); + } + } + else { + throw new RetryException("Non-skippable exception in recoverer", t); + } + return null; + + } + }); + + retryOperations.execute(retryCallback); + + } } } 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 58b2bcf87..11a4c7a3b 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 @@ -233,8 +233,7 @@ public class StepHandlerStep extends AbstractStep { AttributeAccessor attributes = attributeQueue.poll(); if (attributes == null) { - attributes = new AttributeAccessorSupport() { - }; + attributes = new BasicAttributeAccessor(); } try { @@ -258,6 +257,11 @@ public class StepHandlerStep extends AbstractStep { if (attributes.attributeNames().length > 0) { attributeQueue.add(attributes); } + + // Apply the contribution to the step + // even if unsuccessful + stepExecution.apply(contribution); + } contribution.incrementCommitCount(); @@ -274,10 +278,6 @@ public class StepHandlerStep extends AbstractStep { Thread.currentThread().interrupt(); } - // Apply the contribution to the step - // only if chunk was successful - stepExecution.apply(contribution); - try { stream.update(stepExecution.getExecutionContext()); } @@ -322,11 +322,11 @@ public class StepHandlerStep extends AbstractStep { } catch (Error e) { - processRollback(stepExecution, contribution, fatalException, transaction); + processRollback(stepExecution, fatalException, transaction); throw e; } catch (Exception e) { - processRollback(stepExecution, contribution, fatalException, transaction); + processRollback(stepExecution, fatalException, transaction); throw e; } finally { @@ -351,15 +351,12 @@ public class StepHandlerStep extends AbstractStep { /** * @param stepExecution - * @param contribution * @param fatalException * @param transaction */ - private void processRollback(final StepExecution stepExecution, final StepContribution contribution, - final ExceptionHolder fatalException, TransactionStatus transaction) { + private void processRollback(final StepExecution stepExecution, final ExceptionHolder fatalException, + TransactionStatus transaction) { - stepExecution.incrementReadSkipCountBy(contribution.getReadSkipCount()); - stepExecution.incrementWriteSkipCountBy(contribution.getWriteSkipCount()); /* * Any exception thrown within the transaction should automatically * cause the transaction to rollback. @@ -387,6 +384,13 @@ public class StepHandlerStep extends AbstractStep { } } + /** + * @author Dave Syer + * + */ + private static final class BasicAttributeAccessor extends AttributeAccessorSupport { + } + private static class ExceptionHolder { private Exception exception; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java index 50053c49e..0f962bcec 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java @@ -343,14 +343,13 @@ public class SkipLimitStepFactoryBeanTests { factory.setSkipLimit(4); factory.setItemReader(reader); factory.setItemWriter(writer); + factory.setCommitInterval(3); // includes all expected skips Step step = (Step) factory.getObject(); StepExecution stepExecution = jobExecution.createStepExecution(step); step.execute(stepExecution); - System.err.println(writer.written); - System.err.println(reader.processed); assertEquals(4, stepExecution.getSkipCount()); assertEquals(2, stepExecution.getReadSkipCount()); assertEquals(2, stepExecution.getWriteSkipCount()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java index 9973a3c14..9a786ab77 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java @@ -43,6 +43,7 @@ import org.springframework.batch.core.repository.dao.MapStepExecutionDao; import org.springframework.batch.core.repository.support.SimpleJobRepository; import org.springframework.batch.core.step.AbstractStep; import org.springframework.batch.core.step.skip.SkipLimitExceededException; +import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.ListItemReader; @@ -283,17 +284,17 @@ public class StatefulRetryStepFactoryBeanTests { assertEquals(4, processed.size()); // [] assertEquals(0, recovered.size()); - assertEquals(0, stepExecution.getItemCount()); + assertEquals(1, stepExecution.getItemCount()); } @Test public void testNonSkippableException() throws Exception { - + // Very specific skippable exception factory.setSkippableExceptionClasses(new Class[] { UnsupportedOperationException.class }); // ...which is not retryable... factory.setRetryableExceptionClasses(new Class[0]); - + factory.setSkipLimit(1); List items = Arrays.asList(new String[] { "b" }); ItemReader provider = new ListItemReader(items) { @@ -323,7 +324,7 @@ public class StatefulRetryStepFactoryBeanTests { catch (RuntimeException e) { // expected String message = e.getMessage(); - assertTrue("Wrong message: "+message, message.contains("Write error - planned but not skippable.")); + assertTrue("Wrong message: " + message, message.contains("Write error - planned but not skippable.")); } assertEquals(0, stepExecution.getSkipCount()); @@ -333,7 +334,7 @@ public class StatefulRetryStepFactoryBeanTests { assertEquals(1, processed.size()); // [] assertEquals(0, recovered.size()); - assertEquals(0, stepExecution.getItemCount()); + assertEquals(1, stepExecution.getItemCount()); } @Test @@ -376,7 +377,7 @@ public class StatefulRetryStepFactoryBeanTests { assertEquals(4, processed.size()); // [] assertEquals(0, recovered.size()); - assertEquals(0, stepExecution.getItemCount()); + assertEquals(1, stepExecution.getItemCount()); } @Test @@ -390,7 +391,7 @@ public class StatefulRetryStepFactoryBeanTests { factory.setCacheCapacity(2); ItemReader provider = new ItemReader() { public String read() { - String item = ""+count; + String item = "" + count; provided.add(item); count++; if (count >= 10) { @@ -409,6 +410,12 @@ public class StatefulRetryStepFactoryBeanTests { }; factory.setItemReader(provider); factory.setItemWriter(itemWriter); + factory.setItemKeyGenerator(new ItemKeyGenerator() { + public Object getKey(Object item) { + // return random object so the cache fills up + return new Object(); + } + }); AbstractStep step = (AbstractStep) factory.getObject(); StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); @@ -420,13 +427,14 @@ public class StatefulRetryStepFactoryBeanTests { // expected } - assertEquals(1, stepExecution.getSkipCount()); + // We added a bogus key generator so no items are actually skipped + // because they aren't recognised as eligible + assertEquals(0, stepExecution.getSkipCount()); // only one item processed but three (the commit interval) were provided // [0, 1, 2] assertEquals(3, provided.size()); - // TODO: this is a bug: 0 was skipped but it came back in the buffer for the second try - // [0, 0, 1, 0, 0] - assertEquals(5, processed.size()); + // [0, 0, 0] + assertEquals(3, processed.size()); // [] assertEquals(0, recovered.size()); }