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 f16fff382..c5d67d659 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 @@ -77,6 +77,8 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement private boolean saveState = true; + private boolean forceSync = false; + private boolean shouldDeleteIfExists = true; private boolean shouldDeleteIfEmpty = false; @@ -109,6 +111,19 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement } } + /** + * Flag to indicate that changes should be force-synced to disk on flush. + * Defaults to false, which means that even with a local disk changes could + * be lost if the OS crashes in between a write and a cache flush. Setting + * to true may result in slower performance for usage patterns involving many + * frequent writes. + * + * @param forceSync the flag value to set + */ + public void setForceSync(boolean forceSync) { + this.forceSync = forceSync; + } + /** * Public setter for the line separator. Defaults to the System property * line.separator. @@ -562,7 +577,16 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement */ private Writer getBufferedWriter(FileChannel fileChannel, String encoding) { try { - Writer writer = Channels.newWriter(fileChannel, encoding); + final FileChannel channel = fileChannel; + Writer writer = new BufferedWriter(Channels.newWriter(fileChannel, encoding)) { + @Override + public void flush() throws IOException { + super.flush(); + if (forceSync) { + channel.force(false); + } + } + }; if (transactional) { return new TransactionAwareBufferedWriter(writer, new Runnable() { public void run() { 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 044d632dd..6500448b9 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 @@ -23,6 +23,7 @@ import java.io.IOException; import java.io.OutputStreamWriter; import java.io.UnsupportedEncodingException; import java.io.Writer; +import java.nio.channels.Channels; import java.nio.channels.FileChannel; import java.util.List; import java.util.Map; @@ -139,6 +140,8 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen private boolean transactional = true; + private boolean forceSync; + public StaxEventItemWriter() { setName(ClassUtils.getShortName(StaxEventItemWriter.class)); } @@ -186,6 +189,19 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen this.transactional = transactional; } + /** + * Flag to indicate that changes should be force-synced to disk on flush. + * Defaults to false, which means that even with a local disk changes could + * be lost if the OS crashes in between a write and a cache flush. Setting + * to true may result in slower performance for usage patterns involving + * many frequent writes. + * + * @param forceSync the flag value to set + */ + public void setForceSync(boolean forceSync) { + this.forceSync = forceSync; + } + /** * Get used encoding. * @@ -359,12 +375,14 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen File file; FileOutputStream os = null; + FileChannel fileChannel = null; try { file = resource.getFile(); FileUtils.setUpOutputFile(file, restarted, false, overwriteOutput); Assert.state(resource.exists(), "Output resource must exist"); os = new FileOutputStream(file, true); + fileChannel = os.getChannel(); channel = os.getChannel(); setPosition(position); } @@ -384,24 +402,33 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen } if (outputFactory.isPropertySupported("com.ctc.wstx.outputValidateStructure")) { // On restart we don't write the root element so we have to disable - // structural validation (see: + // structural validation (see: // http://jira.springframework.org/browse/BATCH-1681). outputFactory.setProperty("com.ctc.wstx.outputValidateStructure", Boolean.FALSE); } try { + final FileChannel channel = fileChannel; + Writer writer = new BufferedWriter(new OutputStreamWriter(os, encoding)) { + @Override + public void flush() throws IOException { + super.flush(); + if (forceSync) { + channel.force(false); + } + } + }; if (transactional) { - bufferedWriter = new TransactionAwareBufferedWriter(new OutputStreamWriter(os, encoding), - new Runnable() { - public void run() { - closeStream(); - } - }); + bufferedWriter = new TransactionAwareBufferedWriter(writer, new Runnable() { + public void run() { + closeStream(); + } + }); } else { - bufferedWriter = new BufferedWriter(new OutputStreamWriter(os, encoding)); + bufferedWriter = writer; } - delegateEventWriter = createXmlEventWriter(outputFactory,bufferedWriter); + delegateEventWriter = createXmlEventWriter(outputFactory, bufferedWriter); eventWriter = new NoStartEndDocumentStreamWriter(delegateEventWriter); if (!restarted) { startDocument(delegateEventWriter); @@ -424,7 +451,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen * @return an xml writer * @throws XMLStreamException */ - protected XMLEventWriter createXmlEventWriter(XMLOutputFactory outputFactory,Writer writer) + protected XMLEventWriter createXmlEventWriter(XMLOutputFactory outputFactory, Writer writer) throws XMLStreamException { return outputFactory.createXMLEventWriter(writer); } @@ -434,22 +461,20 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen * @return a factory for the xml output * @throws FactoryConfigurationError */ - protected XMLOutputFactory createXmlOutputFactory() - throws FactoryConfigurationError { + protected XMLOutputFactory createXmlOutputFactory() throws FactoryConfigurationError { return XMLOutputFactory.newInstance(); } - + /** * Subclasses can override to customize the event factory. * @return a factory for the xml events * @throws FactoryConfigurationError */ - protected XMLEventFactory createXmlEventFactory() - throws FactoryConfigurationError { + protected XMLEventFactory createXmlEventFactory() throws FactoryConfigurationError { XMLEventFactory factory = XMLEventFactory.newInstance(); return factory; } - + /** * Subclasses can override to customize the stax result. * @return a result for writing to @@ -613,7 +638,7 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen Assert.state(marshaller.supports(object.getClass()), "Marshaller must support the class of the marshalled object"); Result result = createStaxResult(); - marshaller.marshal(object, result ); + marshaller.marshal(object, result); } try { eventWriter.flush(); 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 6fdc2d867..21e0aed6a 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 @@ -202,6 +202,17 @@ public class FlatFileItemWriterTests { assertEquals(TEST_STRING, lineFromFile); } + @Test + public void testForcedWriteString() throws Exception { + writer.setForceSync(true); + writer.open(executionContext); + writer.write(Collections.singletonList(TEST_STRING)); + writer.close(); + String lineFromFile = readLine(); + + assertEquals(TEST_STRING, lineFromFile); + } + /** * Regular usage of write(String) method * 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 78ed7b85f..703a2783d 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 @@ -90,6 +90,16 @@ public class StaxEventItemWriterTests { assertTrue("Wrong content: " + content, content.contains(TEST_STRING)); } + @Test + public void testWriteAndForceFlush() throws Exception { + writer.setForceSync(true); + writer.open(executionContext); + writer.write(items); + writer.close(); + String content = getOutputFileContent(); + assertTrue("Wrong content: " + content, content.contains(TEST_STRING)); + } + /** * Restart scenario - content is appended to the output file after restart. */