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

RESOLVED - BATCH-853: broken transactional item processing

Browse Source 
added testcase that checks items are re-processed after writer failure
This commit is contained in:
robokaso
2008-10-17 08:59:08 +00:00
parent 66799db095
commit e9ca8a4d82
1 changed files with 89 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

137
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java
View File

@@ -61,7 +61,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
private List<Integer> processed = new ArrayList<Integer>(); private List<Integer> processed = new ArrayList<Integer>();
private FaultTolerantChunkOrientedTasklet<Integer, String> handler; private FaultTolerantChunkOrientedTasklet<Integer, String> tasklet;
private RepeatTemplate chunkOperations = new RepeatTemplate(); private RepeatTemplate chunkOperations = new RepeatTemplate();
@@ -109,16 +109,16 @@ public class FaultTolerantChunkOrientedTaskletTests {
@Test @Test
public void testBasicHandle() throws Exception { public void testBasicHandle() throws Exception {
handler = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader, itemProcessor, itemWriter, chunkOperations, tasklet = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader, itemProcessor, itemWriter,
retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
StepContribution contribution = new StepExecution("foo", null).createStepContribution(); StepContribution contribution = new StepExecution("foo", null).createStepContribution();
handler.execute(contribution, new ChunkContext()); tasklet.execute(contribution, new ChunkContext());
assertEquals(limit, contribution.getReadCount()); assertEquals(limit, contribution.getReadCount());
} }
@Test @Test
public void testSkipOnRead() throws Exception { public void testSkipOnRead() throws Exception {
handler = new FaultTolerantChunkOrientedTasklet<Integer, String>(new ItemReader<Integer>() { tasklet = new FaultTolerantChunkOrientedTasklet<Integer, String>(new ItemReader<Integer>() {
public Integer read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { public Integer read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
throw new RuntimeException("Barf!"); throw new RuntimeException("Barf!");
} }
@@ -128,7 +128,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
StepContribution contribution = new StepExecution("foo", null).createStepContribution(); StepContribution contribution = new StepExecution("foo", null).createStepContribution();
ChunkContext attributes = new ChunkContext(); ChunkContext attributes = new ChunkContext();
try { try {
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
fail("Expected SkipLimitExceededException"); fail("Expected SkipLimitExceededException");
} }
catch (SkipLimitExceededException e) { catch (SkipLimitExceededException e) {
@@ -140,24 +140,25 @@ public class FaultTolerantChunkOrientedTaskletTests {
@Test @Test
public void testSkipSingleItemOnWrite() throws Exception { public void testSkipSingleItemOnWrite() throws Exception {
handler = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader, itemProcessor, new ItemWriter<String>() { tasklet = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader, itemProcessor,
public void write(List<? extends String> items) throws Exception { new ItemWriter<String>() {
written.addAll(items); public void write(List<? extends String> items) throws Exception {
throw new RuntimeException("Barf!"); written.addAll(items);
} throw new RuntimeException("Barf!");
}, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); }
}, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1));
StepContribution contribution = new StepExecution("foo", null).createStepContribution(); StepContribution contribution = new StepExecution("foo", null).createStepContribution();
ChunkContext attributes = new ChunkContext(); ChunkContext attributes = new ChunkContext();
try { try {
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
fail("Expected RuntimeException"); fail("Expected RuntimeException");
} }
catch (Exception e) { catch (Exception e) {
assertEquals("Barf!", e.getMessage()); assertEquals("Barf!", e.getMessage());
} }
assertTrue(attributes.hasAttribute("SKIPPED_OUTPUTS_BUFFER_KEY")); assertTrue(attributes.hasAttribute("SKIPPED_OUTPUTS_BUFFER_KEY"));
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
assertEquals(1, contribution.getReadCount()); assertEquals(1, contribution.getReadCount());
assertEquals(1, contribution.getWriteSkipCount()); assertEquals(1, contribution.getWriteSkipCount());
assertEquals(1, written.size()); assertEquals(1, written.size());
@@ -165,13 +166,14 @@ public class FaultTolerantChunkOrientedTaskletTests {
@Test @Test
public void testSkipMultipleItemsOnWrite() throws Exception { public void testSkipMultipleItemsOnWrite() throws Exception {
handler = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader, itemProcessor, new ItemWriter<String>() { tasklet = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader, itemProcessor,
public void write(List<? extends String> items) throws Exception { new ItemWriter<String>() {
logger.debug("Writing items: " + items); public void write(List<? extends String> items) throws Exception {
written.addAll(items); logger.debug("Writing items: " + items);
throw new RuntimeException("Barf!"); written.addAll(items);
} throw new RuntimeException("Barf!");
}, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); }
}, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2)); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2));
StepContribution contribution = new StepExecution("foo", null).createStepContribution(); StepContribution contribution = new StepExecution("foo", null).createStepContribution();
ChunkContext attributes = new ChunkContext(); ChunkContext attributes = new ChunkContext();
@@ -179,7 +181,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
// Count to 3: (try + skip + skip) // Count to 3: (try + skip + skip)
for (int i = 0; i < 3; i++) { for (int i = 0; i < 3; i++) {
try { try {
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
fail("Expected RuntimeException on i=" + i); fail("Expected RuntimeException on i=" + i);
} }
catch (Exception e) { catch (Exception e) {
@@ -191,18 +193,18 @@ public class FaultTolerantChunkOrientedTaskletTests {
List<String> chunk = (List<String>) attributes.getAttribute("SKIPPED_OUTPUTS_BUFFER_KEY"); List<String> chunk = (List<String>) attributes.getAttribute("SKIPPED_OUTPUTS_BUFFER_KEY");
assertEquals(1, chunk.size()); assertEquals(1, chunk.size());
// The last recovery for this chunk... // The last recovery for this chunk...
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
attributes = new ChunkContext(); attributes = new ChunkContext();
try { try {
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
fail("Expected RuntimeException"); fail("Expected RuntimeException");
} }
catch (Exception e) { catch (Exception e) {
assertEquals("Barf!", e.getMessage()); assertEquals("Barf!", e.getMessage());
} }
try { try {
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
fail("Expected SkipLimitExceededException"); fail("Expected SkipLimitExceededException");
} }
catch (SkipLimitExceededException e) { catch (SkipLimitExceededException e) {
@@ -217,16 +219,17 @@ public class FaultTolerantChunkOrientedTaskletTests {
@Test @Test
public void testSkipSingleItemOnProcess() throws Exception { public void testSkipSingleItemOnProcess() throws Exception {
handler = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader, new ItemProcessor<Integer, String>() { tasklet = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader,
public String process(Integer item) throws Exception { new ItemProcessor<Integer, String>() {
logger.debug("Processing item: " + item); public String process(Integer item) throws Exception {
processed.add(item); logger.debug("Processing item: " + item);
if (item == 3) { processed.add(item);
throw new RuntimeException("Barf!"); if (item == 3) {
} throw new RuntimeException("Barf!");
return "p" + item; }
} return "p" + item;
}, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, }
}, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy,
writeSkipPolicy); writeSkipPolicy);
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(3)); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(3));
StepContribution contribution = new StepExecution("foo", null).createStepContribution(); StepContribution contribution = new StepExecution("foo", null).createStepContribution();
@@ -234,7 +237,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
// try // try
try { try {
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
fail("Expected RuntimeException"); fail("Expected RuntimeException");
} }
catch (Exception e) { catch (Exception e) {
@@ -246,7 +249,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
Chunk<Integer> chunk = (Chunk<Integer>) attributes.getAttribute("INPUT_BUFFER_KEY"); Chunk<Integer> chunk = (Chunk<Integer>) attributes.getAttribute("INPUT_BUFFER_KEY");
// skip... // skip...
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
assertEquals(1, chunk.getSkips().size()); assertEquals(1, chunk.getSkips().size());
assertEquals(3, contribution.getReadCount()); assertEquals(3, contribution.getReadCount());
@@ -257,13 +260,14 @@ public class FaultTolerantChunkOrientedTaskletTests {
@Test @Test
public void testSkipOverLimitOnProcess() throws Exception { public void testSkipOverLimitOnProcess() throws Exception {
handler = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader, new ItemProcessor<Integer, String>() { tasklet = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader,
public String process(Integer item) throws Exception { new ItemProcessor<Integer, String>() {
logger.debug("Processing item: " + item); public String process(Integer item) throws Exception {
processed.add(item); logger.debug("Processing item: " + item);
throw new RuntimeException("Barf!"); processed.add(item);
} throw new RuntimeException("Barf!");
}, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, }
}, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy,
writeSkipPolicy); writeSkipPolicy);
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2)); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2));
StepContribution contribution = new StepExecution("foo", null).createStepContribution(); StepContribution contribution = new StepExecution("foo", null).createStepContribution();
@@ -272,7 +276,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
// Count to 3: (try + skip + try) // Count to 3: (try + skip + try)
for (int i = 0; i < 2; i++) { for (int i = 0; i < 2; i++) {
try { try {
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
fail("Expected RuntimeException on i=" + i); fail("Expected RuntimeException on i=" + i);
} }
catch (Exception e) { catch (Exception e) {
@@ -285,19 +289,19 @@ public class FaultTolerantChunkOrientedTaskletTests {
assertEquals(1, chunk.getSkips().size()); assertEquals(1, chunk.getSkips().size());
// The last recovery for this chunk... // The last recovery for this chunk...
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
assertEquals(2, chunk.getSkips().size()); assertEquals(2, chunk.getSkips().size());
attributes = new ChunkContext(); attributes = new ChunkContext();
try { try {
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
fail("Expected RuntimeException"); fail("Expected RuntimeException");
} }
catch (Exception e) { catch (Exception e) {
assertEquals("Barf!", e.getMessage()); assertEquals("Barf!", e.getMessage());
} }
try { try {
handler.execute(contribution, attributes); tasklet.execute(contribution, attributes);
fail("Expected SkipLimitExceededException"); fail("Expected SkipLimitExceededException");
} }
catch (SkipLimitExceededException e) { catch (SkipLimitExceededException e) {
@@ -309,4 +313,41 @@ public class FaultTolerantChunkOrientedTaskletTests {
// Just before the skip at the end we process once more // Just before the skip at the end we process once more
assertEquals(3, processed.size()); assertEquals(3, processed.size());
} }
/**
* When writer throws an exception that causes rollback, items are
* re-processed in next iteration.
*/
@Test
public void testReprocessAfterWriterRollback() {
final String WRITER_FAILED_MESSAGE = "writer failed";
final int CHUNK_SIZE = 2;
tasklet = new FaultTolerantChunkOrientedTasklet<Integer, String>(itemReader,
new ItemProcessor<Integer, String>() {
public String process(Integer item) throws Exception {
processed.add(item);
return String.valueOf(item);
}
}, new ItemWriter<String>() {
public void write(List<? extends String> items) throws Exception {
throw new RuntimeException(WRITER_FAILED_MESSAGE);
}
}, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(CHUNK_SIZE));
StepContribution contribution = new StepExecution("foo", null).createStepContribution();
ChunkContext attributes = new ChunkContext();
for (int i = 1; i <= 2; i++) {
try {
tasklet.execute(contribution, attributes);
fail();
}
catch (Exception e) {
assertEquals(WRITER_FAILED_MESSAGE, e.getMessage());
assertEquals(i*CHUNK_SIZE, processed.size());
}
}
}
} }