This class will be updated in the future to use a buffering approach + * to handling transactions, rather than outputting directly to the file and + * truncating on rollback
+ * * @author Waseem Malik * @author Tomas Slanina * @author Robert Kasanicky @@ -465,23 +469,31 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements return true; } - /* + /* To be deleted once interface changes are complete * (non-Javadoc) * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#mark(org.springframework.batch.item.ExecutionContext) */ public void mark() { - getOutputState().mark(); + } - /* + /* To be deleted once interface changes are complete * (non-Javadoc) * @see org.springframework.batch.io.support.AbstractTransactionalIoSource#reset(org.springframework.batch.item.ExecutionContext) */ public void reset() throws ResetFailedException { + + } + + public void clear() throws Exception { try { getOutputState().reset(); } catch (BatchCriticalException e) { throw new ResetFailedException(e); } } + + public void flush() throws Exception { + getOutputState().mark(); + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/HibernateAwareItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/HibernateAwareItemWriter.java index f51d2e33c..42171f378 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/HibernateAwareItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/HibernateAwareItemWriter.java @@ -146,32 +146,7 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini * @see org.springframework.batch.repeat.RepeatListener#close(org.springframework.batch.repeat.RepeatContext) */ public void close(RepeatContext context) { - try { - if (delegate instanceof RepeatListener) { - RepeatListener interceptor = (RepeatListener) delegate; - interceptor.close(context); - } - flush(); - } catch (RuntimeException e) { - synchronized (failed) { - failed.addAll(getProcessed()); - } - // onError will not be called after close() by the framework so we - // have to do it here. - onError(context, e); - throw e; - } - unsetContext(); - } - /** - * Wrapper for Hibernate flush. - */ - private void flush() { - hibernateTemplate.flush(); - // This should happen when the transaction commits anyway, but to be - // sure... - hibernateTemplate.clear(); } /** @@ -246,7 +221,7 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini * * @return the context */ - private void flushIfNecessary(Object output) { + private void flushIfNecessary(Object output) throws Exception{ RepeatContext context = (RepeatContext) TransactionSynchronizationManager.getResource(WRITER_REPEAT_CONTEXT); boolean flush; synchronized (failed) { @@ -264,4 +239,34 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini } + public void clear() throws Exception { + if(delegate != null){ + delegate.clear(); + } + hibernateTemplate.clear(); + } + + /** + * Flush the Hibernate session. The delegate flush will also be called before finishing. + */ + public void flush() throws Exception { + try { + if (delegate != null) { + delegate.flush(); + } + hibernateTemplate.flush(); + // This should happen when the transaction commits anyway, but to be + // sure... + hibernateTemplate.clear(); + } catch (RuntimeException e) { + synchronized (failed) { + failed.addAll(getProcessed()); + } + // This used to contain a call to onError, however, I think this + // should be handled within the step. + throw e; + } + unsetContext(); + } + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java index 786193848..3a59cf6a1 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java @@ -438,20 +438,27 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, Initializing return true; } - /* + /* TODO remove once ItemStream interface is modified. * (non-Javadoc) * @see org.springframework.batch.item.ItemStream#mark(org.springframework.batch.item.ExecutionContext) */ public void mark() { + } + + public void flush() throws Exception { lastCommitPointPosition = getPosition(); lastCommitPointRecordCount = currentRecordCount; } - /* + /* TODO remove once ItemStream interface is modified. * (non-Javadoc) * @see org.springframework.batch.item.ItemStream#reset(org.springframework.batch.item.ExecutionContext) */ public void reset() { + + } + + public void clear() throws Exception { currentRecordCount = lastCommitPointRecordCount; // close output close(); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemWriter.java index 78c1d5150..3086ebc2c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemWriter.java @@ -17,10 +17,18 @@ package org.springframework.batch.item; /** - * Basic interface for generic output operations. Class implementing this + *Basic interface for generic output operations. Class implementing this * interface will be responsible for serializing objects as necessary. * Generally, it is responsibility of implementing class to decide which - * technology to use for mapping and how it should be configured. + * technology to use for mapping and how it should be configured.
+ * + *+ * Due to the nature of batch processing, it is expected that most writers + * will buffer output. A flush method is provided to the interface in order + * to ensure that any buffers can be flushed before a transaction is + * committed. Along the same lines, if a transaction has been rolled back, + * then the contents of any buffers should be thrown away. + *
* * @author Dave Syer * @author Lucas Ward @@ -38,4 +46,19 @@ public interface ItemWriter { */ public void write(Object item) throws Exception; + /** + * Flush any buffers that are being held. This will usually be performed + * prior to committing any transactions. + * + * @throws Exception + */ + public void flush() throws Exception; + + /** + * Clear any buffers that are being held. This will usually be performed + * prior to rolling back any transactions. + * + * @throws Exception + */ + public void clear() throws Exception; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/AbstractItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/AbstractItemWriter.java index cd602786f..876aac4b0 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/AbstractItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/AbstractItemWriter.java @@ -19,11 +19,17 @@ import org.springframework.batch.item.ItemWriter; /** * Abstract {@link ItemWriter} that allows for base classes to only - * implement the close method if they need it. + * implement the close method if they need it. Because it is likely + * that the flush and clear methods may not need to be implemented, + * they are provided in this class. * * @author Lucas Ward - * */ public abstract class AbstractItemWriter implements ItemWriter{ - + + public void flush() throws Exception { + } + + public void clear() throws Exception { + } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java index 2742a8de6..77385fdda 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/DelegatingItemWriter.java @@ -1,6 +1,5 @@ package org.springframework.batch.item.writer; -import org.springframework.batch.io.Skippable; import org.springframework.batch.item.ItemWriter; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; @@ -11,7 +10,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 writer; @@ -43,14 +42,22 @@ public class DelegatingItemWriter implements ItemWriter, Skippable, Initializing this.writer = writer; } - public void skip() { - if (writer instanceof Skippable) { - ((Skippable) writer).skip(); - } - } - public void afterPropertiesSet() throws Exception { Assert.notNull(writer); } + /** + * Delegates to {@link ItemWriter#clear()} + */ + public void clear() throws Exception { + writer.clear(); + } + + /** + * Delegates to {@link ItemWriter#flush()} + */ + public void flush() throws Exception { + writer.flush(); + } + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/ItemWriterAdapter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/ItemWriterAdapter.java index c62257fe7..12560cf18 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/ItemWriterAdapter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/ItemWriterAdapter.java @@ -37,11 +37,15 @@ public class ItemWriterAdapter extends AbstractMethodInvokingDelegator implement /* * No-op, can't call more than one method. * - * (non-Javadoc) - * @see org.springframework.batch.item.ItemWriter#close() */ - public void close() throws Exception { - + public void clear() throws Exception { + } + + /* + * No-op, can't call more than one method. + * + */ + public void flush() throws Exception { } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/PropertyExtractingDelegatingItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/PropertyExtractingDelegatingItemWriter.java index 4af027cd2..a0e9f1d1b 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/PropertyExtractingDelegatingItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/writer/PropertyExtractingDelegatingItemWriter.java @@ -67,6 +67,10 @@ public class PropertyExtractingDelegatingItemWriter extends AbstractMethodInvoki } - public void close() throws Exception { + public void clear() throws Exception { + } + + + public void flush() throws Exception { } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java index 5d65cacef..f31f93533 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/FlatFileItemWriterTests.java @@ -24,7 +24,6 @@ import java.util.Collections; import junit.framework.TestCase; -import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.writer.ItemTransformer; import org.springframework.core.io.FileSystemResource; @@ -302,12 +301,12 @@ public class FlatFileItemWriterTests extends TestCase { assertEquals(0, streamContext.getLong(FlatFileItemWriter.RESTART_DATA_NAME)); } - private void commit() { - ((ItemStream) inputSource).mark(); + private void commit() throws Exception{ + inputSource.flush(); } - private void rollback() { - ((ItemStream) inputSource).reset(); + private void rollback() throws Exception{ + inputSource.clear(); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/HibernateAwareItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/HibernateAwareItemWriterTests.java index 22d45783e..3216051b1 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/HibernateAwareItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/support/HibernateAwareItemWriterTests.java @@ -73,6 +73,14 @@ public class HibernateAwareItemWriterTests extends TestCase { public void close() throws Exception { } + + public void clear() throws Exception { + list.add("clear"); + } + + public void flush() throws Exception { + list.add("flush"); + } } HibernateAwareItemWriter writer = new HibernateAwareItemWriter(); @@ -148,7 +156,7 @@ public class HibernateAwareItemWriterTests extends TestCase { * Test method for * {@link org.springframework.batch.io.support.HibernateAwareItemWriter#write(java.lang.Object)}. */ - public void testCloseWithFailure() { + public void testCloseWithFailure() throws Exception{ final RuntimeException ex = new RuntimeException("bar"); writer.setHibernateTemplate(new HibernateTemplate() { public void flush() throws DataAccessException { @@ -156,14 +164,13 @@ public class HibernateAwareItemWriterTests extends TestCase { } }); try { - writer.close(context); + writer.flush(); fail("Expected RuntimeException"); } catch (RuntimeException e) { assertEquals("bar", e.getMessage()); } - assertEquals(2, list.size()); - assertTrue(list.contains(ex)); - assertTrue(list.contains(context)); + assertEquals(1, list.size()); + assertTrue(list.contains("flush")); } /** @@ -180,14 +187,13 @@ public class HibernateAwareItemWriterTests extends TestCase { }); writer.write("foo"); try { - writer.close(context); + writer.flush(); fail("Expected RuntimeException"); } catch (RuntimeException e) { assertEquals("bar", e.getMessage()); } - assertEquals(3, list.size()); - assertTrue(list.contains(ex)); - assertTrue(list.contains(context)); + assertEquals(2, list.size()); + assertTrue(list.contains("flush")); writer.setHibernateTemplate(new HibernateTemplateWrapper() { public void flush() throws DataAccessException { list.add("flush"); @@ -221,8 +227,8 @@ public class HibernateAwareItemWriterTests extends TestCase { * Test method for * {@link org.springframework.batch.io.support.HibernateAwareItemWriter#close(org.springframework.batch.repeat.RepeatContext)}. */ - public void testClose() { - writer.close(context); + public void testFlush() throws Exception{ + writer.flush(); assertEquals(3, list.size()); assertTrue(list.contains("flush")); } @@ -231,11 +237,11 @@ public class HibernateAwareItemWriterTests extends TestCase { * Test method for * {@link org.springframework.batch.io.support.HibernateAwareItemWriter#close(org.springframework.batch.repeat.RepeatContext)}. */ - public void testCloseAfterClear() { + public void testCloseAfterClear() throws Exception{ Map map = TransactionSynchronizationManager.getResourceMap(); String key = (String) map.keySet().iterator().next(); TransactionSynchronizationManager.unbindResource(key); - writer.close(context); + writer.flush(); assertEquals(3, list.size()); assertTrue(list.contains("flush")); assertTrue(list.contains("clear")); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java similarity index 93% rename from spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java rename to spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java index 4406603bb..780fc69ea 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventWriterItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/xml/StaxEventItemWriterTests.java @@ -23,7 +23,7 @@ import org.springframework.xml.transform.StaxResult; /** * Tests for {@link StaxStreamWriterOutputSource}. */ -public class StaxEventWriterItemWriterTests extends TestCase { +public class StaxEventItemWriterTests extends TestCase { // object under test private StaxEventItemWriter writer; @@ -67,7 +67,7 @@ public class StaxEventWriterItemWriterTests extends TestCase { public void testRollback() throws Exception { writer.write(record); // rollback - writer.reset(); + writer.clear(); assertEquals("", outputFileContent()); } @@ -77,7 +77,7 @@ public class StaxEventWriterItemWriterTests extends TestCase { public void testCommit() throws Exception { writer.write(record); // commit - writer.mark(); + writer.flush(); assertTrue(outputFileContent().contains(TEST_STRING)); } @@ -126,7 +126,7 @@ public class StaxEventWriterItemWriterTests extends TestCase { /** * Open method writes the root tag, close method adds corresponding end tag. */ - public void testOpenAndClose() throws IOException { + public void testOpenAndClose() throws Exception { writer.setRootTagName("testroot"); writer.setRootElementAttributes(new HashMap() { { @@ -134,7 +134,7 @@ public class StaxEventWriterItemWriterTests extends TestCase { } }); writer.open(); - writer.mark(); + writer.flush(); assertTrue(outputFileContent().indexOf("