From 36e3bda486ea9d4739033fbb17135a19b6f2d89f Mon Sep 17 00:00:00 2001 From: jpraet Date: Sat, 29 Jun 2013 13:54:24 +0200 Subject: [PATCH] BATCH-2052: StaxEventItemWriter should only force sync once per chunk --- .../batch/item/file/FlatFileItemWriter.java | 3 +- .../batch/item/xml/StaxEventItemWriter.java | 26 ++-- .../TransactionAwareBufferedWriter.java | 30 +++- .../TransactionAwareBufferedWriterTests.java | 132 ++++++++++++------ 4 files changed, 129 insertions(+), 62 deletions(-) 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 62fd65f5e..2664c102c 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 @@ -614,6 +614,7 @@ InitializingBean { }); writer.setEncoding(encoding); + writer.setForceSync(forceSync); return writer; } else { @@ -627,7 +628,7 @@ InitializingBean { } }; - return new BufferedWriter(writer); + return writer; } } catch (UnsupportedCharsetException ucse) { 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 30dd1840f..9b4468fae 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 @@ -395,7 +395,6 @@ ResourceAwareItemWriterItemStream, InitializingBean { /** * Helper method for opening output source at given file position */ - @SuppressWarnings("resource") private void open(long position, boolean restarted) { File file; @@ -443,25 +442,20 @@ ResourceAwareItemWriterItemStream, InitializingBean { }); writer.setEncoding(encoding); + writer.setForceSync(forceSync); bufferedWriter = writer; } else { - Writer writer = new BufferedWriter(new OutputStreamWriter(os, encoding)) { - @Override - public void flush() throws IOException { - super.flush(); - if (forceSync) { - channel.force(false); - } - } - }; - bufferedWriter = writer; + bufferedWriter = new BufferedWriter(new OutputStreamWriter(os, encoding)); } delegateEventWriter = createXmlEventWriter(outputFactory, bufferedWriter); eventWriter = new NoStartEndDocumentStreamWriter(delegateEventWriter); initNamespaceContext(delegateEventWriter); if (!restarted) { startDocument(delegateEventWriter); + if (forceSync) { + channel.force(false); + } } } catch (XMLStreamException xse) { @@ -470,8 +464,10 @@ ResourceAwareItemWriterItemStream, InitializingBean { catch (UnsupportedEncodingException e) { throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "] with encoding=[" + encoding + "]", e); + } + catch (IOException e) { + throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", e); } - } /** @@ -716,9 +712,15 @@ ResourceAwareItemWriterItemStream, InitializingBean { } try { eventWriter.flush(); + if (forceSync) { + channel.force(false); + } } catch (XMLStreamException e) { throw new WriteFailedException("Failed to flush the events", e); + } + catch (IOException e) { + throw new WriteFailedException("Failed to flush the events", 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 5074fd49b..5bd5d978a 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 @@ -48,12 +48,14 @@ public class TransactionAwareBufferedWriter extends Writer { private FileChannel channel; private final Runnable closeCallback; - + // default encoding for writing to output files - set to UTF-8. private static final String DEFAULT_CHARSET = "UTF-8"; - + private String encoding = DEFAULT_CHARSET; - + + private boolean forceSync = false; + /** * Create a new instance with the underlying file channel provided, and a callback * to execute on close. The callback should clean up related resources like @@ -74,6 +76,19 @@ public class TransactionAwareBufferedWriter extends Writer { this.encoding = encoding; } + /** + * 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; + } + /** * @return */ @@ -88,7 +103,7 @@ public class TransactionAwareBufferedWriter extends Writer { public void afterCompletion(int status) { clear(); } - + @Override public void beforeCommit(boolean readOnly) { try { @@ -112,6 +127,9 @@ public class TransactionAwareBufferedWriter extends Writer { if(bytesWritten != bufferLength) { throw new IOException("All bytes to be written were not successfully written"); } + if (forceSync) { + channel.force(false); + } if (TransactionSynchronizationManager.hasResource(closeKey)) { closeCallback.run(); } @@ -145,7 +163,7 @@ public class TransactionAwareBufferedWriter extends Writer { if (!transactionActive()) { return 0L; } - try { + try { return getCurrentBuffer().toString().getBytes(encoding).length; } catch (UnsupportedEncodingException e) { throw new WriteFailedException("Could not determine buffer size because of unsupported encoding: " + encoding, e); @@ -182,7 +200,7 @@ public class TransactionAwareBufferedWriter extends Writer { */ @Override public void flush() throws IOException { - if (!transactionActive()) { + if (!transactionActive() && forceSync) { channel.force(false); } } 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 d562543e3..857532d84 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 @@ -19,6 +19,9 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; import static org.mockito.Matchers.anyObject; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.IOException; @@ -26,14 +29,14 @@ import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; import org.mockito.ArgumentCaptor; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionTemplate; -//import org.easymock.Capture; /** * @author Dave Syer @@ -46,13 +49,13 @@ public class TransactionAwareBufferedWriterTests { private FileChannel fileChannel; private TransactionAwareBufferedWriter writer; - + @Before public void init() { fileChannel = mock(FileChannel.class); - + writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { - @Override + @Override public void run() { try { ByteBuffer bb = ByteBuffer.wrap("c".getBytes()); @@ -63,7 +66,7 @@ public class TransactionAwareBufferedWriterTests { } } }); - + writer.setEncoding("UTF-8"); } @@ -77,45 +80,63 @@ public class TransactionAwareBufferedWriterTests { */ @Test public void testWriteOutsideTransaction() throws Exception { -// Capture bb = new Capture(); ArgumentCaptor bb = ArgumentCaptor.forClass(ByteBuffer.class); -// when(fileChannel.write(capture(bb))).thenReturn(3); when(fileChannel.write(bb.capture())).thenReturn(3); - fileChannel.force(false); writer.write("foo"); writer.flush(); // Not closed yet - + String s = getStringFromByteBuffer(bb.getValue()); - + assertEquals("foo", s); + + verify(fileChannel, never()).force(false); } @Test - public void testBufferSizeOutsideTransaction() throws Exception { -// Capture bb = new Capture(); + public void testWriteOutsideTransactionForceSync() throws Exception { + writer.setForceSync(true); ArgumentCaptor bb = ArgumentCaptor.forClass(ByteBuffer.class); when(fileChannel.write(bb.capture())).thenReturn(3); writer.write("foo"); - + writer.flush(); + // Not closed yet + + String s = getStringFromByteBuffer(bb.getValue()); + + assertEquals("foo", s); + + verify(fileChannel, times(1)).force(false); + } + + @Test + public void testBufferSizeOutsideTransaction() throws Exception { + ArgumentCaptor bb = ArgumentCaptor.forClass(ByteBuffer.class); + when(fileChannel.write(bb.capture())).thenReturn(3); + + writer.write("foo"); + assertEquals(0, writer.getBufferSize()); } - - @Ignore //TODO - need to fix capture test + @Test public void testCloseOutsideTransaction() throws Exception { - ArgumentCaptor writeBuffer = ArgumentCaptor.forClass(ByteBuffer.class); - ArgumentCaptor commitBuffer = ArgumentCaptor.forClass(ByteBuffer.class); - when(fileChannel.write(writeBuffer.capture())).thenReturn(4); - when(fileChannel.write(commitBuffer.capture())).thenReturn(1); + ArgumentCaptor byteBufferCaptor = ArgumentCaptor.forClass(ByteBuffer.class); + + when(fileChannel.write(byteBufferCaptor.capture())).thenAnswer(new Answer() { + @Override + public Integer answer(InvocationOnMock invocation) throws Throwable { + return ((ByteBuffer) invocation.getArguments()[0]).remaining(); + } + }); writer.write("foo"); writer.close(); - - assertEquals("foo", getStringFromByteBuffer(writeBuffer.getValue())); - assertEquals("c", getStringFromByteBuffer(commitBuffer.getValue())); + + assertEquals("foo", getStringFromByteBuffer(byteBufferCaptor.getAllValues().get(0))); + assertEquals("c", getStringFromByteBuffer(byteBufferCaptor.getAllValues().get(1))); } @Test @@ -124,7 +145,7 @@ public class TransactionAwareBufferedWriterTests { when(fileChannel.write((ByteBuffer)anyObject())).thenReturn(3); new TransactionTemplate(transactionManager).execute(new TransactionCallback() { - @Override + @Override public Object doInTransaction(TransactionStatus status) { try { writer.write("foo"); @@ -137,7 +158,32 @@ public class TransactionAwareBufferedWriterTests { return null; } }); - + + verify(fileChannel, never()).force(false); + } + + @Test + @SuppressWarnings({"unchecked", "rawtypes"}) + public void testFlushInTransactionForceSync() throws Exception { + writer.setForceSync(true); + when(fileChannel.write((ByteBuffer)anyObject())).thenReturn(3); + + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + @Override + public Object doInTransaction(TransactionStatus status) { + try { + writer.write("foo"); + writer.flush(); + } + catch (IOException e) { + throw new IllegalStateException("Unexpected IOException", e); + } + assertEquals(3, writer.getBufferSize()); + return null; + } + }); + + verify(fileChannel, times(1)).force(false); } @Test @@ -145,9 +191,9 @@ public class TransactionAwareBufferedWriterTests { public void testWriteWithCommit() throws Exception { ArgumentCaptor bb = ArgumentCaptor.forClass(ByteBuffer.class); when(fileChannel.write(bb.capture())).thenReturn(3); - + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { - @Override + @Override public Object doInTransaction(TransactionStatus status) { try { writer.write("foo"); @@ -159,7 +205,7 @@ public class TransactionAwareBufferedWriterTests { return null; } }); - + assertEquals(0, writer.getBufferSize()); } @@ -170,7 +216,7 @@ public class TransactionAwareBufferedWriterTests { when(fileChannel.write(bb.capture())).thenReturn(3); new TransactionTemplate(transactionManager).execute(new TransactionCallback() { - @Override + @Override public Object doInTransaction(TransactionStatus status) { try { writer.write("foo"); @@ -182,10 +228,10 @@ public class TransactionAwareBufferedWriterTests { return null; } }); - + assertEquals(0, writer.getBufferSize()); } - + @Test @SuppressWarnings({"unchecked", "rawtypes"}) // BATCH-1959 @@ -194,7 +240,7 @@ public class TransactionAwareBufferedWriterTests { when(fileChannel.write(bb.capture())).thenReturn(5); new TransactionTemplate(transactionManager).execute(new TransactionCallback() { - @Override + @Override public Object doInTransaction(TransactionStatus status) { try { writer.write("fóó"); @@ -206,21 +252,21 @@ public class TransactionAwareBufferedWriterTests { return null; } }); - + assertEquals(0, writer.getBufferSize()); - } + } @Test @SuppressWarnings({"unchecked", "rawtypes"}) // BATCH-1959 public void testBufferSizeInTransactionWithMultiByteCharacterUTF16BE() throws Exception { writer.setEncoding("UTF-16BE"); - + ArgumentCaptor bb = ArgumentCaptor.forClass(ByteBuffer.class); when(fileChannel.write(bb.capture())).thenReturn(6); new TransactionTemplate(transactionManager).execute(new TransactionCallback() { - @Override + @Override public Object doInTransaction(TransactionStatus status) { try { writer.write("fóó"); @@ -232,16 +278,16 @@ public class TransactionAwareBufferedWriterTests { return null; } }); - + assertEquals(0, writer.getBufferSize()); - } + } @Test @SuppressWarnings({"unchecked", "rawtypes"}) public void testWriteWithRollback() throws Exception { try { new TransactionTemplate(transactionManager).execute(new TransactionCallback() { - @Override + @Override public Object doInTransaction(TransactionStatus status) { try { writer.write("foo"); @@ -267,19 +313,19 @@ public class TransactionAwareBufferedWriterTests { testWriteWithRollback(); testWriteWithCommit(); } - + @Test @SuppressWarnings({"unchecked", "rawtypes"}) public void testExceptionOnFlush() throws Exception { writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { - @Override + @Override public void run() { } }); try { new TransactionTemplate(transactionManager).execute(new TransactionCallback() { - @Override + @Override public Object doInTransaction(TransactionStatus status) { try { writer.write("foo"); @@ -290,7 +336,7 @@ public class TransactionAwareBufferedWriterTests { return null; } }); - + fail("Exception was not thrown"); } catch (FlushFailedException ffe) { assertEquals("Could not write to output buffer", ffe.getMessage());