OPEN - issue BATCH-788: Remove flush/clear from ItemWriter

Infrastructure tests fixed

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/util/FileUtils.java

@@ -4,65 +4,66 @@ import java.io.File;
 import java.io.IOException;
 
 import org.springframework.batch.item.ItemStreamException;
-import org.springframework.dao.DataAccessResourceFailureException;
 import org.springframework.util.Assert;
 
 /**
  * Utility methods for files used in batch processing.
- *
+ * 
  * @author Peter Zozom
  */
 public class FileUtils {
 
 	// forbids instantiation
-	private FileUtils() {}
+	private FileUtils() {
+	}
 
 	/**
-	 * Set up output file for batch processing. This method implements common logic for
-	 * handling output files when starting or restarting job/step.
-	 * When starting output file processing, method creates/overwrites new file.
-	 * When restarting output file processing, method checks whether file is writable.
-	 *
+	 * Set up output file for batch processing. This method implements common
+	 * logic for handling output files when starting or restarting file I/O.
+	 * When starting output file processing, creates/overwrites new file. When
+	 * restarting output file processing, checks whether file is writable.
+	 * 
 	 * @param file file to be set up
-	 * @param restarted TRUE signalizes that we are restarting output file processing
-	 * @param overwriteOutputFile If set to TRUE, output file will be overwritten
-	 * (this flag is ignored when processing is restart)
-	 *
-	 * @throws IllegalArgumentException when file is NULL
-	 * @throws IllegalStateException when staring output file processing, file exists and
-	 * flag "shouldDeleteExisting" is set to FALSE
-	 * @throws DataAccessResourceFailureException when unable to create file or file is not writable
+	 * @param restarted true signals that we are restarting output file
+	 * processing
+	 * @param overwriteOutputFile If set to true, output file will be
+	 * overwritten (this flag is ignored when processing is restart)
+	 * 
+	 * @throws IllegalArgumentException when file is null
+	 * @throws ItemStreamException when starting output file processing, file
+	 * exists and flag "overwriteOutputFile" is set to false
+	 * @throws ItemStreamException when unable to create file or file is not
+	 * writable
 	 */
-	public static void setUpOutputFile(File file, boolean restarted,
-			boolean overwriteOutputFile) {
+	public static void setUpOutputFile(File file, boolean restarted, boolean overwriteOutputFile) {
 
 		Assert.notNull(file);
 
 		try {
 			if (!restarted) {
 				if (file.exists()) {
-					if(!overwriteOutputFile){
-						throw new  ItemStreamException("File already exists: ["
-						+ file.getAbsolutePath() + "]");
+					if (!overwriteOutputFile) {
+						throw new ItemStreamException("File already exists: [" + file.getAbsolutePath() + "]");
 					}
 					file.delete();
 				}
 
-				if (file.getParent() != null ) {
+				if (file.getParent() != null) {
 					new File(file.getParent()).mkdirs();
 				}
 				file.createNewFile();
-				Assert.state(file.exists(), "Output file must exist");
+				if (!file.exists()) {
+					throw new ItemStreamException("Output file was not created: [" + file.getAbsolutePath()
+								+ "]");
+				}
 			}
-		} catch (IOException ioe) {
-			throw new ItemStreamException(
-					"Unable to create file: [" + file.getAbsolutePath() + "]",
-					ioe);
+		}
+		catch (IOException ioe) {
+			throw new ItemStreamException("Unable to create file: [" + file.getAbsolutePath() + "]", ioe);
 		}
 
 		if (!file.canWrite()) {
-			throw new ItemStreamException(
-					"File is not writable: [" + file.getAbsolutePath() + "]");
+			throw new ItemStreamException("File is not writable: [" + file.getAbsolutePath() + "]");
 		}
 	}
 }

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

@@ -106,9 +106,6 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 	// XML event writer
 	private XMLEventWriter delegateEventWriter;
 
-	// byte offset in file channel at last commit point
-	private long lastCommitPointPosition = 0;
-
 	// current count of processed records
 	private long currentRecordCount = 0;
 
@@ -412,7 +409,6 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 			throw new FlushFailedException("Failed to flush the events", e);
 		}
 
-		lastCommitPointPosition = getPosition();
 	}
 
 	/**
@@ -458,8 +454,6 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
 	private void setPosition(long newPosition) {
 
 		try {
-			Assert.state(channel.size() >= lastCommitPointPosition,
-					"Current file size is smaller than size at last commit");
 			channel.truncate(newPosition);
 			channel.position(newPosition);
 		}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/FlatFileItemWriterTests.java

@@ -66,11 +66,6 @@ public class FlatFileItemWriterTests extends TestCase {
 	 */
 	protected void setUp() throws Exception {
 
-		if (TransactionSynchronizationManager.isSynchronizationActive()) {
-			TransactionSynchronizationManager.clearSynchronization();
-		}
-		TransactionSynchronizationManager.initSynchronization();
-
 		outputFile = File.createTempFile("flatfile-test-output-", ".tmp");
 
 		writer.setResource(new FileSystemResource(outputFile));
@@ -109,10 +104,8 @@ public class FlatFileItemWriterTests extends TestCase {
 
 		writer.open(executionContext);
 		writer.write(Collections.singletonList("test1"));
-		writer.flush();
 		writer.open(executionContext);
 		writer.write(Collections.singletonList("test2"));
-		writer.flush();
 		assertEquals("test1", readLine());
 		assertEquals("test2", readLine());
 	}
@@ -131,7 +124,6 @@ public class FlatFileItemWriterTests extends TestCase {
 	public void testWriteString() throws Exception {
 		writer.open(executionContext);
 		writer.write(Collections.singletonList(TEST_STRING));
-		writer.flush();
 		writer.close(null);
 		String lineFromFile = readLine();
 
@@ -152,7 +144,6 @@ public class FlatFileItemWriterTests extends TestCase {
 		String data = "string";
 		writer.open(executionContext);
 		writer.write(Collections.singletonList(data));
-		writer.flush();
 		String lineFromFile = readLine();
 		// converter not used if input is String
 		assertEquals("FOO:" + data, lineFromFile);
@@ -171,7 +162,6 @@ public class FlatFileItemWriterTests extends TestCase {
 		});
 		writer.open(executionContext);
 		writer.write(Collections.singletonList(TEST_STRING));
-		writer.flush();
 		String lineFromFile = readLine();
 		assertEquals("FOO:" + TEST_STRING, lineFromFile);
 	}
@@ -184,7 +174,6 @@ public class FlatFileItemWriterTests extends TestCase {
 	public void testWriteRecord() throws Exception {
 		writer.open(executionContext);
 		writer.write(Collections.singletonList("1"));
-		writer.flush();
 		String lineFromFile = readLine();
 		assertEquals("1", lineFromFile);
 	}
@@ -193,7 +182,6 @@ public class FlatFileItemWriterTests extends TestCase {
 		writer.setLineSeparator("|");
 		writer.open(executionContext);
 		writer.write(Arrays.asList(new String[] { "1", "2" }));
-		writer.flush();
 		String lineFromFile = readLine();
 		assertEquals("1|2|", lineFromFile);
 	}
@@ -273,11 +261,6 @@ public class FlatFileItemWriterTests extends TestCase {
 		assertEquals(0, executionContext.getLong(ClassUtils.getShortName(FlatFileItemWriter.class) + ".current.count"));
 	}
 
-	/**
-	 * Regular usage of <code>write(String)</code> method
-	 * 
-	 * @throws Exception
-	 */
 	public void testWriteStringWithBogusEncoding() throws Exception {
 		writer.setEncoding("BOGUS");
 		try {
@@ -290,17 +273,11 @@ public class FlatFileItemWriterTests extends TestCase {
 		writer.close(null);
 	}
 
-	/**
-	 * Regular usage of <code>write(String)</code> method
-	 * 
-	 * @throws Exception
-	 */
 	public void testWriteStringWithEncodingAfterClose() throws Exception {
 		testWriteStringWithBogusEncoding();
 		writer.setEncoding("UTF-8");
 		writer.open(executionContext);
 		writer.write(Collections.singletonList(TEST_STRING));
-		writer.flush();
 		String lineFromFile = readLine();
 
 		assertEquals(TEST_STRING, lineFromFile);
@@ -310,7 +287,6 @@ public class FlatFileItemWriterTests extends TestCase {
 		writer.setHeaderLines(new String[] { "a", "b" });
 		writer.open(executionContext);
 		writer.write(Collections.singletonList(TEST_STRING));
-		writer.flush();
 		writer.close(null);
 		String lineFromFile = readLine();
 		assertEquals("a", lineFromFile);
@@ -324,11 +300,9 @@ public class FlatFileItemWriterTests extends TestCase {
 		writer.setHeaderLines(new String[] { "a", "b" });
 		writer.open(executionContext);
 		writer.write(Collections.singletonList(TEST_STRING));
-		writer.clear();
 		writer.close(executionContext);
 		writer.open(executionContext);
 		writer.write(Collections.singletonList(TEST_STRING));
-		writer.flush();
 		writer.close(executionContext);
 		String lineFromFile = readLine();
 		assertEquals("a", lineFromFile);
@@ -344,10 +318,8 @@ public class FlatFileItemWriterTests extends TestCase {
 		writer.setHeaderLines(new String[] { "a", "b" });
 		writer.open(executionContext);
 		writer.write(Collections.singletonList(TEST_STRING));
-		writer.flush();
 		writer.update(executionContext);
 		writer.write(Collections.singletonList(TEST_STRING));
-		writer.clear();
 		writer.close(executionContext);
 		String lineFromFile = readLine();
 		assertEquals("a", lineFromFile);
@@ -357,7 +329,6 @@ public class FlatFileItemWriterTests extends TestCase {
 		assertEquals(TEST_STRING, lineFromFile);
 		writer.open(executionContext);
 		writer.write(Collections.singletonList(TEST_STRING));
-		writer.flush();
 		writer.close(executionContext);
 		reader = null;
 		lineFromFile = readLine();
@@ -370,12 +341,4 @@ public class FlatFileItemWriterTests extends TestCase {
 		assertEquals(TEST_STRING, lineFromFile);
 	}
 
-	private void commit() throws Exception {
-		writer.flush();
-	}
-
-	private void rollback() throws Exception {
-		writer.clear();
-	}
-
 }

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/util/FileUtilsTests.java

@@ -132,7 +132,7 @@ public class FileUtilsTests {
 	public void testCouldntCreateFile(){
 
 		File file = new File("new file"){
-
+			
 			@Override
 			public boolean exists() {
 				return false;
@@ -142,8 +142,9 @@ public class FileUtilsTests {
 		try{
 			FileUtils.setUpOutputFile(file, false, false);
 			fail();
-		}catch(IllegalStateException ex){
-			assertEquals("Output file must exist", ex.getMessage());
+		}catch(ItemStreamException ex){
+			String message = ex.getMessage();
+			assertTrue("Wrong message: "+message, message.startsWith("Output file was not created"));
 		}finally{
 			file.delete();
 		}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/StaxEventItemWriterTests.java

@@ -5,13 +5,11 @@ import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.replay;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import java.io.File;
 import java.io.IOException;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -119,43 +117,6 @@ public class StaxEventItemWriterTests {
 		assertTrue("Wrong content: "+content, content.contains(TEST_STRING));
 	}
 
-	/**
-	 * Item is written to the output file only after flush.
-	 */
-	@Test
-	public void testWriteWithHeaderAfterRollback() throws Exception {
-		Object header = new Object();
-		writer.setHeaderItems(new Object[] {header});
-		writer.open(executionContext);
-		writer.write(items);
-		writer.open(executionContext);
-		writer.write(items);
-		writer.close(executionContext);
-		String content = outputFileContent();
-		assertEquals("Wrong content: "+content, 1, StringUtils.countOccurrencesOf(content, ("<!--" + header + "-->")));
-		assertEquals("Wrong content: "+content, 1, StringUtils.countOccurrencesOf(content, TEST_STRING));
-	}
-
-	/**
-	 * Item is written to the output file only after flush.
-	 */
-	@Test
-	public void testWriteWithHeaderAfterFlushAndRollback() throws Exception {
-		Object header = new Object();
-		writer.setHeaderItems(new Object[] {header});
-		writer.open(executionContext);
-		writer.write(items);
-		writer.update(executionContext);
-		writer.close(executionContext);
-		writer.open(executionContext);
-		writer.write(items);
-		writer.close(executionContext);
-		String content = outputFileContent();
-		assertEquals("Wrong content: "+content, 1, StringUtils.countOccurrencesOf(content, ("<!--" + header + "-->")));
-		// THis test is not transactional, so the body gets written twice, but at least there's only one header
-		assertEquals("Wrong content: "+content, 2, StringUtils.countOccurrencesOf(content, TEST_STRING));
-	}
-
 	/**
 	 * Count of 'records written so far' is returned as statistics.
 	 */

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/xml/TransactionalStaxEventItemWriterTests.java

@@ -0,0 +1,205 @@
+package org.springframework.batch.item.xml;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+
+import javax.xml.stream.XMLEventFactory;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.transform.Result;
+
+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.xml.oxm.MarshallingEventWriterSerializer;
+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;
+import org.springframework.xml.transform.StaxResult;
+
+/**
+ * Tests for {@link StaxEventItemWriter}.
+ */
+public class TransactionalStaxEventItemWriterTests {
+
+	// object under test
+	private StaxEventItemWriter<Object> writer;
+
+	private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
+
+	// output file
+	private Resource resource;
+
+	private ExecutionContext executionContext;
+
+	// test item for writing to output
+	private Object item = new Object() {
+		public String toString() {
+			return ClassUtils.getShortName(StaxEventItemWriter.class) + "-testString";
+		}
+	};
+
+	private List<? extends Object> items = Collections.singletonList(item);
+
+	private static final String TEST_STRING = "<!--" + ClassUtils.getShortName(StaxEventItemWriter.class)
+			+ "-testString-->";
+
+	@Before
+	public void setUp() throws Exception {
+		resource = new FileSystemResource(File.createTempFile("StaxEventWriterOutputSourceTests", ".xml"));
+		writer = createItemWriter();
+		executionContext = new ExecutionContext();
+	}
+
+	/**
+	 * Item is written to the output file only after flush.
+	 */
+	@Test
+	public void testWriteAndFlush() throws Exception {
+		writer.open(executionContext);
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				writer.write(items);
+				return null;
+			}
+		});
+		writer.close(executionContext);
+		String content = outputFileContent();
+		assertTrue("Wrong content: " + content, content.contains(TEST_STRING));
+	}
+
+	/**
+	 * Item is written to the output file only after flush.
+	 */
+	@Test
+	public void testWriteWithHeaderAfterRollback() throws Exception {
+		Object header = new Object();
+		writer.setHeaderItems(new Object[] { header });
+		writer.open(executionContext);
+		try {
+			new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+				public Object doInTransaction(TransactionStatus status) {
+					writer.write(items);
+					throw new RuntimeException("Planned");
+				}
+			});
+			fail("Expected RuntimeException");
+		}
+		catch (RuntimeException e) {
+			// expected
+		}
+		writer.close(executionContext);
+		writer.open(executionContext);
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				writer.write(items);
+				return null;
+			}
+		});
+		writer.close(executionContext);
+		String content = outputFileContent();
+		assertEquals("Wrong content: " + content, 1, StringUtils.countOccurrencesOf(content, ("<!--" + header + "-->")));
+		assertEquals("Wrong content: " + content, 1, StringUtils.countOccurrencesOf(content, TEST_STRING));
+	}
+
+	/**
+	 * Item is written to the output file only after flush.
+	 */
+	@Test
+	public void testWriteWithHeaderAfterFlushAndRollback() throws Exception {
+		Object header = new Object();
+		writer.setHeaderItems(new Object[] { header });
+		writer.open(executionContext);
+		new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+			public Object doInTransaction(TransactionStatus status) {
+				writer.write(items);
+				return null;
+			}
+		});
+		writer.update(executionContext);
+		writer.close(executionContext);
+		writer.open(executionContext);
+		try {
+			new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
+				public Object doInTransaction(TransactionStatus status) {
+					writer.write(items);
+					throw new RuntimeException("Planned");
+				}
+			});
+			fail("Expected RuntimeException");
+		}
+		catch (RuntimeException e) {
+			// expected
+		}
+		writer.close(executionContext);
+		String content = outputFileContent();
+		assertEquals("Wrong content: " + content, 1, StringUtils.countOccurrencesOf(content, ("<!--" + header + "-->")));
+		assertEquals("Wrong content: " + content, 1, StringUtils.countOccurrencesOf(content, TEST_STRING));
+	}
+
+	/**
+	 * @return output file content as String
+	 */
+	private String outputFileContent() throws IOException {
+		return FileUtils.readFileToString(resource.getFile(), null);
+	}
+
+	/**
+	 * Writes object's toString representation as XML comment.
+	 */
+	private static class SimpleMarshaller implements Marshaller {
+		public void marshal(Object graph, Result result) throws XmlMappingException, IOException {
+			Assert.isInstanceOf(StaxResult.class, result);
+
+			StaxResult staxResult = (StaxResult) result;
+			try {
+				staxResult.getXMLEventWriter().add(XMLEventFactory.newInstance().createComment(graph.toString()));
+			}
+			catch (XMLStreamException e) {
+				throw new RuntimeException("Exception while writing to output file", e);
+			}
+		}
+
+		@SuppressWarnings("unchecked")
+		public boolean supports(Class clazz) {
+			return true;
+		}
+	}
+
+	/**
+	 * @return new instance of fully configured writer
+	 */
+	private StaxEventItemWriter<Object> createItemWriter() throws Exception {
+		StaxEventItemWriter<Object> source = new StaxEventItemWriter<Object>();
+		source.setResource(resource);
+
+		Marshaller marshaller = new SimpleMarshaller();
+		MarshallingEventWriterSerializer<Object> serializer = new MarshallingEventWriterSerializer<Object>(marshaller);
+		source.setSerializer(serializer);
+
+		source.setEncoding("UTF-8");
+		source.setRootTagName("root");
+		source.setVersion("1.0");
+		source.setOverwriteOutput(true);
+		source.setSaveState(true);
+
+		source.afterPropertiesSet();
+
+		return source;
+	}
+}