diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java index 4dcd9cc06..7bf4fcea9 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java @@ -214,7 +214,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement public void close() { if (state != null) { try { - if (footerCallback != null) { + if (footerCallback != null && state.outputBufferedWriter != null) { footerCallback.writeFooter(state.outputBufferedWriter); state.outputBufferedWriter.flush(); } @@ -350,7 +350,7 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement } outputBufferedWriter.flush(); - pos = fileChannel.position(); + pos = fileChannel.position() + ((TransactionAwareBufferedWriter)outputBufferedWriter).getBufferSize(); return pos; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java index 61dd85ddd..9a31543f7 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java @@ -126,6 +126,8 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen private StaxWriterCallback footerCallback; + private TransactionAwareBufferedWriter transactionAwareBufferedWriter; + public StaxEventItemWriter() { setName(ClassUtils.getShortName(StaxEventItemWriter.class)); } @@ -315,8 +317,9 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen XMLOutputFactory outputFactory = XMLOutputFactory.newInstance(); try { - delegateEventWriter = outputFactory.createXMLEventWriter(new TransactionAwareBufferedWriter( - new OutputStreamWriter(os, encoding), getName())); + transactionAwareBufferedWriter = new TransactionAwareBufferedWriter( + new OutputStreamWriter(os, encoding), getName()); + delegateEventWriter = outputFactory.createXMLEventWriter(transactionAwareBufferedWriter); eventWriter = new NoStartEndDocumentStreamWriter(delegateEventWriter); if (!restarted) { startDocument(delegateEventWriter); @@ -487,7 +490,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen try { eventWriter.flush(); - position = channel.position(); + position = channel.position() + transactionAwareBufferedWriter.getBufferSize(); } catch (Exception e) { throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", e); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java index 6dc5dc130..5a99404f3 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java @@ -81,6 +81,19 @@ public class TransactionAwareBufferedWriter extends Writer { } + /** + * Convenience method for clients to determine if there is any unflushed + * data. + * + * @return the current size of unflushed buffered data + */ + public long getBufferSize() { + if (!transactionActive()) { + return 0L; + } + return getCurrentBuffer().length(); + } + /** * @return */ diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java index f3daf488c..6745f1fa5 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java @@ -39,9 +39,15 @@ import org.junit.Before; import org.junit.Test; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.UnexpectedInputException; import org.springframework.batch.item.file.transform.LineAggregator; import org.springframework.batch.item.file.transform.PassThroughLineAggregator; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.core.io.FileSystemResource; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.ClassUtils; /** @@ -246,6 +252,72 @@ public class FlatFileItemWriterTests { } + @Test + public void testTransactionalRestart() throws Exception { + + writer.setFooterCallback(new FlatFileFooterCallback() { + + public void writeFooter(Writer writer) throws IOException { + writer.write("footer"); + } + + }); + + writer.open(executionContext); + + PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); + + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + try { + // write some lines + writer.write(Arrays.asList(new String[] { "testLine1", "testLine2", "testLine3" })); + // write more lines + writer.write(Arrays.asList(new String[] { "testLine4", "testLine5" })); + } + catch (Exception e) { + throw new UnexpectedInputException("Could not write data", e); + } + // get restart data + writer.update(executionContext); + return null; + } + }); + // close template + writer.close(); + + // init with correct data + writer.open(executionContext); + + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + try { + // write more lines + writer.write(Arrays.asList(new String[] { "testLine6", "testLine7", "testLine8" })); + } + catch (Exception e) { + throw new UnexpectedInputException("Could not write data", e); + } + // get restart data + writer.update(executionContext); + return null; + } + }); + // close template + writer.close(); + + // verify what was written to the file + for (int i = 1; i <= 8; i++) { + assertEquals("testLine" + i, readLine()); + } + + assertEquals("footer", readLine()); + + // 3 lines were written to the file after restart + assertEquals(3, executionContext.getLong(ClassUtils.getShortName(FlatFileItemWriter.class) + ".written")); + + } + @Test public void testOpenWithNonWritableFile() throws Exception { writer = new FlatFileItemWriter(); @@ -254,9 +326,9 @@ public class FlatFileItemWriterTests { writer.setResource(file); new File(file.getFile().getParent()).mkdirs(); file.getFile().createNewFile(); - assertTrue("Test file must exist: "+file, file.exists()); - assertTrue("Test file set to read-only: "+file, file.getFile().setReadOnly()); - assertFalse("Should be readonly file: "+file, file.getFile().canWrite()); + assertTrue("Test file must exist: " + file, file.exists()); + assertTrue("Test file set to read-only: " + file, file.getFile().setReadOnly()); + assertFalse("Should be readonly file: " + file, file.getFile().canWrite()); writer.afterPropertiesSet(); try { writer.open(executionContext); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemWriterTests.java index 11e0723d5..549412eb7 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemWriterTests.java @@ -22,10 +22,16 @@ import org.apache.commons.io.FileUtils; import org.junit.Before; import org.junit.Test; import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; import org.springframework.oxm.Marshaller; import org.springframework.oxm.XmlMappingException; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; @@ -99,13 +105,60 @@ public class StaxEventItemWriterTests { assertTrue(outputFile.contains("" + TEST_STRING + TEST_STRING + "")); } + @Test + public void testTransactionalRestart() throws Exception { + writer.open(executionContext); + + PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); + + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + try { + // write item + writer.write(items); + } + catch (Exception e) { + throw new UnexpectedInputException("Could not write data", e); + } + // get restart data + writer.update(executionContext); + return null; + } + }); + writer.close(); + + // create new writer from saved restart data and continue writing + writer = createItemWriter(); + writer.open(executionContext); + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + try { + writer.write(items); + } + catch (Exception e) { + throw new UnexpectedInputException("Could not write data", e); + } + // get restart data + writer.update(executionContext); + return null; + } + }); + writer.close(); + + // check the output is concatenation of 'before restart' and 'after + // restart' writes. + String outputFile = outputFileContent(); + assertEquals(2, StringUtils.countOccurrencesOf(outputFile, TEST_STRING)); + assertTrue(outputFile.contains("" + TEST_STRING + TEST_STRING + "")); + } + /** * Item is written to the output file only after flush. */ @Test public void testWriteWithHeader() throws Exception { - - writer.setHeaderCallback(new StaxWriterCallback(){ + + writer.setHeaderCallback(new StaxWriterCallback() { public void write(XMLEventWriter writer) throws IOException { XMLEventFactory factory = XMLEventFactory.newInstance(); @@ -116,9 +169,9 @@ public class StaxEventItemWriterTests { catch (XMLStreamException e) { throw new RuntimeException(e); } - + } - + }); writer.open(executionContext); writer.write(items); @@ -150,7 +203,7 @@ public class StaxEventItemWriterTests { */ @Test public void testOpenAndClose() throws Exception { - writer.setHeaderCallback(new StaxWriterCallback(){ + writer.setHeaderCallback(new StaxWriterCallback() { public void write(XMLEventWriter writer) throws IOException { XMLEventFactory factory = XMLEventFactory.newInstance(); @@ -161,9 +214,9 @@ public class StaxEventItemWriterTests { catch (XMLStreamException e) { throw new RuntimeException(e); } - + } - + }); writer.setFooterCallback(new StaxWriterCallback() { @@ -176,16 +229,16 @@ public class StaxEventItemWriterTests { catch (XMLStreamException e) { throw new RuntimeException(e); } - + } - + }); writer.setRootTagName("testroot"); writer.setRootElementAttributes(Collections. singletonMap("attribute", "value")); writer.open(executionContext); writer.close(); String content = outputFileContent(); - + assertTrue(content.contains("")); assertTrue(content.contains("
")); assertTrue(content.contains("")); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java index 188bc5edc..d11a6d2fa 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java @@ -15,15 +15,15 @@ */ package org.springframework.batch.support.transaction; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import java.io.IOException; import java.io.StringWriter; import java.io.Writer; import org.junit.Test; -import org.springframework.batch.support.transaction.ResourcelessTransactionManager; -import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; @@ -56,6 +56,12 @@ public class TransactionAwareBufferedWriterTests { assertEquals("foo", stringWriter.toString()); } + @Test + public void testBufferSizeOutsideTransaction() throws Exception { + writer.write("foo"); + assertEquals(0, writer.getBufferSize()); + } + @Test public void testCloseOutsideTransaction() throws Exception { writer.write("foo"); @@ -114,6 +120,22 @@ public class TransactionAwareBufferedWriterTests { assertEquals("foo", stringWriter.toString()); } + @Test + public void tesBufferSizeInTransaction() throws Exception { + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + try { + writer.write("foo"); + } + catch (IOException e) { + throw new IllegalStateException("Unexpected IOException", e); + } + assertEquals(3, writer.getBufferSize()); + return null; + } + }); + } + @Test public void testWriteWithRollback() throws Exception { try {