diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/Skippable.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/Skippable.java deleted file mode 100644 index 23ab4d760..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/Skippable.java +++ /dev/null @@ -1,39 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.batch.item; - -/** - * Implementation of this interface indicates to the framework that this object - * is capable of skipping a record in cases where it cannot be processed because - * it is invalid, incomplete for other non critical reasons. - * - * @author Waseem Malik - * @author Dave Syer - * - */ -public interface Skippable { - - /** - * Skip the current record. This method can be invoked whenever an input - * source provides an invalid object. The implementing class should skip the - * current record the next time it is encountered in the same process (e.g. - * after a rollback and retry of a transaction). - * - */ - public void skip(); - -} 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 e894ca4b2..023d3d471 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 @@ -203,6 +203,7 @@ public class DrivingQueryItemReader implements ItemReader, InitializingBean, */ 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 77071f14f..5e383b75d 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 @@ -15,9 +15,6 @@ */ package org.springframework.batch.item.database; -import java.util.ArrayList; -import java.util.List; - import org.hibernate.ScrollableResults; import org.hibernate.Session; import org.hibernate.SessionFactory; @@ -26,35 +23,33 @@ import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ExecutionContextUserSupport; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.Skippable; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; -import org.springframework.util.StringUtils; /** * {@link ItemReader} for reading database records built on top of Hibernate. * - * It executes the HQL {@link #queryString} when initialized and iterates over the result set as {@link #read()} method - * is called, returning an object corresponding to current row. + * It executes the HQL {@link #queryString} when initialized and iterates over + * the result set as {@link #read()} method is called, returning an object + * corresponding to current row. * - * Input source can be configured to use either {@link StatelessSession} sufficient for simple mappings without the need - * to cascade to associated objects or standard hibernate {@link Session} for more advanced mappings or when caching is - * desired. + * Input source can be configured to use either {@link StatelessSession} + * sufficient for simple mappings without the need to cascade to associated + * objects or standard hibernate {@link Session} for more advanced mappings or + * when caching is desired. * - * When stateful session is used it will be cleared after successful commit without being flushed (no inserts or updates - * are expected). + * When stateful session is used it will be cleared after successful commit + * without being flushed (no inserts or updates are expected). * * @author Robert Kasanicky * @author Dave Syer */ public class HibernateCursorItemReader extends ExecutionContextUserSupport implements ItemReader, ItemStream, - Skippable, InitializingBean { + InitializingBean { private static final String RESTART_DATA_ROW_NUMBER_KEY = "row.number"; - private static final String SKIPPED_ROWS = "skipped.rows"; - private SessionFactory sessionFactory; private StatelessSession statelessSession; @@ -69,10 +64,6 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple private int lastCommitRowNumber = 0; - private final List skippedRows = new ArrayList(); - - private int skipCount = 0; - /* Current count of processed records. */ private int currentProcessedRow = 0; @@ -90,15 +81,6 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple } if (cursor.next()) { currentProcessedRow++; - if (!skippedRows.isEmpty()) { - // while is necessary to handle successive skips. - while (skippedRows.contains(new Integer(currentProcessedRow))) { - if (!cursor.next()) { - return null; - } - currentProcessedRow++; - } - } Object data = cursor.get(0); return data; } @@ -110,14 +92,19 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple */ public void close(ExecutionContext executionContext) { initialized = false; - cursor.close(); + if (cursor != null) { + cursor.close(); + } currentProcessedRow = 0; - skippedRows.clear(); - skipCount = 0; if (useStatelessSession) { - statelessSession.close(); - } else { - statefulSession.close(); + if (statelessSession != null) { + statelessSession.close(); + } + } + else { + if (statelessSession != null) { + statelessSession.close(); + } } } @@ -130,7 +117,8 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple if (useStatelessSession) { statelessSession = sessionFactory.openStatelessSession(); cursor = statelessSession.createQuery(queryString).scroll(); - } else { + } + else { statefulSession = sessionFactory.openSession(); cursor = statefulSession.createQuery(queryString).scroll(); } @@ -141,13 +129,6 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple cursor.setRowNumber(currentProcessedRow - 1); } - if (executionContext.containsKey(getKey(SKIPPED_ROWS))) { - String[] skipped = StringUtils.commaDelimitedListToStringArray(executionContext - .getString(getKey(SKIPPED_ROWS))); - for (int i = 0; i < skipped.length; i++) { - this.skippedRows.add(new Integer(skipped[i])); - } - } } /** @@ -172,8 +153,9 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple /** * Can be set only in uninitialized state. * - * @param useStatelessSession true to use {@link StatelessSession} false to use - * standard hibernate {@link Session} + * @param useStatelessSession true to use + * {@link StatelessSession} false to use standard hibernate + * {@link Session} */ public void setUseStatelessSession(boolean useStatelessSession) { Assert.state(!initialized); @@ -186,23 +168,14 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple if (saveState) { Assert.notNull(executionContext, "ExecutionContext must not be null"); executionContext.putString(getKey(RESTART_DATA_ROW_NUMBER_KEY), "" + currentProcessedRow); - String skipped = skippedRows.toString(); - executionContext.putString(getKey(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1)); } } /** - * Skip the current row. If the transaction is rolled back, this row will not be represented when read() is called. - * For example, if you read in row 2, find the data to be bad, and call skip(), then continue processing and find - */ - public void skip() { - skippedRows.add(new Integer(currentProcessedRow)); - skipCount++; - } - - /** - * 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. + * 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. * * @see org.springframework.batch.item.ItemReader#mark() */ @@ -222,7 +195,8 @@ public class HibernateCursorItemReader extends ExecutionContextUserSupport imple currentProcessedRow = lastCommitRowNumber; if (lastCommitRowNumber == 0) { cursor.beforeFirst(); - } else { + } + else { // Set the cursor so that next time it is advanced it will // come back to the committed row. cursor.setRowNumber(lastCommitRowNumber - 1); 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 2b6360a53..1c0593c84 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 @@ -22,8 +22,6 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.SQLWarning; import java.sql.Statement; -import java.util.ArrayList; -import java.util.List; import javax.sql.DataSource; @@ -34,7 +32,6 @@ import org.springframework.batch.item.ExecutionContextUserSupport; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ResetFailedException; -import org.springframework.batch.item.Skippable; import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.DataAccessException; import org.springframework.dao.InvalidDataAccessResourceUsageException; @@ -47,7 +44,6 @@ import org.springframework.jdbc.support.SQLExceptionTranslator; import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; -import org.springframework.util.StringUtils; /** *

@@ -96,7 +92,7 @@ import org.springframework.util.StringUtils; * @author Peter Zozom */ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements ItemReader, InitializingBean, - ItemStream, Skippable { + ItemStream { private static Log log = LogFactory.getLog(JdbcCursorItemReader.class); @@ -104,10 +100,6 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements private static final String CURRENT_PROCESSED_ROW = "last.processed.row.number"; - private static final String SKIPPED_ROWS = "skipped.rows"; - - private static final String SKIP_COUNT = "skipped.record.count"; - private Connection con; private PreparedStatement preparedStatement; @@ -120,10 +112,6 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements private String sql; - private final List skippedRows = new ArrayList(); - - private int skipCount = 0; - private int fetchSize = VALUE_NOT_SET; private int maxRows = VALUE_NOT_SET; @@ -193,15 +181,6 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements return null; } else { currentProcessedRow++; - if (!skippedRows.isEmpty()) { - // while is necessary to handle successive skips. - while (skippedRows.contains(new Long(currentProcessedRow))) { - if (!rs.next()) { - return null; - } - currentProcessedRow++; - } - } Object mappedResult = mapper.mapRow(rs, (int) currentProcessedRow); @@ -224,7 +203,6 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements */ public void mark() { lastCommittedRow = currentProcessedRow; - skippedRows.clear(); } /** @@ -259,8 +237,6 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements JdbcUtils.closeStatement(this.preparedStatement); JdbcUtils.closeConnection(this.con); this.currentProcessedRow = 0; - skippedRows.clear(); - skipCount = 0; } // Check the result set is in synch with the currentRow attribute. This is @@ -366,10 +342,7 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements public void update(ExecutionContext executionContext) { if (saveState) { Assert.notNull(executionContext, "ExecutionContext must not be null"); - String skipped = skippedRows.toString(); - executionContext.putString(getKey(SKIPPED_ROWS), skipped.substring(1, skipped.length() - 1)); executionContext.putLong(getKey(CURRENT_PROCESSED_ROW), currentProcessedRow); - executionContext.putLong(getKey(SKIP_COUNT), skipCount); } } @@ -397,24 +370,6 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", sql, se); } - if (!context.containsKey(getKey(SKIPPED_ROWS))) { - return; - } - - String[] skipped = StringUtils.commaDelimitedListToStringArray(context.getString(getKey(SKIPPED_ROWS))); - for (int i = 0; i < skipped.length; i++) { - this.skippedRows.add(new Long(skipped[i])); - } - } - - /** - * Skip the current row. If the transaction is rolled back, this row will not be represented to the RowMapper when - * read() is called. For example, if you read in row 2, find the data to be bad, and call skip(), then continue - * processing and find - */ - public void skip() { - skippedRows.add(new Long(currentProcessedRow)); - skipCount++; } /** 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 92c3881ec..b9d2b3770 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 @@ -17,8 +17,6 @@ package org.springframework.batch.item.file; import java.io.IOException; -import java.util.HashSet; -import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -29,7 +27,6 @@ import org.springframework.batch.item.ItemReaderException; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ReaderNotOpenException; -import org.springframework.batch.item.Skippable; import org.springframework.batch.item.file.mapping.FieldSet; import org.springframework.batch.item.file.mapping.FieldSetMapper; import org.springframework.batch.item.file.separator.LineReader; @@ -63,22 +60,18 @@ import org.springframework.util.ClassUtils; * @author Robert Kasanicky * @author Dave Syer */ -public class FlatFileItemReader extends ExecutionContextUserSupport implements ItemReader, Skippable, ItemStream, +public class FlatFileItemReader extends ExecutionContextUserSupport implements ItemReader, ItemStream, InitializingBean { private static Log log = LogFactory.getLog(FlatFileItemReader.class); private static final String LINES_READ_COUNT = "lines.read.count"; - private static final String SKIPPED_STATISTICS_NAME = "skipped.lines.count"; - // default encoding for input files public static final String DEFAULT_CHARSET = "ISO-8859-1"; private String encoding = DEFAULT_CHARSET; - private Set skippedLines = new HashSet(); - private Resource resource; private String path; @@ -209,7 +202,6 @@ public class FlatFileItemReader extends ExecutionContextUserSupport implements I if(saveState){ Assert.notNull(executionContext, "ExecutionContext must not be null"); executionContext.putLong(getKey(LINES_READ_COUNT), reader.getPosition()); - executionContext.putLong(getKey(SKIPPED_STATISTICS_NAME), skippedLines.size()); } } @@ -221,7 +213,6 @@ public class FlatFileItemReader extends ExecutionContextUserSupport implements I */ public void mark() { getReader().mark(); - skippedLines.clear(); } /* @@ -234,31 +225,9 @@ public class FlatFileItemReader extends ExecutionContextUserSupport implements I } /** - * Skip the current line which is being processed. + * @return next line to be tokenized and mapped. */ - public void skip() { - Integer count = new Integer(getReader().getPosition()); - // we are not really thread safe so we don't need to synchronize - skippedLines.add(count); - log.debug("Skipping line in template=[" + this + "], line=" + count); - } - - /** - * @return next line to be tokenized and mapped (possibly skips multiple lines). - */ - protected String readLine() { - String line = nextLine(); - - while (line != null && skippedLines.contains(new Integer(getReader().getPosition()))) { - line = nextLine(); - } - return line; - } - - /** - * @return next line from the input - */ - private String nextLine() { + private String readLine() { try { return (String) getReader().read(); } catch (ItemStreamException e) { 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 f463a95f0..58dc0c570 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 @@ -18,7 +18,6 @@ package org.springframework.batch.item.support; import org.springframework.batch.item.AbstractItemReader; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.Skippable; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; @@ -29,7 +28,7 @@ import org.springframework.util.Assert; * * @author Dave Syer */ -public class DelegatingItemReader extends AbstractItemReader implements Skippable, InitializingBean { +public class DelegatingItemReader extends AbstractItemReader implements InitializingBean { private ItemReader itemReader; @@ -69,12 +68,6 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl this.itemReader = source; } - public void skip() { - if (itemReader instanceof Skippable) { - ((Skippable) itemReader).skip(); - } - } - /* * (non-Javadoc) * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext) diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemWriter.java index 65cb1baaa..1cd7c2d47 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/DelegatingItemWriter.java @@ -3,7 +3,6 @@ package org.springframework.batch.item.support; import org.springframework.batch.item.ClearFailedException; import org.springframework.batch.item.FlushFailedException; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.Skippable; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; @@ -13,7 +12,7 @@ import org.springframework.util.Assert; * @author Dave Syer * @author Robert Kasanicky */ -public class DelegatingItemWriter implements ItemWriter, Skippable, InitializingBean { +public class DelegatingItemWriter implements ItemWriter, InitializingBean { private ItemWriter delegate; @@ -78,14 +77,4 @@ public class DelegatingItemWriter implements ItemWriter, Skippable, Initializing delegate.flush(); } - /** - * Delegates to {@link Skippable#skip()} if delegate implements {@link Skippable}. - */ - public void skip() { - if (delegate instanceof Skippable) { - ((Skippable) delegate).skip(); - } - - } - } 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 54fc4f9dc..b710a6fa7 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 @@ -2,8 +2,6 @@ package org.springframework.batch.item.xml; import java.io.IOException; import java.io.InputStream; -import java.util.ArrayList; -import java.util.List; import javax.xml.namespace.QName; import javax.xml.stream.XMLEventReader; @@ -17,7 +15,6 @@ import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ReaderNotOpenException; -import org.springframework.batch.item.Skippable; import org.springframework.batch.item.xml.stax.DefaultFragmentEventReader; import org.springframework.batch.item.xml.stax.DefaultTransactionalEventReader; import org.springframework.batch.item.xml.stax.FragmentEventReader; @@ -31,14 +28,15 @@ import org.springframework.util.ClassUtils; /** * Input source for reading XML input based on StAX. * - * It extracts fragments from the input XML document which correspond to records for processing. The fragments are - * wrapped with StartDocument and EndDocument events so that the fragments can be further processed like standalone XML + * It extracts fragments from the input XML document which correspond to records + * for processing. The fragments are wrapped with StartDocument and EndDocument + * events so that the fragments can be further processed like standalone XML * documents. * * @author Robert Kasanicky */ -public class StaxEventItemReader extends ExecutionContextUserSupport implements ItemReader, Skippable, ItemStream, - InitializingBean { +public class StaxEventItemReader extends ExecutionContextUserSupport implements ItemReader, ItemStream, + InitializingBean { private static final String READ_COUNT_STATISTICS_NAME = "read.count"; @@ -60,8 +58,6 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements private long currentRecordCount = 0; - private List skipRecords = new ArrayList(); - private boolean saveState = false; public StaxEventItemReader() { @@ -80,14 +76,12 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements } Object item = null; - do { - currentRecordCount++; - if (moveCursorToNextFragment(fragmentReader)) { - fragmentReader.markStartFragment(); - item = eventReaderDeserializer.deserializeFragment(fragmentReader); - fragmentReader.markFragmentProcessed(); - } - } while (skipRecords.contains(new Long(currentRecordCount))); + currentRecordCount++; + if (moveCursorToNextFragment(fragmentReader)) { + fragmentReader.markStartFragment(); + item = eventReaderDeserializer.deserializeFragment(fragmentReader); + fragmentReader.markFragmentProcessed(); + } if (item == null) { currentRecordCount--; @@ -103,11 +97,14 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements try { fragmentReader.close(); inputStream.close(); - } catch (XMLStreamException e) { + } + catch (XMLStreamException e) { throw new DataAccessResourceFailureException("Error while closing event reader", e); - } catch (IOException e) { + } + catch (IOException e) { throw new DataAccessResourceFailureException("Error while closing input stream", e); - } finally { + } + finally { fragmentReader = null; inputStream = null; } @@ -119,11 +116,13 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements try { inputStream = resource.getInputStream(); txReader = new DefaultTransactionalEventReader(XMLInputFactory.newInstance().createXMLEventReader( - inputStream)); + inputStream)); fragmentReader = new DefaultFragmentEventReader(txReader); - } catch (XMLStreamException xse) { + } + catch (XMLStreamException xse) { throw new DataAccessResourceFailureException("Unable to create XML reader", xse); - } catch (IOException ioe) { + } + catch (IOException ioe) { throw new DataAccessResourceFailureException("Unable to get input stream", ioe); } initialized = true; @@ -151,35 +150,28 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements } /** - * @param eventReaderDeserializer maps xml fragments corresponding to records to objects + * @param eventReaderDeserializer maps xml fragments corresponding to + * records to objects */ public void setFragmentDeserializer(EventReaderDeserializer eventReaderDeserializer) { this.eventReaderDeserializer = eventReaderDeserializer; } /** - * @param fragmentRootElementName name of the root element of the fragment TODO String can be ambiguous due to - * namespaces, use QName? + * @param fragmentRootElementName name of the root element of the fragment + * TODO String can be ambiguous due to namespaces, use QName? */ public void setFragmentRootElementName(String fragmentRootElementName) { this.fragmentRootElementName = fragmentRootElementName; } /** - * Mark the last read record as 'skipped', so that I will not be returned from read() in the case of a rollback. - * - * @see Skippable#skip() - */ - public void skip() { - skipRecords.add(new Long(currentRecordCount)); - } - - /** - * Ensure that all required dependencies for the ItemReader to run are provided after all properties have been set. + * Ensure that all required dependencies for the ItemReader to run are + * provided after all properties have been set. * * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() - * @throws IllegalArgumentException if the Resource, FragmentDeserializer or FragmentRootElementName is null, or if - * the root element is empty. + * @throws IllegalArgumentException if the Resource, FragmentDeserializer or + * FragmentRootElementName is null, or if the root element is empty. * @throws IllegalStateException if the Resource does not exist. */ public void afterPropertiesSet() throws Exception { @@ -199,12 +191,15 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements } /** - * Responsible for moving the cursor before the StartElement of the fragment root. + * Responsible for moving the cursor before the StartElement of the fragment + * root. * - * This implementation simply looks for the next corresponding element, it does not care about element nesting. You - * will need to override this method to correctly handle composite fragments. + * This implementation simply looks for the next corresponding element, it + * does not care about element nesting. You will need to override this + * method to correctly handle composite fragments. * - * @return true if next fragment was found, false otherwise. + * @return true if next fragment was found, + * false otherwise. */ protected boolean moveCursorToNextFragment(XMLEventReader reader) { try { @@ -218,25 +213,28 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements QName startElementName = ((StartElement) reader.peek()).getName(); if (startElementName.getLocalPart().equals(fragmentRootElementName)) { return true; - } else { + } + else { reader.nextEvent(); } } - } catch (XMLStreamException e) { + } + catch (XMLStreamException e) { throw new DataAccessResourceFailureException("Error while reading from event reader", e); } } /** - * 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. + * 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. * * @see org.springframework.batch.item.AbstractItemReader#mark() */ public void mark() { lastCommitPointRecordCount = currentRecordCount; txReader.onCommit(); - skipRecords = new ArrayList(); } /* @@ -250,6 +248,13 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements fragmentReader.reset(); } + /** + * Set the flag that determines whether to save internal data for + * {@link ExecutionContext}. Only switch this to false if you don't want to + * save any state from this stream, and you don't need it to be restartable. + * + * @param saveState flag value (default true) + */ public void setSaveState(boolean saveState) { this.saveState = saveState; } 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 7f9d47d93..e5d38ca42 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 @@ -3,7 +3,6 @@ package org.springframework.batch.item.database; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.Skippable; import org.springframework.batch.item.sample.Foo; import org.springframework.beans.factory.InitializingBean; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; @@ -162,45 +161,11 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends } /** - * Rollback scenario with skip - input source rollbacks to last commit - * point. - * @throws Exception - */ - public void testRollbackAndSkip() throws Exception { - - if (!(reader instanceof Skippable)) { - return; - } - - Foo foo1 = (Foo) reader.read(); - - commit(); - - Foo foo2 = (Foo) reader.read(); - Assert.state(!foo2.equals(foo1)); - - Foo foo3 = (Foo) reader.read(); - Assert.state(!foo2.equals(foo3)); - - getAsSkippable(reader).skip(); - - rollback(); - - assertEquals(foo2, reader.read()); - Foo foo4 = (Foo) reader.read(); - assertEquals(4, foo4.getValue()); - } - - /** - * Rollback scenario with skip and restart - input source rollbacks to last + * Rollback scenario with restart - input source rollbacks to last * commit point. * @throws Exception */ - public void testRollbackSkipAndRestart() throws Exception { - - if (!(reader instanceof Skippable)) { - return; - } + public void testRollbackAndRestart() throws Exception { getAsItemStream(reader).open(executionContext); @@ -214,8 +179,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends Foo foo3 = (Foo) reader.read(); Assert.state(!foo2.equals(foo3)); - getAsSkippable(reader).skip(); - rollback(); getAsItemStream(reader).update(executionContext); @@ -226,8 +189,7 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends getAsItemStream(reader).open(executionContext); assertEquals(foo2, reader.read()); - Foo foo4 = (Foo) reader.read(); - assertEquals(4, foo4.getValue()); + assertEquals(foo3, reader.read()); } private void commit() { @@ -238,10 +200,6 @@ public abstract class AbstractDataSourceItemReaderIntegrationTests extends reader.reset(); } - private Skippable getAsSkippable(ItemReader source) { - return (Skippable) source; - } - private ItemStream getAsItemStream(ItemReader source) { return (ItemStream) source; } 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 b462f23fc..0cd427654 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 @@ -83,11 +83,11 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { } /** - * Test skip and skipRollback functionality + * Test rollback functionality * * @throws IOException */ - public void testSkip() throws Exception { + public void testReset() throws Exception { reader.close(null); reader.setResource(getInputResource("testLine1\ntestLine2\ntestLine3\ntestLine4\ntestLine5\ntestLine6")); @@ -100,21 +100,13 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { reader.mark(); // read next record reader.read(); // # 3 - // mark record as skipped - reader.skip(); // read next records reader.reset(); - // we should now process all records after first commit point, that are - // not marked as skipped - assertEquals("[testLine4]", reader.read().toString()); + // we should now process all records after first commit point + assertEquals("[testLine3]", reader.read().toString()); - // TODO update - // Map statistics = template.getStatistics(); - // assertEquals("6", - // statistics.get(FlatFileInputTemplate.READ_STATISTICS_NAME)); - // assertEquals("2", - // statistics.get(FlatFileInputTemplate.SKIPPED_STATISTICS_NAME)); + // TODO update and assert ExecutionContext } @@ -123,7 +115,7 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { * * @throws IOException */ - public void testSkipFirstChunk() throws Exception { + public void testFailOnFirstChunk() throws Exception { reader.close(null); reader.setResource(getInputResource("testLine1\ntestLine2\ntestLine3\ntestLine4\ntestLine5\ntestLine6")); @@ -133,8 +125,6 @@ public class FlatFileItemReaderAdvancedTests extends TestCase { reader.read(); // #1 reader.read(); // #2 reader.read(); // #3 - // mark record as skipped - reader.skip(); // rollback reader.reset(); // read next record diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/DelegatingItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/DelegatingItemReaderTests.java index 66d707bc4..85cfa3bff 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/DelegatingItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/DelegatingItemReaderTests.java @@ -22,8 +22,6 @@ import org.springframework.batch.item.AbstractItemReader; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.Skippable; -import org.springframework.batch.item.support.DelegatingItemReader; /** * Unit test for {@link DelegatingItemReader} @@ -71,12 +69,7 @@ public class DelegatingItemReaderTests extends TestCase { assertSame("domain object is provided by the input template", this, result); } - public void testSkip() throws Exception { - itemProvider.skip(); - assertEquals("after skip", itemProvider.read()); - } - - private static class MockItemReader extends AbstractItemReader implements ItemReader, ItemStream, Skippable { + private static class MockItemReader extends AbstractItemReader implements ItemReader, ItemStream { private Object value; @@ -98,10 +91,6 @@ public class DelegatingItemReaderTests extends TestCase { public void open(ExecutionContext executionContext) { } - public void skip() { - value = "after skip"; - } - public void mark() { } 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 93f26e894..2bad8a762 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 @@ -148,20 +148,6 @@ public class StaxEventItemReaderTests extends TestCase { source.update(executionContext); } - /** - * Skipping marked records after rollback. - */ - public void testSkip() { - source.open(executionContext); - List first = (List) source.read(); - source.skip(); - List second = (List) source.read(); - assertFalse(first.equals(second)); - source.reset(); - - assertEquals(second, source.read()); - } - /** * Rollback to last commited record. */