From 85f815b094dd30af7ed9bea8faf51387e9abb75b Mon Sep 17 00:00:00 2001 From: robokaso Date: Mon, 15 Dec 2008 10:19:36 +0000 Subject: [PATCH] RESOLVED - BATCH-969: FlatFileItemWriters interference in CompositeItemWriter configurable transactional buffer name in TransactionAwareBufferedWriter sample showing correct usage of two file writers simultaneously --- .../batch/item/file/FlatFileItemWriter.java | 2 +- .../util/ExecutionContextUserSupport.java | 8 +++ .../batch/item/xml/StaxEventItemWriter.java | 2 +- .../TransactionAwareBufferedWriter.java | 39 +++++++---- .../TransactionAwareBufferedWriterTests.java | 4 +- .../jobs/compositeItemWriterSampleJob.xml | 45 ++++++++----- ...positeItemWriterSampleFunctionalTests.java | 65 +++++++++---------- 7 files changed, 99 insertions(+), 66 deletions(-) diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java index 9f23ae328..07951d45c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/FlatFileItemWriter.java @@ -450,7 +450,7 @@ public class FlatFileItemWriter 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) { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/util/ExecutionContextUserSupport.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/util/ExecutionContextUserSupport.java index fbe4d81cc..5672e1535 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/util/ExecutionContextUserSupport.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/util/ExecutionContextUserSupport.java @@ -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. diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java index 8647b1908..4f68990cc 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemWriter.java @@ -300,7 +300,7 @@ public class StaxEventItemWriter 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); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java index f52d661be..8ac34f8a8 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriter.java @@ -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); - + } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java index eda0d880d..188bc5edc 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/transaction/TransactionAwareBufferedWriterTests.java @@ -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 { diff --git a/spring-batch-samples/src/main/resources/jobs/compositeItemWriterSampleJob.xml b/spring-batch-samples/src/main/resources/jobs/compositeItemWriterSampleJob.xml index b13ebd412..b0dd56fc6 100644 --- a/spring-batch-samples/src/main/resources/jobs/compositeItemWriterSampleJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/compositeItemWriterSampleJob.xml @@ -1,29 +1,36 @@ + + + + - + + - + - + @@ -39,13 +46,13 @@ class="org.springframework.batch.sample.domain.trade.internal.TradeWriter"> - + + - + @@ -65,8 +72,7 @@ - + - + id="fileItemWriter1"> + + + + + + + + + diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/CompositeItemWriterSampleFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/CompositeItemWriterSampleFunctionalTests.java index 9bc36a223..c09a831b3 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/CompositeItemWriterSampleFunctionalTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/CompositeItemWriterSampleFunctionalTests.java @@ -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 trades = new ArrayList() {{ + final List trades = new ArrayList() { + { 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 outputLines = IOUtils.readLines( - new FileInputStream("target/test-outputs/20070122.testStream.ParallelCustomerReportStep.TEMP.txt")); - + private void checkOutputFile(String fileName) throws IOException { + @SuppressWarnings("unchecked") + List outputLines = IOUtils.readLines(new FileInputStream(fileName)); + String output = ""; for (String line : outputLines) { output += line; } - + assertEquals(EXPECTED_OUTPUT_FILE, output); } - + }