RESOLVED - issue BATCH-786: Transactional flushing for file writers

Added TransactionAwareBuffererdWriter

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java

@@ -16,10 +16,10 @@
 
 package org.springframework.batch.item.file;
 
-import java.io.BufferedWriter;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.io.Writer;
 import java.nio.channels.Channels;
 import java.nio.channels.FileChannel;
 import java.nio.charset.UnsupportedCharsetException;
@@ -40,6 +40,7 @@ import org.springframework.batch.item.file.mapping.FieldSet;
 import org.springframework.batch.item.file.transform.LineAggregator;
 import org.springframework.batch.item.util.ExecutionContextUserSupport;
 import org.springframework.batch.item.util.FileUtils;
+import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.core.io.Resource;
 import org.springframework.util.Assert;
@@ -64,7 +65,8 @@ import org.springframework.util.ClassUtils;
  * @author Robert Kasanicky
  * @author Dave Syer
  */
-public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implements ItemWriter<T>, ItemStream, InitializingBean {
+public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implements ItemWriter<T>, ItemStream,
+		InitializingBean {
 
 	private static final String DEFAULT_LINE_SEPARATOR = System.getProperty("line.separator");
 
@@ -84,8 +86,6 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
 
 	private String encoding = OutputState.DEFAULT_CHARSET;
 
-	private int bufferSize = OutputState.DEFAULT_BUFFER_SIZE;
-
 	private List<String> lineBuffer = new ArrayList<String>();
 
 	private List<String> headerLines = new ArrayList<String>();
@@ -140,13 +140,6 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
 		this.encoding = newEncoding;
 	}
 
-	/**
-	 * Sets buffer size for output template
-	 */
-	public void setBufferSize(int newSize) {
-		this.bufferSize = newSize;
-	}
-
 	/**
 	 * @param shouldDeleteIfExists the shouldDeleteIfExists to set
 	 */
@@ -219,7 +212,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
 	public void open(ExecutionContext executionContext) throws ItemStreamException {
 
 		Assert.notNull(resource, "The resource must be set");
-		
+
 		if (!getOutputState().isInitialized()) {
 			doOpen(executionContext);
 		}
@@ -292,7 +285,6 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
 			}
 			state = new OutputState();
 			state.setDeleteIfExists(shouldDeleteIfExists);
-			state.setBufferSize(bufferSize);
 			state.setEncoding(encoding);
 		}
 		return (OutputState) state;
@@ -306,10 +298,8 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
 		// default encoding for writing to output files - set to UTF-8.
 		private static final String DEFAULT_CHARSET = "UTF-8";
 
-		private static final int DEFAULT_BUFFER_SIZE = 2048;
-
 		// The bufferedWriter over the file channel that is actually written
-		BufferedWriter outputBufferedWriter;
+		Writer outputBufferedWriter;
 
 		FileChannel fileChannel;
 
@@ -317,9 +307,6 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
 		// output file
 		String encoding = DEFAULT_CHARSET;
 
-		// Optional write buffer size
-		int bufferSize = DEFAULT_BUFFER_SIZE;
-
 		boolean restarted = false;
 
 		long lastMarkedByteOffsetPosition = 0;
@@ -363,13 +350,6 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
 			this.shouldDeleteIfExists = shouldDeleteIfExists;
 		}
 
-		/**
-		 * @param bufferSize
-		 */
-		public void setBufferSize(int bufferSize) {
-			this.bufferSize = bufferSize;
-		}
-
 		/**
 		 * @param encoding
 		 */
@@ -444,7 +424,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
 
 			fileChannel = (new FileOutputStream(file.getAbsolutePath(), true)).getChannel();
 
