From 5d879ab27b6521a10ca0c48578b08a7dc51e695f Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Thu, 8 Nov 2012 14:15:10 -0600 Subject: [PATCH 1/4] BATCH-1799: Updated to allow exceptions to be bubbled during flush/close --- .../TransactionAwareBufferedWriter.java | 15 ++-- .../TransactionAwareBufferedWriterTests.java | 80 ++++++++++++++++--- 2 files changed, 78 insertions(+), 17 deletions(-) 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 d5de7014f..1efc62214 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2012 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. @@ -28,6 +28,7 @@ import org.springframework.transaction.support.TransactionSynchronizationManager * underlying writer only when the transaction is committed. * * @author Dave Syer + * @author Michael Minella * */ public class TransactionAwareBufferedWriter extends Writer { @@ -72,17 +73,17 @@ public class TransactionAwareBufferedWriter extends Writer { TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { @Override public void afterCompletion(int status) { + clear(); + } + + @Override + public void afterCommit() { try { - if (status == STATUS_COMMITTED) { - complete(); - } + complete(); } catch (IOException e) { throw new FlushFailedException("Could not write to output buffer", e); } - finally { - clear(); - } } private void complete() throws IOException { 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 de834facc..9d3711a38 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2012 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. @@ -18,11 +18,13 @@ package org.springframework.batch.support.transaction; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.io.IOException; import java.io.StringWriter; import java.io.Writer; +import org.junit.Before; import org.junit.Test; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; @@ -31,22 +33,28 @@ import org.springframework.transaction.support.TransactionTemplate; /** * @author Dave Syer + * @author Michael Minella * */ public class TransactionAwareBufferedWriterTests { private Writer stringWriter = new StringWriter(); - private TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(stringWriter, new Runnable() { - public void run() { - try { - stringWriter.append("c"); + private TransactionAwareBufferedWriter writer; + + @Before + public void init() { + writer = new TransactionAwareBufferedWriter(stringWriter, new Runnable() { + public void run() { + try { + stringWriter.append("c"); + } + catch (IOException e) { + throw new IllegalStateException(e); + } } - catch (IOException e) { - throw new IllegalStateException(e); - } - } - }); + }); + } private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); @@ -80,6 +88,7 @@ public class TransactionAwareBufferedWriterTests { } @Test + @SuppressWarnings({"unchecked", "rawtypes"}) public void testFlushInTransaction() throws Exception { Writer mock = new Writer() { @Override @@ -117,6 +126,7 @@ public class TransactionAwareBufferedWriterTests { } @Test + @SuppressWarnings({"unchecked", "rawtypes"}) public void testWriteWithCommit() throws Exception { new TransactionTemplate(transactionManager).execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { @@ -135,6 +145,7 @@ public class TransactionAwareBufferedWriterTests { } @Test + @SuppressWarnings({"unchecked", "rawtypes"}) public void tesBufferSizeInTransaction() throws Exception { new TransactionTemplate(transactionManager).execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { @@ -151,6 +162,7 @@ public class TransactionAwareBufferedWriterTests { } @Test + @SuppressWarnings({"unchecked", "rawtypes"}) public void testWriteWithRollback() throws Exception { try { new TransactionTemplate(transactionManager).execute(new TransactionCallback() { @@ -179,5 +191,53 @@ public class TransactionAwareBufferedWriterTests { testWriteWithRollback(); testWriteWithCommit(); } + + @Test + @SuppressWarnings({"unchecked", "rawtypes"}) + public void testExceptionOnFlush() throws Exception { + final Writer badWriter = new Writer() { + + @Override + public void write(char[] cbuf, int off, int len) throws IOException { + } + + @Override + public void flush() throws IOException { + throw new IOException("This should be bubbled"); + } + + @Override + public void close() throws IOException { + } + }; + writer = new TransactionAwareBufferedWriter(badWriter, new Runnable() { + public void run() { + try { + badWriter.append("c"); + } + catch (IOException e) { + throw new IllegalStateException(e); + } + } + }); + try { + 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("", stringWriter.toString()); + return null; + } + }); + + fail("Exception was not thrown"); + } catch (FlushFailedException ffe) { + assertEquals("Could not write to output buffer", ffe.getMessage()); + } + } } From 491359aacce4da643f8e41e3b8c83aa089e9c867 Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Tue, 13 Nov 2012 09:51:02 -0600 Subject: [PATCH 2/4] BATCH-1799: Intermediate commit with POC code --- .../batch/item/file/FlatFileItemWriter.java | 11 +++- .../batch/item/xml/StaxEventItemWriter.java | 3 +- .../TransactionAwareBufferedWriter.java | 46 ++++++++++---- .../TransactionAwareBufferedWriterTests.java | 63 +++++++++++++++++-- spring-batch-parent/pom.xml | 4 +- 5 files changed, 104 insertions(+), 23 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 c5d67d659..dfdfa5837 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 @@ -581,15 +581,24 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement Writer writer = new BufferedWriter(Channels.newWriter(fileChannel, encoding)) { @Override public void flush() throws IOException { + System.out.println("***************************** Flush Thread:" + Thread.currentThread().getId() + "|" + Thread.currentThread().getName()); super.flush(); + System.out.println("++++++++++++++++++++ super.flush called"); if (forceSync) { channel.force(false); } + System.out.println("~~~~~~~~~~~~~~~~~~~~~~ force complete"); + try { + Thread.sleep(10*1000); + } catch (InterruptedException e) { + } } }; if (transactional) { - return new TransactionAwareBufferedWriter(writer, new Runnable() { + return new TransactionAwareBufferedWriter(channel, new Runnable() { +// return new TransactionAwareBufferedWriter(writer, new Runnable() { public void run() { + System.out.println("============================ closing stream"); closeStream(); } }); 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 bc96c648e..7a0665ed6 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 @@ -418,7 +418,8 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen } }; if (transactional) { - bufferedWriter = new TransactionAwareBufferedWriter(writer, new Runnable() { + bufferedWriter = new TransactionAwareBufferedWriter(channel, new Runnable() { +// bufferedWriter = new TransactionAwareBufferedWriter(writer, new Runnable() { public void run() { closeStream(); } 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 1efc62214..4ec058b89 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 @@ -17,6 +17,9 @@ package org.springframework.batch.support.transaction; import java.io.IOException; import java.io.Writer; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.util.Arrays; import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.transaction.support.TransactionSynchronizationManager; @@ -41,7 +44,8 @@ public class TransactionAwareBufferedWriter extends Writer { private final String closeKey; - private Writer writer; +// private Writer writer; + private FileChannel channel; private final Runnable closeCallback; @@ -53,9 +57,10 @@ public class TransactionAwareBufferedWriter extends Writer { * @param writer actually writes to output * @param closeCallback callback to execute on close */ - public TransactionAwareBufferedWriter(Writer writer, Runnable closeCallback) { + public TransactionAwareBufferedWriter(FileChannel channel, Runnable closeCallback) { super(); - this.writer = writer; +// this.writer = writer; + this.channel = channel; this.closeCallback = closeCallback; this.bufferKey = BUFFER_KEY_PREFIX + "." + hashCode(); this.closeKey = CLOSE_KEY_PREFIX + "." + hashCode(); @@ -77,9 +82,12 @@ public class TransactionAwareBufferedWriter extends Writer { } @Override - public void afterCommit() { + public void beforeCommit(boolean readOnly) { try { - complete(); + System.out.println("***************************** TransactionSynchronization Thread:" + Thread.currentThread().getId() + "|" + Thread.currentThread().getName()); + if(!readOnly) { + complete(); + } } catch (IOException e) { throw new FlushFailedException("Could not write to output buffer", e); @@ -89,10 +97,18 @@ public class TransactionAwareBufferedWriter extends Writer { private void complete() throws IOException { StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(bufferKey); if (buffer != null) { - writer.write(buffer.toString()); - writer.flush(); + String string = buffer.toString(); + int bufferLength = string.length(); + ByteBuffer bb = ByteBuffer.wrap(string.getBytes()); + int bytesWritten = channel.write(bb); + System.out.println("$$$$$$$$$$$$$$$$$$$$$$$$$ length = " + bufferLength + " written = " + bytesWritten); + if(bytesWritten != bufferLength) { + throw new RuntimeException("Unable to write all of the crap you wanted!!!!"); + } +// writer.write(buffer.toString()); +// writer.flush(); if (TransactionSynchronizationManager.hasResource(closeKey)) { - writer.close(); +// writer.close(); closeCallback.run(); } } @@ -148,7 +164,7 @@ public class TransactionAwareBufferedWriter extends Writer { } return; } - writer.close(); +// writer.close(); closeCallback.run(); } @@ -160,7 +176,8 @@ public class TransactionAwareBufferedWriter extends Writer { @Override public void flush() throws IOException { if (!transactionActive()) { - writer.flush(); + channel.force(false); +// writer.flush(); } } @@ -173,13 +190,16 @@ public class TransactionAwareBufferedWriter extends Writer { public void write(char[] cbuf, int off, int len) throws IOException { if (!transactionActive()) { - writer.write(cbuf, off, len); + ByteBuffer bb = ByteBuffer.wrap(new String(Arrays.copyOfRange(cbuf, off, off + len)).getBytes()); + int bytesWritten = channel.write(bb); + if(bytesWritten != len) { + throw new IOException("Unable to write all data. Bytes to write: " + len + ". Bytes written: " + bytesWritten); + } +// writer.write(cbuf, off, len); return; } StringBuffer buffer = getCurrentBuffer(); buffer.append(cbuf, off, len); - } - } 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 9d3711a38..eefb7c510 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,6 +15,11 @@ */ package org.springframework.batch.support.transaction; +import static org.easymock.EasyMock.capture; +import static org.easymock.EasyMock.createMock; +import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.replay; +import static org.easymock.EasyMock.verify; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -23,7 +28,10 @@ import static org.junit.Assert.fail; import java.io.IOException; import java.io.StringWriter; import java.io.Writer; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import org.easymock.Capture; import org.junit.Before; import org.junit.Test; import org.springframework.transaction.PlatformTransactionManager; @@ -39,15 +47,20 @@ import org.springframework.transaction.support.TransactionTemplate; public class TransactionAwareBufferedWriterTests { private Writer stringWriter = new StringWriter(); + + private FileChannel fileChannel; private TransactionAwareBufferedWriter writer; @Before public void init() { - writer = new TransactionAwareBufferedWriter(stringWriter, new Runnable() { + fileChannel = createMock(FileChannel.class); + + writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { public void run() { try { - stringWriter.append("c"); + ByteBuffer bb = ByteBuffer.wrap("c".getBytes()); + fileChannel.write(bb); } catch (IOException e) { throw new IllegalStateException(e); @@ -68,23 +81,61 @@ public class TransactionAwareBufferedWriterTests { */ @Test public void testWriteOutsideTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + fileChannel.force(false); + replay(fileChannel); + writer.write("foo"); writer.flush(); // Not closed yet - assertEquals("foo", stringWriter.toString()); + + String s = getStringFromByteBuffer(bb.getValue()); + + verify(fileChannel); + assertEquals("foo", s); + } + + private String getStringFromByteBuffer(ByteBuffer bb) { + byte[] bytearr = new byte[bb.remaining()]; + bb.get(bytearr); + String s = new String(bytearr); + return s; } @Test public void testBufferSizeOutsideTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + replay(fileChannel); + writer.write("foo"); + + verify(fileChannel); assertEquals(0, writer.getBufferSize()); } @Test public void testCloseOutsideTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + expect(fileChannel.write(capture(bb))).andReturn(1); + replay(fileChannel); + writer.write("foo"); writer.close(); - assertEquals("fooc", stringWriter.toString()); + + verify(fileChannel); + + String output = ""; + + for (ByteBuffer curBuffer : bb.getValues()) { + output = output + getStringFromByteBuffer(curBuffer); + + } +// assertEquals("foo", getStringFromByteBuffer(writeBuffer.getValue())); +// assertEquals("c", getStringFromByteBuffer(commitBuffer.getValue())); + assertEquals("fooc", output); } @Test @@ -105,7 +156,7 @@ public class TransactionAwareBufferedWriterTests { public void write(char[] cbuf, int off, int len) throws IOException { } }; - writer = new TransactionAwareBufferedWriter(mock, new Runnable() { + writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { public void run() { } }); @@ -210,7 +261,7 @@ public class TransactionAwareBufferedWriterTests { public void close() throws IOException { } }; - writer = new TransactionAwareBufferedWriter(badWriter, new Runnable() { + writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { public void run() { try { badWriter.append("c"); diff --git a/spring-batch-parent/pom.xml b/spring-batch-parent/pom.xml index 62ee4e5c3..34c576b14 100644 --- a/spring-batch-parent/pom.xml +++ b/spring-batch-parent/pom.xml @@ -370,13 +370,13 @@ org.easymock easymock - 2.4 + 3.1 test org.easymock easymockclassextension - 2.4 + 3.1 test From 13fbcf992a0e5c032282825269f4794952b69c40 Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Tue, 13 Nov 2012 16:24:58 -0600 Subject: [PATCH 3/4] BATCH-1799: Updated TransactionAwareBufferedWriter to throw exceptions at flush --- .../AbstractSqlPagingQueryProvider.java | 2 +- .../SqlPagingQueryProviderFactoryBean.java | 4 +- .../batch/item/file/FlatFileItemWriter.java | 12 +- .../transform/DelimitedLineTokenizer.java | 2 +- .../batch/item/xml/StaxEventItemWriter.java | 4 +- .../TransactionAwareBufferedWriter.java | 18 +-- .../TransactionAwareBufferedWriterTests.java | 142 +++++++----------- 7 files changed, 64 insertions(+), 120 deletions(-) diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractSqlPagingQueryProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractSqlPagingQueryProvider.java index 444cb8556..fa54047f4 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractSqlPagingQueryProvider.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractSqlPagingQueryProvider.java @@ -68,7 +68,7 @@ public abstract class AbstractSqlPagingQueryProvider implements PagingQueryProvi /** * The setter for the group by clause * - * @param SQL GROUP BY clause part of the SQL query string + * @param groupClause SQL GROUP BY clause part of the SQL query string */ public void setGroupClause(String groupClause) { if (StringUtils.hasText(groupClause)) { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlPagingQueryProviderFactoryBean.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlPagingQueryProviderFactoryBean.java index d4779ef32..a2e383cc5 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlPagingQueryProviderFactoryBean.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlPagingQueryProviderFactoryBean.java @@ -81,7 +81,7 @@ public class SqlPagingQueryProviderFactoryBean implements FactoryBean { } /** - * @param SQL GROUP BY clause part of the SQL query string + * @param groupClause SQL GROUP BY clause part of the SQL query string */ public void setGroupClause(String groupClause) { this.groupClause = groupClause; @@ -123,7 +123,7 @@ public class SqlPagingQueryProviderFactoryBean implements FactoryBean { } /** - * @param sortKey the sortKey to set + * @param sortKeys the sortKey to set */ public void setSortKeys(Map sortKeys) { this.sortKeys = sortKeys; 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 dfdfa5837..c0b47f516 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2012 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. @@ -55,6 +55,7 @@ import org.springframework.util.ClassUtils; * @author Tomas Slanina * @author Robert Kasanicky * @author Dave Syer + * @author Michael Minella */ public class FlatFileItemWriter extends ExecutionContextUserSupport implements ResourceAwareItemWriterItemStream, InitializingBean { @@ -581,24 +582,15 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement Writer writer = new BufferedWriter(Channels.newWriter(fileChannel, encoding)) { @Override public void flush() throws IOException { - System.out.println("***************************** Flush Thread:" + Thread.currentThread().getId() + "|" + Thread.currentThread().getName()); super.flush(); - System.out.println("++++++++++++++++++++ super.flush called"); if (forceSync) { channel.force(false); } - System.out.println("~~~~~~~~~~~~~~~~~~~~~~ force complete"); - try { - Thread.sleep(10*1000); - } catch (InterruptedException e) { - } } }; if (transactional) { return new TransactionAwareBufferedWriter(channel, new Runnable() { -// return new TransactionAwareBufferedWriter(writer, new Runnable() { public void run() { - System.out.println("============================ closing stream"); closeStream(); } }); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java index fd53c4eb2..077911bab 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java @@ -64,7 +64,7 @@ public class DelimitedLineTokenizer extends AbstractLineTokenizer { * Create a new instance of the {@link DelimitedLineTokenizer} class for the * common case where the delimiter is a {@link #DELIMITER_COMMA comma}. * - * @see #DelimitedLineTokenizer(char) + * @see #DelimitedLineTokenizer(String) * @see #DELIMITER_COMMA */ public DelimitedLineTokenizer() { 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 7a0665ed6..003ee1c77 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2007 the original author or authors. + * Copyright 2006-2012 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. @@ -66,6 +66,7 @@ import org.springframework.util.StringUtils; * * @author Peter Zozom * @author Robert Kasanicky + * @author Michael Minella * */ public class StaxEventItemWriter extends ExecutionContextUserSupport implements @@ -419,7 +420,6 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen }; if (transactional) { bufferedWriter = new TransactionAwareBufferedWriter(channel, new Runnable() { -// bufferedWriter = new TransactionAwareBufferedWriter(writer, new Runnable() { public void run() { closeStream(); } 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 4ec058b89..24ca2ee4f 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 @@ -25,7 +25,7 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter import org.springframework.transaction.support.TransactionSynchronizationManager; /** - * Wrapper for a {@link Writer} that delays actually writing to or closing the + * Wrapper for a {@link FileChannel} that delays actually writing to or closing the * buffer if a transaction is active. If a transaction is detected on the call * to {@link #write(String)} the parameter is buffered and passed on to the * underlying writer only when the transaction is committed. @@ -44,22 +44,20 @@ public class TransactionAwareBufferedWriter extends Writer { private final String closeKey; -// private Writer writer; private FileChannel channel; private final Runnable closeCallback; /** - * Create a new instance with the underlying writer provided, and a callback + * 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 * output streams or channels. * - * @param writer actually writes to output + * @param channel channel used to do the actuall file IO * @param closeCallback callback to execute on close */ public TransactionAwareBufferedWriter(FileChannel channel, Runnable closeCallback) { super(); -// this.writer = writer; this.channel = channel; this.closeCallback = closeCallback; this.bufferKey = BUFFER_KEY_PREFIX + "." + hashCode(); @@ -84,7 +82,6 @@ public class TransactionAwareBufferedWriter extends Writer { @Override public void beforeCommit(boolean readOnly) { try { - System.out.println("***************************** TransactionSynchronization Thread:" + Thread.currentThread().getId() + "|" + Thread.currentThread().getName()); if(!readOnly) { complete(); } @@ -101,14 +98,10 @@ public class TransactionAwareBufferedWriter extends Writer { int bufferLength = string.length(); ByteBuffer bb = ByteBuffer.wrap(string.getBytes()); int bytesWritten = channel.write(bb); - System.out.println("$$$$$$$$$$$$$$$$$$$$$$$$$ length = " + bufferLength + " written = " + bytesWritten); if(bytesWritten != bufferLength) { - throw new RuntimeException("Unable to write all of the crap you wanted!!!!"); + throw new IOException("All bytes to be written were not successfully written"); } -// writer.write(buffer.toString()); -// writer.flush(); if (TransactionSynchronizationManager.hasResource(closeKey)) { -// writer.close(); closeCallback.run(); } } @@ -164,7 +157,6 @@ public class TransactionAwareBufferedWriter extends Writer { } return; } -// writer.close(); closeCallback.run(); } @@ -177,7 +169,6 @@ public class TransactionAwareBufferedWriter extends Writer { public void flush() throws IOException { if (!transactionActive()) { channel.force(false); -// writer.flush(); } } @@ -195,7 +186,6 @@ public class TransactionAwareBufferedWriter extends Writer { if(bytesWritten != len) { throw new IOException("Unable to write all data. Bytes to write: " + len + ". Bytes written: " + bytesWritten); } -// writer.write(cbuf, off, len); return; } 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 eefb7c510..e9db8e161 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,19 +15,16 @@ */ package org.springframework.batch.support.transaction; +import static org.easymock.EasyMock.anyObject; import static org.easymock.EasyMock.capture; import static org.easymock.EasyMock.createMock; import static org.easymock.EasyMock.expect; import static org.easymock.EasyMock.replay; import static org.easymock.EasyMock.verify; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import java.io.IOException; -import java.io.StringWriter; -import java.io.Writer; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; @@ -46,8 +43,6 @@ import org.springframework.transaction.support.TransactionTemplate; */ public class TransactionAwareBufferedWriterTests { - private Writer stringWriter = new StringWriter(); - private FileChannel fileChannel; private TransactionAwareBufferedWriter writer; @@ -71,8 +66,6 @@ public class TransactionAwareBufferedWriterTests { private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); - private boolean flushed = false; - /** * Test method for * {@link org.springframework.batch.support.transaction.TransactionAwareBufferedWriter#write(java.lang.String)} @@ -96,13 +89,6 @@ public class TransactionAwareBufferedWriterTests { assertEquals("foo", s); } - private String getStringFromByteBuffer(ByteBuffer bb) { - byte[] bytearr = new byte[bb.remaining()]; - bb.get(bytearr); - String s = new String(bytearr); - return s; - } - @Test public void testBufferSizeOutsideTransaction() throws Exception { Capture bb = new Capture(); @@ -117,9 +103,10 @@ public class TransactionAwareBufferedWriterTests { @Test public void testCloseOutsideTransaction() throws Exception { - Capture bb = new Capture(); - expect(fileChannel.write(capture(bb))).andReturn(3); - expect(fileChannel.write(capture(bb))).andReturn(1); + Capture writeBuffer = new Capture(); + Capture commitBuffer = new Capture(); + expect(fileChannel.write(capture(writeBuffer))).andReturn(3); + expect(fileChannel.write(capture(commitBuffer))).andReturn(1); replay(fileChannel); writer.write("foo"); @@ -127,39 +114,16 @@ public class TransactionAwareBufferedWriterTests { verify(fileChannel); - String output = ""; - - for (ByteBuffer curBuffer : bb.getValues()) { - output = output + getStringFromByteBuffer(curBuffer); - - } -// assertEquals("foo", getStringFromByteBuffer(writeBuffer.getValue())); -// assertEquals("c", getStringFromByteBuffer(commitBuffer.getValue())); - assertEquals("fooc", output); + assertEquals("foo", getStringFromByteBuffer(writeBuffer.getValue())); + assertEquals("c", getStringFromByteBuffer(commitBuffer.getValue())); } @Test @SuppressWarnings({"unchecked", "rawtypes"}) public void testFlushInTransaction() throws Exception { - Writer mock = new Writer() { - @Override - public void close() throws IOException { - throw new UnsupportedOperationException(); - } + expect(fileChannel.write((ByteBuffer)anyObject())).andReturn(3); + replay(fileChannel); - @Override - public void flush() throws IOException { - flushed = true; - } - - @Override - public void write(char[] cbuf, int off, int len) throws IOException { - } - }; - writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { - public void run() { - } - }); new TransactionTemplate(transactionManager).execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { try { @@ -169,35 +133,21 @@ public class TransactionAwareBufferedWriterTests { catch (IOException e) { throw new IllegalStateException("Unexpected IOException", e); } - assertFalse(flushed); + assertEquals(3, writer.getBufferSize()); return null; } }); - assertTrue(flushed); + + verify(fileChannel); } @Test @SuppressWarnings({"unchecked", "rawtypes"}) public void testWriteWithCommit() 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("", stringWriter.toString()); - return null; - } - }); - // Not closed in transaction - assertEquals("foo", stringWriter.toString()); - } - - @Test - @SuppressWarnings({"unchecked", "rawtypes"}) - public void tesBufferSizeInTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + replay(fileChannel); + new TransactionTemplate(transactionManager).execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { try { @@ -210,6 +160,33 @@ public class TransactionAwareBufferedWriterTests { return null; } }); + + verify(fileChannel); + assertEquals(0, writer.getBufferSize()); + } + + @Test + @SuppressWarnings({"unchecked", "rawtypes"}) + public void testBufferSizeInTransaction() throws Exception { + Capture bb = new Capture(); + expect(fileChannel.write(capture(bb))).andReturn(3); + replay(fileChannel); + + 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; + } + }); + + verify(fileChannel); + assertEquals(0, writer.getBufferSize()); } @Test @@ -224,17 +201,17 @@ public class TransactionAwareBufferedWriterTests { catch (IOException e) { throw new IllegalStateException("Unexpected IOException", e); } - assertEquals("", stringWriter.toString()); throw new RuntimeException("Planned failure"); } }); + fail("Exception was not thrown"); } catch (RuntimeException e) { // expected String message = e.getMessage(); assertEquals("Wrong message: " + message, "Planned failure", message); } - assertEquals("", stringWriter.toString()); + assertEquals(0, writer.getBufferSize()); } @Test @@ -246,29 +223,8 @@ public class TransactionAwareBufferedWriterTests { @Test @SuppressWarnings({"unchecked", "rawtypes"}) public void testExceptionOnFlush() throws Exception { - final Writer badWriter = new Writer() { - - @Override - public void write(char[] cbuf, int off, int len) throws IOException { - } - - @Override - public void flush() throws IOException { - throw new IOException("This should be bubbled"); - } - - @Override - public void close() throws IOException { - } - }; writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() { public void run() { - try { - badWriter.append("c"); - } - catch (IOException e) { - throw new IllegalStateException(e); - } } }); @@ -281,7 +237,6 @@ public class TransactionAwareBufferedWriterTests { catch (IOException e) { throw new IllegalStateException("Unexpected IOException", e); } - assertEquals("", stringWriter.toString()); return null; } }); @@ -291,4 +246,11 @@ public class TransactionAwareBufferedWriterTests { assertEquals("Could not write to output buffer", ffe.getMessage()); } } + + private String getStringFromByteBuffer(ByteBuffer bb) { + byte[] bytearr = new byte[bb.remaining()]; + bb.get(bytearr); + String s = new String(bytearr); + return s; + } } From 4ec551e0ff9fb91e42d4664d7db6f805b46251e5 Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Fri, 16 Nov 2012 13:11:27 -0600 Subject: [PATCH 4/4] BATCH-1799: Updated to handle encoding correctly --- .../batch/item/file/FlatFileItemWriter.java | 24 +++++++++++-------- .../batch/item/xml/StaxEventItemWriter.java | 23 ++++++++++-------- .../TransactionAwareBufferedWriter.java | 22 +++++++++++++---- 3 files changed, 44 insertions(+), 25 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 c0b47f516..e09ff0f22 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 @@ -579,23 +579,27 @@ public class FlatFileItemWriter extends ExecutionContextUserSupport implement private Writer getBufferedWriter(FileChannel fileChannel, String encoding) { try { 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(channel, new Runnable() { + TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel, new Runnable() { public void run() { closeStream(); } }); + + writer.setEncoding(encoding); + return writer; } else { + Writer writer = new BufferedWriter(Channels.newWriter(fileChannel, encoding)) { + @Override + public void flush() throws IOException { + super.flush(); + if (forceSync) { + channel.force(false); + } + } + }; + return new BufferedWriter(writer); } } 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 003ee1c77..e427c94c0 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 @@ -409,23 +409,26 @@ public class StaxEventItemWriter extends ExecutionContextUserSupport implemen 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(channel, new Runnable() { + TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(channel, new Runnable() { public void run() { closeStream(); } }); + + writer.setEncoding(encoding); + 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; } delegateEventWriter = createXmlEventWriter(outputFactory, bufferedWriter); 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 24ca2ee4f..73fa5fd6d 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 @@ -47,7 +47,12 @@ 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; + /** * 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 @@ -64,6 +69,10 @@ public class TransactionAwareBufferedWriter extends Writer { this.closeKey = CLOSE_KEY_PREFIX + "." + hashCode(); } + public void setEncoding(String encoding) { + this.encoding = encoding; + } + /** * @return */ @@ -95,8 +104,9 @@ public class TransactionAwareBufferedWriter extends Writer { StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(bufferKey); if (buffer != null) { String string = buffer.toString(); - int bufferLength = string.length(); - ByteBuffer bb = ByteBuffer.wrap(string.getBytes()); + byte[] bytes = string.getBytes(encoding); + int bufferLength = bytes.length; + ByteBuffer bb = ByteBuffer.wrap(bytes); int bytesWritten = channel.write(bb); if(bytesWritten != bufferLength) { throw new IOException("All bytes to be written were not successfully written"); @@ -181,9 +191,11 @@ public class TransactionAwareBufferedWriter extends Writer { public void write(char[] cbuf, int off, int len) throws IOException { if (!transactionActive()) { - ByteBuffer bb = ByteBuffer.wrap(new String(Arrays.copyOfRange(cbuf, off, off + len)).getBytes()); + byte[] bytes = new String(Arrays.copyOfRange(cbuf, off, off + len)).getBytes(encoding); + int length = bytes.length; + ByteBuffer bb = ByteBuffer.wrap(bytes); int bytesWritten = channel.write(bb); - if(bytesWritten != len) { + if(bytesWritten != length) { throw new IOException("Unable to write all data. Bytes to write: " + len + ". Bytes written: " + bytesWritten); } return;