Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #71 from mminella/BATCH-1799

Browse Source 
Batch-1799: Updated TransactionAwareBufferedWriter to throw exceptions on flush
This commit is contained in:
Michael Minella
2012-11-16 11:26:35 -08:00
parent 22894ada90 b9f2ddcb6a
commit 9407ff7ad2
8 changed files with 212 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractSqlPagingQueryProvider.java
View File

@@ -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)) {

4
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlPagingQueryProviderFactoryBean.java
View File

@@ -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<String, Order> sortKeys) {
this.sortKeys = sortKeys;

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

@@ -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<T> extends ExecutionContextUserSupport implements ResourceAwareItemWriterItemStream<T>,
InitializingBean {
@@ -578,23 +579,27 @@ public class FlatFileItemWriter<T> 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(writer, 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);
}
}

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/transform/DelimitedLineTokenizer.java
View File

@@ -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() {

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

@@ -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<T> extends ExecutionContextUserSupport implements
@@ -408,23 +409,26 @@ public class StaxEventItemWriter<T> 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(writer, 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);

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

@@ -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.
@@ -17,17 +17,21 @@ 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;
/**
* 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.
*
* @author Dave Syer
* @author Michael Minella
*
*/
public class TransactionAwareBufferedWriter extends Writer {
@@ -40,26 +44,35 @@ public class TransactionAwareBufferedWriter extends Writer {
private final String closeKey;
private Writer 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 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(Writer writer, Runnable closeCallback) {
public TransactionAwareBufferedWriter(FileChannel channel, Runnable closeCallback) {
super();
this.writer = writer;
this.channel = channel;
this.closeCallback = closeCallback;
this.bufferKey = BUFFER_KEY_PREFIX + "." + hashCode();
this.closeKey = CLOSE_KEY_PREFIX + "." + hashCode();
}
public void setEncoding(String encoding) {
this.encoding = encoding;
}
/**
* @return
*/
@@ -72,26 +85,33 @@ public class TransactionAwareBufferedWriter extends Writer {
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
@Override
public void afterCompletion(int status) {
clear();
}
@Override
public void beforeCommit(boolean readOnly) {
try {
if (status == STATUS_COMMITTED) {
if(!readOnly) {
complete();
}
}
catch (IOException e) {
throw new FlushFailedException("Could not write to output buffer", e);
}
finally {
clear();
}
}
private void complete() throws IOException {
StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(bufferKey);
if (buffer != null) {
writer.write(buffer.toString());
writer.flush();
String string = buffer.toString();
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");
}
if (TransactionSynchronizationManager.hasResource(closeKey)) {
writer.close();
closeCallback.run();
}
}
@@ -147,7 +167,6 @@ public class TransactionAwareBufferedWriter extends Writer {
}
return;
}
writer.close();
closeCallback.run();
}
@@ -159,7 +178,7 @@ public class TransactionAwareBufferedWriter extends Writer {
@Override
public void flush() throws IOException {
if (!transactionActive()) {
writer.flush();
channel.force(false);
}
}
@@ -172,13 +191,17 @@ public class TransactionAwareBufferedWriter extends Writer {
public void write(char[] cbuf, int off, int len) throws IOException {
if (!transactionActive()) {
writer.write(cbuf, off, len);
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 != length) {
throw new IOException("Unable to write all data. Bytes to write: " + len + ". Bytes written: " + bytesWritten);
}
return;
}
StringBuffer buffer = getCurrentBuffer();
buffer.append(cbuf, off, len);
}
}

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

@@ -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.
@@ -15,14 +15,21 @@
*/
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;
import org.easymock.Capture;
import org.junit.Before;
import org.junit.Test;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
@@ -31,27 +38,34 @@ import org.springframework.transaction.support.TransactionTemplate;
/**
* @author Dave Syer
* @author Michael Minella
*
*/
public class TransactionAwareBufferedWriterTests {
private Writer stringWriter = new StringWriter();
private FileChannel fileChannel;
private TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(stringWriter, new Runnable() {
public void run() {
try {
stringWriter.append("c");
private TransactionAwareBufferedWriter writer;
@Before
public void init() {
fileChannel = createMock(FileChannel.class);
writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() {
public void run() {
try {
ByteBuffer bb = ByteBuffer.wrap("c".getBytes());
fileChannel.write(bb);
}
catch (IOException e) {
throw new IllegalStateException(e);
}
}
catch (IOException e) {
throw new IllegalStateException(e);
}
}
});
});
}
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
private boolean flushed = false;
/**
* Test method for
* {@link org.springframework.batch.support.transaction.TransactionAwareBufferedWriter#write(java.lang.String)}
@@ -60,46 +74,56 @@ public class TransactionAwareBufferedWriterTests {
*/
@Test
public void testWriteOutsideTransaction() throws Exception {
Capture<ByteBuffer> bb = new Capture<ByteBuffer>();
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);
}
@Test
public void testBufferSizeOutsideTransaction() throws Exception {
Capture<ByteBuffer> bb = new Capture<ByteBuffer>();
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<ByteBuffer> writeBuffer = new Capture<ByteBuffer>();
Capture<ByteBuffer> commitBuffer = new Capture<ByteBuffer>();
expect(fileChannel.write(capture(writeBuffer))).andReturn(3);
expect(fileChannel.write(capture(commitBuffer))).andReturn(1);
replay(fileChannel);
writer.write("foo");
writer.close();
assertEquals("fooc", stringWriter.toString());
verify(fileChannel);
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(mock, new Runnable() {
public void run() {
}
});
new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
try {
@@ -109,33 +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
public void tesBufferSizeInTransaction() throws Exception {
Capture<ByteBuffer> bb = new Capture<ByteBuffer>();
expect(fileChannel.write(capture(bb))).andReturn(3);
replay(fileChannel);
new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
try {
@@ -148,9 +160,37 @@ public class TransactionAwareBufferedWriterTests {
return null;
}
});
verify(fileChannel);
assertEquals(0, writer.getBufferSize());
}
@Test
@SuppressWarnings({"unchecked", "rawtypes"})
public void testBufferSizeInTransaction() throws Exception {
Capture<ByteBuffer> bb = new Capture<ByteBuffer>();
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
@SuppressWarnings({"unchecked", "rawtypes"})
public void testWriteWithRollback() throws Exception {
try {
new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
@@ -161,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
@@ -179,5 +219,38 @@ public class TransactionAwareBufferedWriterTests {
testWriteWithRollback();
testWriteWithCommit();
}
@Test
@SuppressWarnings({"unchecked", "rawtypes"})
public void testExceptionOnFlush() throws Exception {
writer = new TransactionAwareBufferedWriter(fileChannel, new Runnable() {
public void run() {
}
});
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);
}
return null;
}
});
fail("Exception was not thrown");
} catch (FlushFailedException ffe) {
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;
}
}

4
spring-batch-parent/pom.xml
View File

@@ -370,13 +370,13 @@
<dependency>
<groupId>org.easymock</groupId>
<artifactId>easymock</artifactId>
<version>2.4</version>
<version>3.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.easymock</groupId>
<artifactId>easymockclassextension</artifactId>
<version>2.4</version>
<version>3.1</version>
<scope>test</scope>
</dependency>
<dependency>