-			outputBufferedWriter = getBufferedWriter(fileChannel, encoding, bufferSize);
+			outputBufferedWriter = getBufferedWriter(fileChannel, encoding);
 
 			// in case of restarting reset position to last committed point
 			if (restarted) {
@@ -463,19 +443,10 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
 		 * Returns the buffered writer opened to the beginning of the file
 		 * specified by the absolute path name contained in absoluteFileName.
 		 */
-		private BufferedWriter getBufferedWriter(FileChannel fileChannel, String encoding, int bufferSize) {
+		private Writer getBufferedWriter(FileChannel fileChannel, String encoding) {
 			try {
-
-				BufferedWriter outputBufferedWriter = null;
-
-				// If a buffer was requested, allocate.
-				if (bufferSize > 0) {
-					outputBufferedWriter = new BufferedWriter(Channels.newWriter(fileChannel, encoding), bufferSize);
-				}
-				else {
-					outputBufferedWriter = new BufferedWriter(Channels.newWriter(fileChannel, encoding));
-				}
-
+				TransactionAwareBufferedWriter outputBufferedWriter = new TransactionAwareBufferedWriter(Channels
+						.newWriter(fileChannel, encoding));
 				return outputBufferedWriter;
 			}
 			catch (UnsupportedCharsetException ucse) {

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java

@@ -3,6 +3,8 @@ package org.springframework.batch.item.xml;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.UnsupportedEncodingException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.util.ArrayList;
@@ -26,6 +28,7 @@ import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.util.ExecutionContextUserSupport;
 import org.springframework.batch.item.util.FileUtils;
 import org.springframework.batch.item.xml.stax.NoStartEndDocumentStreamWriter;
+import org.springframework.batch.support.transaction.TransactionAwareBufferedWriter;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.core.io.Resource;
 import org.springframework.dao.DataAccessResourceFailureException;
@@ -53,7 +56,7 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 		InitializingBean {
 
 	private static final Log log = LogFactory.getLog(StaxEventItemWriter.class);
-	
+
 	// default encoding
 	private static final String DEFAULT_ENCODING = "UTF-8";
 
@@ -253,9 +256,9 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 	 * @see org.springframework.batch.item.ItemStream#open(ExecutionContext)
 	 */
 	public void open(ExecutionContext executionContext) {
-		
+
 		Assert.notNull(resource, "The resource must be set");
-		
+
 		long startAtPosition = 0;
 
 		// if restart data is provided, restart from provided offset
@@ -272,7 +275,7 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 				write(iterator.next());
 			}
 		}
-		
+
 	}
 
 	/**
@@ -298,7 +301,8 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 		XMLOutputFactory outputFactory = XMLOutputFactory.newInstance();
 
 		try {
-			delegateEventWriter = outputFactory.createXMLEventWriter(os, encoding);
+			delegateEventWriter = outputFactory.createXMLEventWriter(new TransactionAwareBufferedWriter(
+					new OutputStreamWriter(os, encoding)));
 			eventWriter = new NoStartEndDocumentStreamWriter(delegateEventWriter);
 			if (!restarted) {
 				startDocument(delegateEventWriter);
@@ -307,6 +311,10 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 		catch (XMLStreamException xse) {
 			throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", xse);
 		}
+		catch (UnsupportedEncodingException e) {
+			throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource
+					+ "] with encoding=[" + encoding + "]", e);
+		}
 
 	}
 
@@ -369,7 +377,7 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 	 * @see org.springframework.batch.item.ItemStream#close(ExecutionContext)
 	 */
 	public void close(ExecutionContext executionContext) {
-		
+
 		// harmless event to close the root tag if there were no items
 		XMLEventFactory factory = XMLEventFactory.newInstance();
 		try {
@@ -378,7 +386,7 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 		catch (XMLStreamException e) {
 			log.error(e);
 		}
-		
+
 		flush();
 		try {
 			endDocument(delegateEventWriter);

spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2006-2007 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.support.transaction;
+
+import java.io.IOException;
+import java.io.Writer;
+
+import org.springframework.batch.item.FlushFailedException;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+/**
+ * Wrapper for a {@link Writer} that delays actually writing to 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.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class TransactionAwareBufferedWriter extends Writer {
+
+	private static final String BUFFER_KEY = TransactionAwareBufferedWriter.class.getName() + ".BUFFER_KEY";
+
+	private Writer writer;
+
+	/**
+	 * @param writer
+	 */
+	public TransactionAwareBufferedWriter(Writer writer) {
+		super();
+		this.writer = writer;
+	}
+
+	/**
+	 * @return
+	 */
+	private StringBuffer getCurrentBuffer() {
+
+		if (!TransactionSynchronizationManager.hasResource(BUFFER_KEY)) {
+
+			TransactionSynchronizationManager.bindResource(BUFFER_KEY, new StringBuffer());
+
+			TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
+				@Override
+				public void afterCompletion(int status) {
+					StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(BUFFER_KEY);
+					if (status == STATUS_COMMITTED) {
+						try {
+							writer.write(buffer.toString());
+							writer.flush();
+						}
+						catch (IOException e) {
+							throw new FlushFailedException("Could not write to output buffer", e);
+						}
+					}
+					TransactionSynchronizationManager.unbindResource(BUFFER_KEY);
+				}
+			});
+
+		}
+		
+		return (StringBuffer) TransactionSynchronizationManager.getResource(BUFFER_KEY);
+		
+	}
+
+	/**
+	 * @return
+	 */
+	private boolean transactionActive() {
+		return TransactionSynchronizationManager.isActualTransactionActive();
+	}
+
+	/* (non-Javadoc)
+	 * @see java.io.Writer#close()
+	 */
+	@Override
+	public void close() throws IOException {
+		writer.close();
+	}
+
+	/* (non-Javadoc)
+	 * @see java.io.Writer#flush()
+	 */
+	@Override
+	public void flush() throws IOException {
+		if (!transactionActive()) {
+			writer.flush();
+		}
+	}
+
+	/* (non-Javadoc)
+	 * @see java.io.Writer#write(char[], int, int)
+	 */
+	@Override
+	public void write(char[] cbuf, int off, int len) throws IOException {
+
+		if (!transactionActive()) {
+			writer.write(cbuf, off, len);
+			return;
+		}
+
+		StringBuffer buffer = getCurrentBuffer();
+		buffer.append(cbuf, off, len);
+		
+	}
+
+}

spring-batch-infrastructure/src/main/resources/META-INF/MANIFEST.MF

@@ -62,57 +62,55 @@ Export-Package: org.springframework.batch.item;version="2.0.0.CI-SNAPS
  io",org.springframework.batch.item.file.mapping;version="2.0.0.CI-SNA
  PSHOT";uses:="org.springframework.beans.factory,org.springframework.v
  alidation",org.springframework.batch.item.file.separator;version="2.0
- .0.CI-SNAPSHOT";uses:="org.springframework.batch.item,org.springframe
- work.core.io",org.springframework.batch.item.file.transform;version="
- 2.0.0.CI-SNAPSHOT";uses:="org.springframework.batch.item.file.mapping
- ,org.springframework.batch.item.transform",org.springframework.batch.
- item.jms;version="2.0.0.CI-SNAPSHOT";uses:="javax.jms,org.springframe
- work.jms.core",org.springframework.batch.item.support;version="2.0.0.
- CI-SNAPSHOT";uses:="org.springframework.batch.item,org.springframewor
- k.batch.item.file.mapping",org.springframework.batch.item.transform;v
- ersion="2.0.0.CI-SNAPSHOT",org.springframework.batch.item.util;versio
- n="2.0.0.CI-SNAPSHOT",org.springframework.batch.item.validator;versio
- n="2.0.0.CI-SNAPSHOT";uses:="org.springframework.validation",org.spri
- ngframework.batch.item.xml;version="2.0.0.CI-SNAPSHOT";uses:="javax.x
- ml.stream,org.springframework.batch.item,org.springframework.core.io"
- ,org.springframework.batch.item.xml.oxm;version="2.0.0.CI-SNAPSHOT";u
- ses:="javax.xml.stream,org.springframework.oxm",org.springframework.b
- atch.item.xml.stax;version="2.0.0.CI-SNAPSHOT";uses:="javax.xml.names
- pace,javax.xml.stream,javax.xml.stream.events",org.springframework.ba
- tch.repeat;version="2.0.0.CI-SNAPSHOT",org.springframework.batch.repe
- at.callback;version="2.0.0.CI-SNAPSHOT";uses:="org.springframework.ba
- tch.item,org.springframework.batch.repeat",org.springframework.batch.
- repeat.context;version="2.0.0.CI-SNAPSHOT";uses:="org.springframework
- .batch.repeat",org.springframework.batch.repeat.exception;version="2.
- 0.0.CI-SNAPSHOT";uses:="org.springframework.batch.repeat,org.springfr
- amework.batch.support",org.springframework.batch.repeat.interceptor;v
- ersion="2.0.0.CI-SNAPSHOT";uses:="org.aopalliance.intercept,org.sprin
- gframework.batch.repeat",org.springframework.batch.repeat.listener;ve
- rsion="2.0.0.CI-SNAPSHOT";uses:="org.springframework.batch.repeat",or
- g.springframework.batch.repeat.policy;version="2.0.0.CI-SNAPSHOT";use
- s:="org.springframework.batch.repeat,org.springframework.batch.repeat
- .context",org.springframework.batch.repeat.support;version="2.0.0.CI-
- SNAPSHOT";uses:="org.springframework.batch.repeat,org.springframework
- .batch.repeat.exception,org.springframework.core.task",org.springfram
- ework.batch.retry;version="2.0.0.CI-SNAPSHOT",org.springframework.bat
- ch.retry.backoff;version="2.0.0.CI-SNAPSHOT";uses:="org.springframewo
- rk.batch.retry",org.springframework.batch.retry.callback;version="2.0
+ .0.CI-SNAPSHOT",org.springframework.batch.item.file.transform;version
+ ="2.0.0.CI-SNAPSHOT";uses:="org.springframework.batch.item.file.mappi
+ ng",org.springframework.batch.item.jms;version="2.0.0.CI-SNAPSHOT";us
+ es:="javax.jms,org.springframework.jms.core",org.springframework.batc
+ h.item.support;version="2.0.0.CI-SNAPSHOT";uses:="org.springframework
+ .batch.item,org.springframework.batch.item.file.mapping",org.springfr
+ amework.batch.item.transform;version="2.0.0.CI-SNAPSHOT",org.springfr
+ amework.batch.item.util;version="2.0.0.CI-SNAPSHOT",org.springframewo
+ rk.batch.item.validator;version="2.0.0.CI-SNAPSHOT";uses:="org.spring
+ framework.validation",org.springframework.batch.item.xml;version="2.0
+ .0.CI-SNAPSHOT";uses:="javax.xml.stream,org.springframework.batch.ite
+ m,org.springframework.core.io",org.springframework.batch.item.xml.oxm
+ ;version="2.0.0.CI-SNAPSHOT";uses:="javax.xml.stream,org.springframew
+ ork.oxm",org.springframework.batch.item.xml.stax;version="2.0.0.CI-SN
+ APSHOT";uses:="javax.xml.namespace,javax.xml.stream,javax.xml.stream.
+ events",org.springframework.batch.repeat;version="2.0.0.CI-SNAPSHOT",
+ org.springframework.batch.repeat.callback;version="2.0.0.CI-SNAPSHOT"
+ ;uses:="org.springframework.batch.item,org.springframework.batch.repe
+ at",org.springframework.batch.repeat.context;version="2.0.0.CI-SNAPSH
+ OT";uses:="org.springframework.batch.repeat",org.springframework.batc
+ h.repeat.exception;version="2.0.0.CI-SNAPSHOT";uses:="org.springframe
+ work.batch.repeat,org.springframework.batch.support",org.springframew
+ ork.batch.repeat.interceptor;version="2.0.0.CI-SNAPSHOT";uses:="org.a
+ opalliance.intercept,org.springframework.batch.repeat",org.springfram
+ ework.batch.repeat.listener;version="2.0.0.CI-SNAPSHOT";uses:="org.sp
+ ringframework.batch.repeat",org.springframework.batch.repeat.policy;v
+ ersion="2.0.0.CI-SNAPSHOT";uses:="org.springframework.batch.repeat,or
+ g.springframework.batch.repeat.context",org.springframework.batch.rep
+ eat.support;version="2.0.0.CI-SNAPSHOT";uses:="org.springframework.ba
+ tch.repeat,org.springframework.batch.repeat.exception,org.springframe
+ work.core.task",org.springframework.batch.retry;version="2.0.0.CI-SNA
+ PSHOT",org.springframework.batch.retry.backoff;version="2.0.0.CI-SNAP
+ SHOT";uses:="org.springframework.batch.retry",org.springframework.bat
+ ch.retry.callback;version="2.0.0.CI-SNAPSHOT";uses:="org.springframew
+ ork.batch.retry",org.springframework.batch.retry.context;version="2.0
  .0.CI-SNAPSHOT";uses:="org.springframework.batch.retry",org.springfra
- mework.batch.retry.context;version="2.0.0.CI-SNAPSHOT";uses:="org.spr
- ingframework.batch.retry",org.springframework.batch.retry.interceptor
- ;version="2.0.0.CI-SNAPSHOT";uses:="org.aopalliance.intercept,org.spr
- ingframework.batch.item,org.springframework.batch.retry",org.springfr
- amework.batch.retry.listener;version="2.0.0.CI-SNAPSHOT";uses:="org.s
- pringframework.batch.retry",org.springframework.batch.retry.policy;ve
- rsion="2.0.0.CI-SNAPSHOT";uses:="org.springframework.batch.retry,org.
- springframework.batch.retry.callback,org.springframework.batch.suppor
- t",org.springframework.batch.retry.support;version="2.0.0.CI-SNAPSHOT
- ";uses:="org.springframework.batch.retry,org.springframework.batch.re
- try.backoff",org.springframework.batch.support;version="2.0.0.CI-SNAP
- SHOT";uses:="org.springframework.beans",org.springframework.batch.sup
- port.transaction;version="2.0.0.CI-SNAPSHOT";uses:="org.aopalliance.i
- ntercept,org.springframework.transaction,org.springframework.transact
- ion.support"
+ mework.batch.retry.interceptor;version="2.0.0.CI-SNAPSHOT";uses:="org
+ .aopalliance.intercept,org.springframework.batch.item,org.springframe
+ work.batch.retry",org.springframework.batch.retry.listener;version="2
+ .0.0.CI-SNAPSHOT";uses:="org.springframework.batch.retry",org.springf
+ ramework.batch.retry.policy;version="2.0.0.CI-SNAPSHOT";uses:="org.sp
+ ringframework.batch.retry,org.springframework.batch.retry.callback,or
+ g.springframework.batch.support",org.springframework.batch.retry.supp
+ ort;version="2.0.0.CI-SNAPSHOT";uses:="org.springframework.batch.retr
+ y,org.springframework.batch.retry.backoff",org.springframework.batch.
+ support;version="2.0.0.CI-SNAPSHOT";uses:="org.springframework.beans"
+ ,org.springframework.batch.support.transaction;version="2.0.0.CI-SNAP
+ SHOT";uses:="org.aopalliance.intercept,org.springframework.transactio
+ n,org.springframework.transaction.support"
 Bundle-Version: 2.0.0.CI-SNAPSHOT
 
 

spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2006-2007 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.support.transaction;
+
+import static org.junit.Assert.*;
+
+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;
+import org.springframework.transaction.support.TransactionTemplate;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class TransactionAwareBufferedWriterTests {
+
+	private Writer stringWriter = new StringWriter();
+
+	private TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(stringWriter);
+
+	private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
+
+	private boolean flushed = false;
+
+	/**
+	 * Test method for
+	 * {@link org.springframework.batch.support.transaction.TransactionAwareBufferedWriter#write(java.lang.String)}
+	 * .
+	 * @throws Exception
+	 */
+	@Test
+	public void testWriteOutsideTransaction() throws Exception {
+		writer.write("foo");
+		writer.flush();
+		assertEquals("foo", stringWriter.toString());
+	}
+
+	@Test
+	public void testCloseOutsideTransaction() throws Exception {
+		writer.write("foo");
+		writer.close();
+		assertEquals("foo", stringWriter.toString());
+	}
+
+	@Test
+	public void testFlushInTransaction() throws Exception {
+		Writer mock = new Writer() {
+			@Override
+			public void close() throws IOException {
+				throw new UnsupportedOperationException();
+			}
+
+			@Override
+			public void flush() throws IOException {
+				flushed = true;
+			}
+
+			@Override
+			public void write(char[] cbuf, int off, int len) throws IOException {
+			}
+		};
+		writer = new TransactionAwareBufferedWriter(mock);
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				try {
+					writer.write("foo");
+					writer.flush();
+				}
+				catch (IOException e) {
+					throw new IllegalStateException("Unexpected IOException", e);
+				}
+				assertFalse(flushed);
+				return null;
+			}
+		});
+		assertTrue(flushed);
+	}
+
+	@Test
+	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;
+			}
+		});
+		assertEquals("foo", stringWriter.toString());
+	}
+
+	@Test
+	public void testWriteWithRollback() throws Exception {
+		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());
+					throw new RuntimeException("Planned failure");
+				}
+			});
+		}
+		catch (RuntimeException e) {
+			// expected
+			String message = e.getMessage();
+			assertEquals("Wrong message:  " + message, "Planned failure", message);
+		}
+		assertEquals("", stringWriter.toString());
+	}
+
+}

spring-batch-samples/pom.xml

@@ -1,7 +1,8 @@
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<artifactId>spring-batch-samples</artifactId>
-  <version>2.0.0.CI-SNAPSHOT </version>
+	<version>2.0.0.CI-SNAPSHOT </version>
 	<packaging>jar</packaging>
 	<name>Samples</name>
 	<description>
@@ -375,4 +376,4 @@
 			</plugin>
 		</plugins>
 	</build>
-</project>
+</project>