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

BATCH-1799: Updated TransactionAwareBufferedWriter to throw exceptions at flush

Browse Source
This commit is contained in:
Michael Minella
2012-11-13 16:24:58 -06:00
parent 491359aacc
commit 13fbcf992a
7 changed files with 64 additions and 120 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;

12
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 {
@@ -581,24 +582,15 @@ public class FlatFileItemWriter<T> 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();
}
});

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

4
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
@@ -419,7 +420,6 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
};
if (transactional) {
bufferedWriter = new TransactionAwareBufferedWriter(channel, new Runnable() {
// bufferedWriter = new TransactionAwareBufferedWriter(writer, new Runnable() {
public void run() {
closeStream();
}

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

@@ -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;
}

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

@@ -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<ByteBuffer> bb = new Capture<ByteBuffer>();
@@ -117,9 +103,10 @@ public class TransactionAwareBufferedWriterTests {
@Test
public void testCloseOutsideTransaction() throws Exception {
Capture<ByteBuffer> bb = new Capture<ByteBuffer>();
expect(fileChannel.write(capture(bb))).andReturn(3);
expect(fileChannel.write(capture(bb))).andReturn(1);
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");
@@ -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<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 {
@@ -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<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
@@ -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;
}
}