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

RESOLVED - BATCH-969: FlatFileItemWriters interference in CompositeItemWriter

Browse Source 
configurable transactional buffer name in TransactionAwareBufferedWriter
sample showing correct usage of two file writers simultaneously
This commit is contained in:
robokaso
2008-12-15 10:19:36 +00:00
parent 48517b2b93
commit 85f815b094
7 changed files with 99 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -450,7 +450,7 @@ public class FlatFileItemWriter<T> extends ExecutionContextUserSupport implement
private Writer getBufferedWriter(FileChannel fileChannel, String encoding) {
try {
TransactionAwareBufferedWriter outputBufferedWriter = new TransactionAwareBufferedWriter(Channels
.newWriter(fileChannel, encoding));
.newWriter(fileChannel, encoding), getName());
return outputBufferedWriter;
}
catch (UnsupportedCharsetException ucse) {

8
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/util/ExecutionContextUserSupport.java
View File

@@ -29,6 +29,14 @@ public class ExecutionContextUserSupport {
private String name;
/**
* @return name used to uniquely identify this instance's entries in shared
* context.
*/
protected String getName() {
return this.name;
}
/**
* @param name unique name for the item stream used to create execution
* context keys.

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

@@ -300,7 +300,7 @@ public class StaxEventItemWriter<T> extends ExecutionContextUserSupport implemen
try {
delegateEventWriter = outputFactory.createXMLEventWriter(new TransactionAwareBufferedWriter(
new OutputStreamWriter(os, encoding)));
new OutputStreamWriter(os, encoding), getName()));
eventWriter = new NoStartEndDocumentStreamWriter(delegateEventWriter);
if (!restarted) {
startDocument(delegateEventWriter);

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

@@ -33,16 +33,21 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
*/
public class TransactionAwareBufferedWriter extends Writer {
private static final String BUFFER_KEY = TransactionAwareBufferedWriter.class.getName() + ".BUFFER_KEY";
private static final String BUFFER_KEY_PREFIX = TransactionAwareBufferedWriter.class.getName() + ".BUFFER_KEY";
private String bufferKey;
private Writer writer;
/**
* @param writer
* @param writer actually writes to output
* @param name used to identify the transactional buffer used by this
* instance
*/
public TransactionAwareBufferedWriter(Writer writer) {
public TransactionAwareBufferedWriter(Writer writer, String name) {
super();
this.writer = writer;
this.bufferKey = BUFFER_KEY_PREFIX + "." + name;
}
/**
@@ -50,14 +55,14 @@ public class TransactionAwareBufferedWriter extends Writer {
*/
private StringBuffer getCurrentBuffer() {
if (!TransactionSynchronizationManager.hasResource(BUFFER_KEY)) {
if (!TransactionSynchronizationManager.hasResource(bufferKey)) {
TransactionSynchronizationManager.bindResource(BUFFER_KEY, new StringBuffer());
TransactionSynchronizationManager.bindResource(bufferKey, new StringBuffer());
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
@Override
public void afterCompletion(int status) {
StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(BUFFER_KEY);
StringBuffer buffer = (StringBuffer) TransactionSynchronizationManager.getResource(bufferKey);
if (status == STATUS_COMMITTED) {
try {
writer.write(buffer.toString());
@@ -67,14 +72,14 @@ public class TransactionAwareBufferedWriter extends Writer {
throw new FlushFailedException("Could not write to output buffer", e);
}
}
TransactionSynchronizationManager.unbindResource(BUFFER_KEY);
TransactionSynchronizationManager.unbindResource(bufferKey);
}
});
}
return (StringBuffer) TransactionSynchronizationManager.getResource(BUFFER_KEY);
return (StringBuffer) TransactionSynchronizationManager.getResource(bufferKey);
}
/**
@@ -84,7 +89,9 @@ public class TransactionAwareBufferedWriter extends Writer {
return TransactionSynchronizationManager.isActualTransactionActive();
}
/* (non-Javadoc)
/*
* (non-Javadoc)
*
* @see java.io.Writer#close()
*/
@Override
@@ -92,7 +99,9 @@ public class TransactionAwareBufferedWriter extends Writer {
writer.close();
}
/* (non-Javadoc)
/*
* (non-Javadoc)
*
* @see java.io.Writer#flush()
*/
@Override
@@ -102,7 +111,9 @@ public class TransactionAwareBufferedWriter extends Writer {
}
}
/* (non-Javadoc)
/*
* (non-Javadoc)
*
* @see java.io.Writer#write(char[], int, int)
*/
@Override
@@ -115,7 +126,7 @@ public class TransactionAwareBufferedWriter extends Writer {
StringBuffer buffer = getCurrentBuffer();
buffer.append(cbuf, off, len);
}
}

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

@@ -37,7 +37,7 @@ public class TransactionAwareBufferedWriterTests {
private Writer stringWriter = new StringWriter();
private TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(stringWriter);
private TransactionAwareBufferedWriter writer = new TransactionAwareBufferedWriter(stringWriter, "someName");
private PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
@@ -80,7 +80,7 @@ public class TransactionAwareBufferedWriterTests {
public void write(char[] cbuf, int off, int len) throws IOException {
}
};
writer = new TransactionAwareBufferedWriter(mock);
writer = new TransactionAwareBufferedWriter(mock, "someName");
new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
try {

45
spring-batch-samples/src/main/resources/jobs/compositeItemWriterSampleJob.xml
View File

@@ -1,29 +1,36 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:p="http://www.springframework.org/schema/p" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd">
<description>
<![CDATA[Sample showing usage of CompositeItemWriter.
Note that when two writers of the same class are used simultaneously
they need to be distinguished using the 'name' property
(see the two FlatFileItemWriters used in this example).]]>
</description>
<bean id="compositeItemWriterJob" parent="simpleJob">
<property name="steps">
<bean id="step1" parent="simpleStep">
<property name="streams">
<list>
<ref bean="fileItemReader" />
<ref bean="fileItemWriter" />
<ref bean="fileItemWriter1" />
<ref bean="fileItemWriter2" />
</list>
</property>
<property name="itemReader" ref="fileItemReader" />
<property name="itemProcessor">
<bean class="org.springframework.batch.item.validator.ValidatingItemProcessor">
<bean
class="org.springframework.batch.item.validator.ValidatingItemProcessor">
<constructor-arg ref="fixedValidator" />
</bean>
</property>
</property>
<property name="itemWriter" ref="compositeWriter" />
</bean>
</property>
@@ -39,13 +46,13 @@
class="org.springframework.batch.sample.domain.trade.internal.TradeWriter">
<property name="dao" ref="tradeDao" />
</bean>
<ref bean="fileItemWriter" />
<ref bean="fileItemWriter1" />
<ref bean="fileItemWriter2" />
</list>
</property>
</bean>
<bean id="fileItemReader"
class="org.springframework.batch.item.file.FlatFileItemReader">
<bean id="fileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader">
<property name="resource"
value="classpath:data/fixedLengthImportJob/input/20070122.teststream.ImportTradeDataStep.txt" />
<property name="lineMapper">
@@ -65,8 +72,7 @@
<bean id="fixedValidator"
class="org.springframework.batch.item.validator.SpringValidator">
<property name="validator">
<bean id="tradeValidator"
class="org.springmodules.validation.valang.ValangValidator">
<bean id="tradeValidator" class="org.springmodules.validation.valang.ValangValidator">
<property name="valang">
<value>
<![CDATA[
@@ -90,15 +96,24 @@
</bean>
<bean class="org.springframework.batch.item.file.FlatFileItemWriter"
id="fileItemWriter">
<property name="resource"
value="file:target/test-outputs/20070122.testStream.ParallelCustomerReportStep.TEMP.txt" />
id="fileItemWriter1">
<property name="name" value="fw1" />
<property name="resource" value="file:target/test-outputs/CustomerReport1.txt" />
<property name="lineAggregator">
<bean
class="org.springframework.batch.item.file.transform.PassThroughLineAggregator" />
</property>
</bean>
<bean class="org.springframework.batch.item.file.FlatFileItemWriter"
id="fileItemWriter2">
<property name="name" value="fw2" />
<property name="resource" value="file:target/test-outputs/CustomerReport2.txt" />
<property name="lineAggregator">
<bean
class="org.springframework.batch.item.file.transform.PassThroughLineAggregator" />
</property>
</bean>
<bean id="fieldSetMapper"
class="org.springframework.batch.sample.domain.trade.internal.TradeFieldSetMapper" />

65
spring-batch-samples/src/test/java/org/springframework/batch/sample/CompositeItemWriterSampleFunctionalTests.java
View File

@@ -21,81 +21,80 @@ import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration()
public class CompositeItemWriterSampleFunctionalTests extends AbstractValidatingBatchLauncherTests {
private static final String GET_TRADES = "SELECT isin, quantity, price, customer FROM trade order by isin";
private static final String EXPECTED_OUTPUT_FILE =
"Trade: [isin=UK21341EAH41,quantity=211,price=31.11,customer=customer1]" +
"Trade: [isin=UK21341EAH42,quantity=212,price=32.11,customer=customer2]" +
"Trade: [isin=UK21341EAH43,quantity=213,price=33.11,customer=customer3]" +
"Trade: [isin=UK21341EAH44,quantity=214,price=34.11,customer=customer4]" +
"Trade: [isin=UK21341EAH45,quantity=215,price=35.11,customer=customer5]";
private static final String EXPECTED_OUTPUT_FILE = "Trade: [isin=UK21341EAH41,quantity=211,price=31.11,customer=customer1]"
+ "Trade: [isin=UK21341EAH42,quantity=212,price=32.11,customer=customer2]"
+ "Trade: [isin=UK21341EAH43,quantity=213,price=33.11,customer=customer3]"
+ "Trade: [isin=UK21341EAH44,quantity=214,price=34.11,customer=customer4]"
+ "Trade: [isin=UK21341EAH45,quantity=215,price=35.11,customer=customer5]";
private SimpleJdbcTemplate simpleJdbcTemplate;
private int activeRow = 0;
private int before;
@Autowired
public void setDataSource(DataSource dataSource) {
this.simpleJdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
/* (non-Javadoc)
* @see org.springframework.batch.sample.AbstractLifecycleSpringContextTests#validatePreConditions()
*/
@Override
protected void validatePreConditions() throws Exception {
simpleJdbcTemplate.update("DELETE from TRADE");
before = simpleJdbcTemplate.queryForInt("SELECT COUNT(*) from TRADE");
}
protected void validatePostConditions() throws Exception {
checkOutputFile();
@Override
protected void validatePostConditions() throws Exception {
checkOutputFile("target/test-outputs/CustomerReport1.txt");
checkOutputFile("target/test-outputs/CustomerReport2.txt");
checkOutputTable();
}
private void checkOutputTable() {
final List<Trade> trades = new ArrayList<Trade>() {{
final List<Trade> trades = new ArrayList<Trade>() {
{
add(new Trade("UK21341EAH41", 211, new BigDecimal("31.11"), "customer1"));
add(new Trade("UK21341EAH42", 212, new BigDecimal("32.11"), "customer2"));
add(new Trade("UK21341EAH43", 213, new BigDecimal("33.11"), "customer3"));
add(new Trade("UK21341EAH44", 214, new BigDecimal("34.11"), "customer4"));
add(new Trade("UK21341EAH45", 215, new BigDecimal("35.11"), "customer5"));
}};
}
};
int after = simpleJdbcTemplate.queryForInt("SELECT COUNT(*) from TRADE");
assertEquals(before+5, after);
assertEquals(before + 5, after);
simpleJdbcTemplate.getJdbcOperations().query(GET_TRADES, new RowCallbackHandler() {
public void processRow(ResultSet rs) throws SQLException {
Trade trade = trades.get(activeRow++);
assertEquals(trade.getIsin(), rs.getString(1));
assertEquals(trade.getQuantity(), rs.getLong(2));
assertEquals(trade.getPrice(), rs.getBigDecimal(3));
assertEquals(trade.getCustomer(), rs.getString(4));
}
});
}
@SuppressWarnings("unchecked")
private void checkOutputFile() throws IOException {
List<String> outputLines = IOUtils.readLines(
new FileInputStream("target/test-outputs/20070122.testStream.ParallelCustomerReportStep.TEMP.txt"));
private void checkOutputFile(String fileName) throws IOException {
@SuppressWarnings("unchecked")
List<String> outputLines = IOUtils.readLines(new FileInputStream(fileName));
String output = "";
for (String line : outputLines) {
output += line;
}
assertEquals(EXPECTED_OUTPUT_FILE, output);
}
}