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 d22953ef2..977904ce3 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 @@ -26,8 +26,6 @@ import org.springframework.batch.core.StepContribution; import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.MarkFailedException; -import org.springframework.batch.item.ResetFailedException; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; @@ -180,20 +178,6 @@ public class ItemOrientedStepHandler implements StepHandler { itemWriter.write(Collections.singletonList(item)); } - /** - * @throws MarkFailedException - */ - public void mark() throws MarkFailedException { - itemReader.mark(); - } - - /** - * @throws ResetFailedException - */ - public void reset() throws ResetFailedException { - itemReader.reset(); - } - /** * @author Dave Syer * diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandler.java index 482e4a3f4..a44798439 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandler.java @@ -18,8 +18,6 @@ package org.springframework.batch.core.step.item; import org.springframework.batch.core.StepContribution; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.MarkFailedException; -import org.springframework.batch.item.ResetFailedException; import org.springframework.batch.repeat.ExitStatus; /** @@ -45,16 +43,4 @@ public interface StepHandler { */ ExitStatus handle(StepContribution contribution) throws Exception; - /** - * Implementations should delegate to an {@link ItemReader}. - * - */ - void mark() throws MarkFailedException; - - /** - * Implementations should delegate to an {@link ItemReader}. - * - */ - void reset() throws ResetFailedException; - } 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 345ed9404..ef979fe93 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 @@ -205,7 +205,6 @@ public class StepHandlerStep extends AbstractStep { protected ExitStatus doExecute(final StepExecution stepExecution) throws Exception { stream.update(stepExecution.getExecutionContext()); getJobRepository().updateExecutionContext(stepExecution); - itemHandler.mark(); final ExceptionHolder fatalException = new ExceptionHolder(); @@ -281,7 +280,6 @@ public class StepHandlerStep extends AbstractStep { } try { - itemHandler.mark(); transactionManager.commit(transaction); } catch (Exception e) { @@ -347,7 +345,6 @@ public class StepHandlerStep extends AbstractStep { stepExecution.rollback(); try { - itemHandler.reset(); transactionManager.rollback(transaction); } catch (Exception e) { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/DrivingQueryItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/DrivingQueryItemReader.java index 5b846de29..a7a393829 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/DrivingQueryItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/DrivingQueryItemReader.java @@ -69,14 +69,12 @@ public class DrivingQueryItemReader implements ItemReader, InitializingBea private boolean initialized = false; - private List keys; + private T currentKey = null; private Iterator keysIterator; private int currentIndex = 0; - private int lastCommitIndex = 0; - private KeyCollector keyCollector; private boolean saveState = false; @@ -91,7 +89,6 @@ public class DrivingQueryItemReader implements ItemReader, InitializingBea * @param keys */ public DrivingQueryItemReader(List keys) { - this.keys = keys; this.keysIterator = keys.iterator(); } @@ -105,7 +102,8 @@ public class DrivingQueryItemReader implements ItemReader, InitializingBea if (keysIterator.hasNext()) { currentIndex++; - return keysIterator.next(); + currentKey = keysIterator.next(); + return currentKey; } return null; @@ -119,11 +117,7 @@ public class DrivingQueryItemReader implements ItemReader, InitializingBea * @return the current key. */ protected T getCurrentKey() { - if (initialized && currentIndex > 0) { - return keys.get(currentIndex - 1); - } - - return null; + return currentKey; } /** @@ -133,8 +127,6 @@ public class DrivingQueryItemReader implements ItemReader, InitializingBea public void close(ExecutionContext executionContext) { initialized = false; currentIndex = 0; - lastCommitIndex = 0; - keys = null; keysIterator = null; } @@ -147,9 +139,9 @@ public class DrivingQueryItemReader implements ItemReader, InitializingBea */ public void open(ExecutionContext executionContext) { - Assert.state(keys == null && !initialized, "Cannot open an already opened item reader" + Assert.state(keysIterator == null && !initialized, "Cannot open an already opened item reader" + ", call close() first."); - keys = keyCollector.retrieveKeys(executionContext); + List keys = keyCollector.retrieveKeys(executionContext); Assert.notNull(keys, "Keys must not be null"); keysIterator = keys.listIterator(); initialized = true; @@ -177,26 +169,10 @@ public class DrivingQueryItemReader implements ItemReader, InitializingBea this.keyCollector = keyCollector; } - /** - * Mark is supported as long as this {@link ItemStream} is used in a - * single-threaded environment. The state backing the mark is a single - * counter, keeping track of the current position, so multiple threads - * cannot be accommodated. - */ public void mark() { - lastCommitIndex = currentIndex; } - /* - * (non-Javadoc) - * - * @see - * org.springframework.batch.io.support.AbstractTransactionalIoSource#reset - * (org.springframework.batch.item.ExecutionContext) - */ public void reset() { - keysIterator = keys.listIterator(lastCommitIndex); - currentIndex = lastCommitIndex; } public void setSaveState(boolean saveState) { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java index f271daf4c..c18853cbb 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java @@ -23,7 +23,7 @@ import org.hibernate.SessionFactory; import org.hibernate.StatelessSession; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream; +import org.springframework.batch.item.support.AbstractItemReaderItemStream; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -51,7 +51,7 @@ import org.springframework.util.ClassUtils; * @author Robert Kasanicky * @author Dave Syer */ -public class HibernateCursorItemReader extends AbstractBufferedItemReaderItemStream implements ItemStream, +public class HibernateCursorItemReader extends AbstractItemReaderItemStream implements ItemStream, InitializingBean { private SessionFactory sessionFactory; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java index 6bfb37f3c..b5fa7973c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java @@ -29,7 +29,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream; +import org.springframework.batch.item.support.AbstractItemReaderItemStream; import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.InvalidDataAccessResourceUsageException; import org.springframework.jdbc.SQLWarningException; @@ -97,7 +97,7 @@ import org.springframework.util.ClassUtils; * @author Peter Zozom * @author Robert Kasanicky */ -public class JdbcCursorItemReader extends AbstractBufferedItemReaderItemStream implements InitializingBean { +public class JdbcCursorItemReader extends AbstractItemReaderItemStream implements InitializingBean { private static Log log = LogFactory.getLog(JdbcCursorItemReader.class); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java index 45984c3f5..337dc60a1 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java @@ -27,7 +27,7 @@ import javax.persistence.Query; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream; +import org.springframework.batch.item.support.AbstractItemReaderItemStream; import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.util.Assert; @@ -58,7 +58,7 @@ import org.springframework.util.ClassUtils; * * @author Thomas Risberg */ -public class JpaPagingItemReader extends AbstractBufferedItemReaderItemStream implements InitializingBean { +public class JpaPagingItemReader extends AbstractItemReaderItemStream implements InitializingBean { protected Log logger = LogFactory.getLog(getClass()); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java index 7f52e7bf9..b51575f9b 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemReader.java @@ -33,7 +33,7 @@ import org.springframework.batch.item.file.separator.RecordSeparatorPolicy; import org.springframework.batch.item.file.transform.AbstractLineTokenizer; import org.springframework.batch.item.file.transform.DelimitedLineTokenizer; import org.springframework.batch.item.file.transform.LineTokenizer; -import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream; +import org.springframework.batch.item.support.AbstractItemReaderItemStream; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; import org.springframework.util.Assert; @@ -62,7 +62,7 @@ import org.springframework.util.ClassUtils; * @author Robert Kasanicky * @author Dave Syer */ -public class FlatFileItemReader extends AbstractBufferedItemReaderItemStream implements +public class FlatFileItemReader extends AbstractItemReaderItemStream implements ResourceAwareItemReaderItemStream, InitializingBean { private static Log log = LogFactory.getLog(FlatFileItemReader.class); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java index 90d78ce2d..7d521df39 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/MultiResourceItemReader.java @@ -1,10 +1,7 @@ package org.springframework.batch.item.file; -import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; -import java.util.List; -import java.util.ListIterator; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; @@ -33,12 +30,9 @@ import org.springframework.util.ClassUtils; * * @author Robert Kasanicky */ -public class MultiResourceItemReader extends ExecutionContextUserSupport implements ItemReader, ItemStream { +public class MultiResourceItemReader implements ItemReader, ItemStream { - /** - * Unique object instance that marks resource boundaries in the item buffer - */ - private static final Object END_OF_RESOURCE_MARKER = new Object(); + private final ExecutionContextUserSupport executionContextUserSupport = new ExecutionContextUserSupport(); private ResourceAwareItemReaderItemStream delegate; @@ -46,13 +40,7 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impl private MultiResourceIndex index = new MultiResourceIndex(); - private List itemBuffer = new ArrayList(); - - private ListIterator itemBufferIterator = null; - - private boolean shouldReadBuffer = false; - - private boolean saveState = false; + private boolean saveState = true; private Comparator comparator = new Comparator() { @@ -66,7 +54,7 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impl }; public MultiResourceItemReader() { - setName(ClassUtils.getShortName(MultiResourceItemReader.class)); + executionContextUserSupport.setName(ClassUtils.getShortName(MultiResourceItemReader.class)); } /** @@ -75,13 +63,7 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impl public T read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { T item; - if (shouldReadBuffer) { - item = readBufferedItem(); - } - else { - item = readNextItem(); - } - + item = readNextItem(); index.incrementItemCount(); return item; @@ -103,7 +85,6 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impl if (index.currentResource >= resources.length) { return null; } - itemBuffer.add(END_OF_RESOURCE_MARKER); delegate.close(new ExecutionContext()); delegate.setResource(resources[index.currentResource]); @@ -112,68 +93,20 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impl item = delegate.read(); } - itemBuffer.add(item); - return item; } - /** - * Read next item from buffer while keeping track of the position within the - * input for possible restart. - * @return next item from buffer - */ - @SuppressWarnings("unchecked") - private T readBufferedItem() { - - Object buffered = itemBufferIterator.next(); - while (buffered == END_OF_RESOURCE_MARKER) { - index.incrementResourceCount(); - buffered = itemBufferIterator.next(); - } - - if (!itemBufferIterator.hasNext()) { - // buffer is exhausted, continue reading from file - shouldReadBuffer = false; - itemBufferIterator = null; - } - return (T) buffered; - } - /** * Remove the longer needed items from buffer, mark the index position and * call mark() on delegate so that it clears its buffers. */ public void mark() throws MarkFailedException { - emptyBuffer(); - - index.mark(); - - delegate.mark(); - } - - /** - * Discard the buffered items that have already been read. - */ - private void emptyBuffer() { - if (!shouldReadBuffer) { - itemBuffer.clear(); - itemBufferIterator = null; - } - else { - itemBuffer = itemBuffer.subList(itemBufferIterator.nextIndex(), itemBuffer.size()); - itemBufferIterator = itemBuffer.listIterator(); - } } /** * Switches to 'read from buffer' state. */ public void reset() throws ResetFailedException { - if (!itemBuffer.isEmpty()) { - shouldReadBuffer = true; - itemBufferIterator = itemBuffer.listIterator(); - } - index.reset(); } /** @@ -181,10 +114,7 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impl * and reset instance variable values. */ public void close(ExecutionContext executionContext) throws ItemStreamException { - shouldReadBuffer = false; - itemBufferIterator = null; index = new MultiResourceIndex(); - itemBuffer.clear(); delegate.close(new ExecutionContext()); } @@ -207,7 +137,6 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impl try { for (int i = 0; i < index.currentItem; i++) { delegate.read(); - delegate.mark(); } } catch (Exception e) { @@ -294,18 +223,18 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impl } public void open(ExecutionContext ctx) { - if (ctx.containsKey(getKey(RESOURCE_KEY))) { - currentResource = Long.valueOf(ctx.getLong(getKey(RESOURCE_KEY))).intValue(); + if (ctx.containsKey(executionContextUserSupport.getKey(RESOURCE_KEY))) { + currentResource = Long.valueOf(ctx.getLong(executionContextUserSupport.getKey(RESOURCE_KEY))).intValue(); } - if (ctx.containsKey(getKey(ITEM_KEY))) { - currentItem = ctx.getLong(getKey(ITEM_KEY)); + if (ctx.containsKey(executionContextUserSupport.getKey(ITEM_KEY))) { + currentItem = ctx.getLong(executionContextUserSupport.getKey(ITEM_KEY)); } } public void update(ExecutionContext ctx) { - ctx.putLong(getKey(RESOURCE_KEY), index.currentResource); - ctx.putLong(getKey(ITEM_KEY), index.currentItem); + ctx.putLong(executionContextUserSupport.getKey(RESOURCE_KEY), index.currentResource); + ctx.putLong(executionContextUserSupport.getKey(ITEM_KEY), index.currentItem); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractBufferedItemReaderItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemReaderItemStream.java similarity index 69% rename from spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractBufferedItemReaderItemStream.java rename to spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemReaderItemStream.java index 397345afc..87f025dc5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractBufferedItemReaderItemStream.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AbstractItemReaderItemStream.java @@ -1,17 +1,11 @@ package org.springframework.batch.item.support; -import java.util.ArrayList; -import java.util.List; -import java.util.ListIterator; - import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; -import org.springframework.batch.item.MarkFailedException; import org.springframework.batch.item.NoWorkFoundException; import org.springframework.batch.item.ParseException; -import org.springframework.batch.item.ResetFailedException; import org.springframework.batch.item.UnexpectedInputException; import org.springframework.batch.item.util.ExecutionContextUserSupport; import org.springframework.util.Assert; @@ -26,22 +20,12 @@ import org.springframework.util.Assert; * * @author Robert Kasanicky */ -public abstract class AbstractBufferedItemReaderItemStream implements ItemReader, ItemStream { +public abstract class AbstractItemReaderItemStream implements ItemReader, ItemStream { private static final String READ_COUNT = "read.count"; private int currentItemCount = 0; - private int lastMarkedItemCount = 0; - - private boolean shouldReadBuffer = false; - - private List itemBuffer = new ArrayList(); - - private ListIterator itemBufferIterator = null; - - private int lastMarkedBufferIndex = 0; - private ExecutionContextUserSupport ecSupport = new ExecutionContextUserSupport(); private boolean saveState = true; @@ -75,24 +59,8 @@ public abstract class AbstractBufferedItemReaderItemStream implements ItemRea } public T read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { - currentItemCount++; - - if (shouldReadBuffer) { - if (itemBufferIterator.hasNext()) { - return itemBufferIterator.next(); - } - else { - // buffer is exhausted, continue reading from file - shouldReadBuffer = false; - itemBufferIterator = null; - } - } - - T item = doRead(); - itemBuffer.add(item); - - return item; + return doRead(); } /** @@ -100,28 +68,11 @@ public abstract class AbstractBufferedItemReaderItemStream implements ItemRea * single-threaded environment. The state backing the mark is a single * counter, keeping track of the current position, so multiple threads * cannot be accommodated. - * - * @see org.springframework.batch.item.support.AbstractItemReader#mark() */ - public void mark() throws MarkFailedException { - - if (!shouldReadBuffer) { - itemBuffer.clear(); - itemBufferIterator = null; - lastMarkedBufferIndex = 0; - } - else { - lastMarkedBufferIndex = itemBufferIterator.nextIndex(); - } - - lastMarkedItemCount = currentItemCount; + public void mark() { } - public void reset() throws ResetFailedException { - - currentItemCount = lastMarkedItemCount; - shouldReadBuffer = true; - itemBufferIterator = itemBuffer.listIterator(lastMarkedBufferIndex); + public void reset() { } protected int getCurrentItemCount() { @@ -134,12 +85,6 @@ public abstract class AbstractBufferedItemReaderItemStream implements ItemRea public void close(ExecutionContext executionContext) throws ItemStreamException { currentItemCount = 0; - lastMarkedItemCount = 0; - lastMarkedBufferIndex = 0; - itemBufferIterator = null; - shouldReadBuffer = false; - itemBuffer.clear(); - try { doClose(); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemReader.java index d800f7c4d..b98bdf534 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/AggregateItemReader.java @@ -97,11 +97,9 @@ public class AggregateItemReader implements ItemReader> { } public void mark() throws MarkFailedException { - itemReader.mark(); } public void reset() throws ResetFailedException { - itemReader.reset(); } public void setItemReader(ItemReader> itemReader) { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemReader.java index e63cea983..3da1d3983 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemReader.java @@ -74,7 +74,6 @@ public class DelegatingItemReader extends AbstractItemReader implements In * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext) */ public void mark() { - itemReader.mark(); } /* @@ -82,6 +81,5 @@ public class DelegatingItemReader extends AbstractItemReader implements In * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext) */ public void reset() { - itemReader.reset(); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java index 8bf6b9a79..feae17b2d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java @@ -9,7 +9,7 @@ import javax.xml.stream.XMLStreamException; import javax.xml.stream.events.StartElement; import org.springframework.batch.item.file.ResourceAwareItemReaderItemStream; -import org.springframework.batch.item.support.AbstractBufferedItemReaderItemStream; +import org.springframework.batch.item.support.AbstractItemReaderItemStream; import org.springframework.batch.item.xml.stax.DefaultFragmentEventReader; import org.springframework.batch.item.xml.stax.FragmentEventReader; import org.springframework.beans.factory.InitializingBean; @@ -30,7 +30,7 @@ import org.springframework.util.ClassUtils; * * @author Robert Kasanicky */ -public class StaxEventItemReader extends AbstractBufferedItemReaderItemStream implements +public class StaxEventItemReader extends AbstractItemReaderItemStream implements ResourceAwareItemReaderItemStream, InitializingBean { private FragmentEventReader fragmentReader; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java index 1e5d76f96..518f0ba94 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java @@ -18,7 +18,6 @@ package org.springframework.batch.retry.policy; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.batch.repeat.support.RepeatSynchronizationManager; import org.springframework.batch.retry.ExhaustedRetryException; import org.springframework.batch.retry.RecoveryCallback; import org.springframework.batch.retry.RetryCallback; diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java index 150256235..f7348924f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java @@ -49,51 +49,6 @@ public abstract class CommonItemReaderTests { assertNull(tested.read()); } - /** - * Rollback scenario - reader resets to last marked point. Note the commit - * interval can change dynamically. - */ - @Test - public void testReset() throws Exception { - Foo foo1 = tested.read(); - assertEquals(1, foo1.getValue()); - - tested.mark(); - - Foo foo2 = tested.read(); - assertEquals(2, foo2.getValue()); - - Foo foo3 = tested.read(); - assertEquals(3, foo3.getValue()); - - tested.reset(); - - assertEquals(foo2, tested.read()); - - tested.mark(); - - assertEquals(foo3, tested.read()); - - tested.reset(); - - assertEquals(foo3, tested.read()); - - Foo foo4 = tested.read(); - assertEquals(4, foo4.getValue()); - - tested.mark(); - - Foo foo5 = tested.read(); - assertEquals(5, foo5.getValue()); - - tested.reset(); - - assertEquals(foo5, tested.read()); - - assertNull(tested.read()); - - } - /** * Empty input should be handled gracefully - null is returned on first * read. diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemStreamItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemStreamItemReaderTests.java index 4c1ba6f67..92a3b7a0d 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemStreamItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemStreamItemReaderTests.java @@ -77,14 +77,10 @@ public abstract class CommonItemStreamItemReaderTests extends CommonItemReaderTe Foo foo2 = tested.read(); assertEquals(2, foo2.getValue()); - tested.mark(); + testedAsStream().update(executionContext); Foo foo3 = tested.read(); assertEquals(3, foo3.getValue()); - - tested.reset(); - - testedAsStream().update(executionContext); // create new input source tested = getItemReader(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java index 796639abe..6aff3679f 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractDataSourceItemReaderIntegrationTests.java @@ -149,28 +149,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests { assertEquals(1, foo.getValue()); } - /* - * Rollback scenario - input source rollbacks to last commit point. - */ - @Transactional @Test - public void testRollback() throws Exception { - getAsItemStream(reader).open(executionContext); - - Foo foo1 = reader.read(); - - commit(); - - Foo foo2 = reader.read(); - Assert.state(!foo2.equals(foo1)); - - Foo foo3 = reader.read(); - Assert.state(!foo2.equals(foo3)); - - rollback(); - - assertEquals(foo2, reader.read()); - } - /* * Rollback scenario with restart - input source rollbacks to last * commit point. @@ -182,7 +160,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests { Foo foo1 = reader.read(); - commit(); + getAsItemStream(reader).update(executionContext); Foo foo2 = reader.read(); Assert.state(!foo2.equals(foo1)); @@ -190,10 +168,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests { Foo foo3 = reader.read(); Assert.state(!foo2.equals(foo3)); - rollback(); - - getAsItemStream(reader).update(executionContext); - // create new input source reader = createItemReader(); @@ -220,10 +194,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests { Foo foo3 = reader.read(); Assert.state(!foo2.equals(foo3)); - rollback(); - - getAsItemStream(reader).update(executionContext); - // create new input source reader = createItemReader(); @@ -238,9 +208,10 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests { getAsItemStream(reader).open(executionContext); + Foo foo1 = reader.read(); - commit(); + getAsItemStream(reader).update(executionContext); Foo foo2 = reader.read(); Assert.state(!foo2.equals(foo1)); @@ -248,10 +219,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests { Foo foo3 = reader.read(); Assert.state(!foo2.equals(foo3)); - rollback(); - - getAsItemStream(reader).update(executionContext); - // create new input source reader = createItemReader(); @@ -262,8 +229,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests { getAsItemStream(reader).update(executionContext); - commit(); - // create new input source reader = createItemReader(); @@ -275,14 +240,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests { assertEquals(5, foo5.getValue()); } - private void commit() { - reader.mark(); - } - - private void rollback() { - reader.reset(); - } - private ItemStream getAsItemStream(ItemReader source) { return (ItemStream) source; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractJdbcItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractJdbcItemReaderIntegrationTests.java index 38238e3ec..8672b80c8 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractJdbcItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/AbstractJdbcItemReaderIntegrationTests.java @@ -144,36 +144,6 @@ public abstract class AbstractJdbcItemReaderIntegrationTests { assertEquals(1, foo.getValue()); } - /* - * Rollback scenario. - */ - @Transactional @Test - public void testRollback() throws Exception { - getAsItemStream(itemReader).open(executionContext); - Foo foo1 = itemReader.read(); - - commit(); - - Foo foo2 = itemReader.read(); - Assert.state(!foo2.equals(foo1)); - - Foo foo3 = itemReader.read(); - Assert.state(!foo2.equals(foo3)); - - rollback(); - - assertEquals(foo2, itemReader.read()); - } - - - private void commit() { - itemReader.mark(); - } - - private void rollback() { - itemReader.reset(); - } - private ItemStream getAsItemStream(ItemReader source) { return (ItemStream) source; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/DrivingQueryItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/DrivingQueryItemReaderTests.java index 7b5c8211a..f408a437d 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/DrivingQueryItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/DrivingQueryItemReaderTests.java @@ -11,7 +11,6 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.sample.Foo; import org.springframework.beans.factory.InitializingBean; import org.springframework.transaction.support.TransactionSynchronizationManager; -import org.springframework.util.Assert; public class DrivingQueryItemReaderTests extends TestCase { @@ -135,28 +134,6 @@ public class DrivingQueryItemReaderTests extends TestCase { assertEquals(1, foo.getValue()); } - /** - * Rollback scenario. - * - * @throws Exception - */ - public void testRollback() throws Exception { - getAsItemStream(itemReader).open(new ExecutionContext()); - Foo foo1 = (Foo) itemReader.read(); - - commit(); - - Foo foo2 = (Foo) itemReader.read(); - Assert.state(!foo2.equals(foo1)); - - Foo foo3 = (Foo) itemReader.read(); - Assert.state(!foo2.equals(foo3)); - - rollback(); - - assertEquals(foo2, itemReader.read()); - } - public void testRetriveZeroKeys() { itemReader.setKeyCollector(new KeyCollector() { @@ -176,14 +153,6 @@ public class DrivingQueryItemReaderTests extends TestCase { } - private void commit() { - itemReader.mark(); - } - - private void rollback() { - itemReader.reset(); - } - private InitializingBean getAsInitializingBean(ItemReader source) { return (InitializingBean) source; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/FooInputSource.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/FooInputSource.java index 14ff66959..cf64d740d 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/FooInputSource.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/FooInputSource.java @@ -1,5 +1,7 @@ package org.springframework.batch.item.database; +import javax.sql.DataSource; + import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; @@ -7,8 +9,6 @@ import org.springframework.batch.item.sample.Foo; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; -import javax.sql.DataSource; - class FooItemReader implements ItemStream, ItemReader, DisposableBean, InitializingBean { DrivingQueryItemReader itemReader; @@ -62,7 +62,6 @@ class FooItemReader implements ItemStream, ItemReader, DisposableBean, Init * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.StreamContext) */ public void mark() { - itemReader.mark(); } /* @@ -70,6 +69,5 @@ class FooItemReader implements ItemStream, ItemReader, DisposableBean, Init * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.StreamContext) */ public void reset() { - itemReader.reset(); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaPagingItemReaderCommonTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaPagingItemReaderCommonTests.java index 63350624e..c2d2571dc 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaPagingItemReaderCommonTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JpaPagingItemReaderCommonTests.java @@ -1,17 +1,16 @@ package org.springframework.batch.item.database; -import org.springframework.batch.item.sample.Foo; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ExecutionContext; -import org.springframework.batch.item.CommonItemStreamItemReaderTests; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import org.springframework.test.context.ContextConfiguration; -import org.junit.runner.RunWith; -import org.junit.Test; - import javax.persistence.EntityManagerFactory; +import org.junit.runner.RunWith; +import org.springframework.batch.item.CommonItemStreamItemReaderTests; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.sample.Foo; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration public class JpaPagingItemReaderCommonTests extends CommonItemStreamItemReaderTests { @@ -41,34 +40,4 @@ public class JpaPagingItemReaderCommonTests extends CommonItemStreamItemReaderTe reader.open(new ExecutionContext()); } - - @Test - public void testRestart() throws Exception { - super.testRestart(); - } - - @Test - public void testResetAndRestart() throws Exception { - super.testResetAndRestart(); - } - - @Test - public void testReopen() throws Exception { - super.testReopen(); - } - - @Test - public void testRead() throws Exception { - super.testRead(); - } - - @Test - public void testReset() throws Exception { - super.testReset(); - } - - @Test - public void testEmptyInput() throws Exception { - super.testEmptyInput(); - } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderAdvancedTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderAdvancedTests.java index 792fb60bd..70f31173e 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderAdvancedTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemReaderAdvancedTests.java @@ -16,8 +16,6 @@ package org.springframework.batch.item.file; -import java.io.IOException; - import junit.framework.TestCase; import org.springframework.batch.item.ExecutionContext; @@ -80,58 +78,6 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { return new ByteArrayResource(input.getBytes()); } - /** - * Test rollback functionality - * - * @throws IOException - */ - public void testReset() throws Exception { - - reader.close(null); - reader.setResource(getInputResource("testLine1\ntestLine2\ntestLine3\ntestLine4\ntestLine5\ntestLine6")); - reader.open(executionContext); - - // read some records - reader.read(); // #1 - reader.read(); // #2 - // commit them - reader.mark(); - // read next record - reader.read(); // # 3 - // read next records - reader.reset(); - - // we should now process all records after first commit point - assertEquals("[testLine3]", reader.read().toString()); - - } - - /** - * Test skip and skipRollback functionality - * - * @throws IOException - */ - public void testFailOnFirstChunk() throws Exception { - - reader.close(null); - reader.setResource(getInputResource("testLine1\ntestLine2\ntestLine3\ntestLine4\ntestLine5\ntestLine6")); - reader.open(executionContext); - - // read some records - reader.read(); // #1 - reader.read(); // #2 - reader.read(); // #3 - // rollback - reader.reset(); - // read next record - reader.read(); // should be #1 - - // we should now process all records after first commit point, that are - // not marked as skipped - assertEquals("[testLine2]", reader.read().toString()); - - } - public void testRestart() throws Exception { reader.close(null); @@ -141,9 +87,6 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { // read some records reader.read(); reader.read(); - // commit them - reader.mark(); - // read next two records reader.read(); reader.read(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemReaderIntegrationTests.java index 2115aefdf..a3c1c4d5a 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemReaderIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/MultiResourceItemReaderIntegrationTests.java @@ -43,7 +43,8 @@ public class MultiResourceItemReaderIntegrationTests extends TestCase { tested.setComparator(new Comparator() { public int compare(Resource o1, Resource o2) { return 0; // do not change ordering - }}); + } + }); tested.setResources(new Resource[] { r1, r2, r3, r4, r5 }); } @@ -67,40 +68,67 @@ public class MultiResourceItemReaderIntegrationTests extends TestCase { tested.close(ctx); } - /** - * Read items with a couple of rollbacks, requiring to jump back to items - * from previous resources. - */ - public void testReset() throws Exception { + public void testRestartWhenStateNotSaved() throws Exception { + + tested.setSaveState(false); tested.open(ctx); assertEquals("1", readItem()); - tested.mark(); + tested.update(ctx); assertEquals("2", readItem()); assertEquals("3", readItem()); - tested.reset(); + tested.close(ctx); + + tested.open(ctx); + + assertEquals("1", readItem()); + } + + /** + * + * Read items with a couple of rollbacks, requiring to jump back to items + * from previous resources. + */ + public void testRestartAcrossResourceBoundary() throws Exception { + + tested.open(ctx); + + assertEquals("1", readItem()); + + tested.update(ctx); + + assertEquals("2", readItem()); + assertEquals("3", readItem()); + + tested.close(ctx); + + tested.open(ctx); assertEquals("2", readItem()); assertEquals("3", readItem()); assertEquals("4", readItem()); - tested.reset(); + tested.close(ctx); + + tested.open(ctx); assertEquals("2", readItem()); assertEquals("3", readItem()); assertEquals("4", readItem()); assertEquals("5", readItem()); - tested.mark(); + tested.update(ctx); assertEquals("6", readItem()); assertEquals("7", readItem()); - tested.reset(); + tested.close(ctx); + + tested.open(ctx); assertEquals("6", readItem()); assertEquals("7", readItem()); @@ -116,9 +144,6 @@ public class MultiResourceItemReaderIntegrationTests extends TestCase { */ public void testRestart() throws Exception { - itemReader.setSaveState(true); - tested.setSaveState(true); - tested.open(ctx); assertEquals("1", readItem()); @@ -146,14 +171,13 @@ public class MultiResourceItemReaderIntegrationTests extends TestCase { * Resources are ordered according to injected comparator. */ public void testResourceOrderingWithCustomComparator() { - + Resource r1 = new ByteArrayResource("".getBytes(), "b"); Resource r2 = new ByteArrayResource("".getBytes(), "a"); Resource r3 = new ByteArrayResource("".getBytes(), "c"); - - - Resource[] resources = new Resource[] {r1, r2, r3}; - + + Resource[] resources = new Resource[] { r1, r2, r3 }; + Comparator comp = new Comparator() { /** diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemReaderTests.java index a56abe6d8..dd60eb9b7 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemReaderTests.java @@ -107,10 +107,11 @@ public class StaxEventItemReaderTests extends TestCase { * Save restart data and restore from it. */ public void testRestart() throws Exception { + source.open(executionContext); source.read(); source.update(executionContext); - System.out.println(executionContext); + assertEquals(1, executionContext.getLong(ClassUtils.getShortName(StaxEventItemReader.class) + ".read.count")); List expectedAfterRestart = source.read(); @@ -118,48 +119,14 @@ public class StaxEventItemReaderTests extends TestCase { source.open(executionContext); List afterRestart = source.read(); assertEquals(expectedAfterRestart.size(), afterRestart.size()); - } - // /** - // * Restore point must not exceed end of file, input source must not be - // already initialised when restoring. - // */ - // public void testInvalidRestore() { - // ExecutionContext context = new ExecutionContext(); - // context.putLong(ClassUtils.getShortName(StaxEventItemReader.class) + - // ".item.count", 100000); - // try { - // source.open(context); - // fail("Expected StreamException"); - // } catch (Exception e) { - // // expected - // String message = e.getMessage(); - // assertTrue("Wrong message: " + message, contains(message, - // "must be before")); - // } - // } + } public void testRestoreWorksFromClosedStream() throws Exception { source.close(executionContext); source.update(executionContext); } - /** - * Rollback to last commited record. - */ - public void testRollback() throws Exception { - source.open(executionContext); - // rollback between deserializing records - List first = source.read(); - source.mark(); - List second = source.read(); - assertFalse(first.equals(second)); - source.reset(); - - assertEquals(second, source.read()); - - } - /** * Statistics return the current record count. Calling read after end of * input does not increase the counter.