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

RESOLVED - issue BATCH-1256: Processor is called (and committed) many times for the same items if Writer skips

Browse Source 
http://jira.springframework.org/browse/BATCH-1256
This commit is contained in:
dsyer
2009-06-01 10:36:55 +00:00
parent f2514005ed
commit 685dd789de
5 changed files with 43 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
View File

@@ -19,6 +19,7 @@ package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -39,7 +40,7 @@ import org.springframework.batch.retry.support.DefaultRetryState;
/**
* FaultTolerant implementation of the {@link ChunkProcessor} interface, that
* allows for skipping or retry of items that cause exceptions during writing.
* allows for skipping or retry of items that cause exceptions during writing.
*
*/
public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O> {
@@ -124,6 +125,11 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
protected Chunk<O> transform(final StepContribution contribution, Chunk<I> inputs) throws Exception {
Chunk<O> outputs = new Chunk<O>();
Object userData = inputs.getUserData();
@SuppressWarnings("unchecked")
final Chunk<O> cache = (userData instanceof Chunk) ? (Chunk<O>) userData : null;
final Chunk<O>.ChunkIterator cacheIterator = (cache != null) ? cache.iterator() : null;
final AtomicInteger count = new AtomicInteger(0);
for (final Chunk<I>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) {
@@ -134,7 +140,21 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
public O doWithRetry(RetryContext context) throws Exception {
O output = null;
try {
output = doProcess(item);
count.incrementAndGet();
O cached = (cache != null) ? cacheIterator.next() : null;
if (cached != null && count.get() > 1) {
/*
* If there is a cached chunk then we must be
* scanning for errors in the writer, in which case
* only the first one will be written, and for the
* rest we need to fill in the output from the
* cache.
*/
output = cached;
}
else {
output = doProcess(item);
}
}
catch (Exception e) {
if (rollbackClassifier.classify(e)) {
@@ -259,10 +279,10 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
public Object recover(RetryContext context) throws Exception {
Exception le = (Exception) context.getLastThrowable();
boolean singleton = outputs.size() == 1;
if (singleton && !inputs.isBusy()) {
boolean singleton = outputs.size() == 1 && outputs.getSkips().isEmpty();
if (singleton) {
Chunk<I>.ChunkIterator inputIterator = inputs.iterator();
Chunk<O>.ChunkIterator outputIterator = outputs.iterator();
checkSkipPolicy(inputIterator, outputIterator, le, contribution);

2
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java
View File

@@ -183,7 +183,9 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
contribution.incrementFilterCount(inputsSize - outputs.size() - inputs.getSkips().size());
boolean busy = skips.isBusy();
outputs = new Chunk<O>(outputs.getItems(), skips.getSkips());
outputs.setBusy(busy);
// Remember for next time if there are skips accumulating
inputs.setUserData(outputs);

1
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java
View File

@@ -317,6 +317,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
// [a, b, c, d, e, f, null]
assertEquals(7, provided.size());
// [a, b, b, b, b, b, c, d, d, d, d, d, e, f]
System.err.println(processed);
assertEquals(14, processed.size());
// [b, d]
assertEquals(2, recovered.size());

24
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
View File

@@ -313,11 +313,11 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
// TODO: Fix this with BATCH-1259?
assertEquals("[1, 2, 1, 3, 4, 1, 3, 5]", processor.getProcessed().toString());
assertEquals("[1, 3, 5]", processor.getCommitted().toString());
assertEquals("[1, 3, 5]", writer.getWritten().toString());
assertEquals("[1, 3, 5]", writer.getCommitted().toString());
// TODO: Fix this with BATCH-1259?
assertEquals("[1, 2, 1, 3, 4, 1, 3, 5]", processor.getProcessed().toString());
}
@Test
@@ -347,12 +347,12 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
// TODO: Fix this with BATCH-1256
assertEquals("[1, 2, 3, 4, 5, 1, 2, 3, 4, 5, 2, 3, 4, 5, 3, 4, 5, 4, 5, 5]", processor.getProcessed()
.toString());
assertEquals("[1, 2, 3, 4, 5, 2, 3, 4, 5, 3, 4, 5, 5]", processor.getCommitted().toString());
assertEquals("[1, 2, 3, 4, 1, 2, 3, 4, 5]", writer.getWritten().toString());
assertEquals("[1, 2, 3, 5]", processor.getCommitted().toString());
assertEquals("[1, 2, 3, 5]", writer.getCommitted().toString());
assertEquals("[1, 2, 3, 4, 1, 2, 3, 4, 5]", writer.getWritten().toString());
// TODO: Fix this with BATCH-1259?
assertEquals("[1, 2, 3, 4, 5, 1, 2, 3, 4, 5]", processor.getProcessed()
.toString());
}
@Test
@@ -365,12 +365,12 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
// TODO: Fix this with BATCH-1256
assertEquals("[1, 2, 3, 4, 5, 1, 2, 3, 4, 5, 2, 3, 4, 5, 3, 4, 5, 4, 5, 5]", processor.getProcessed()
.toString());
assertEquals("[1, 2, 3, 4, 5, 3, 4, 5, 5]", processor.getCommitted().toString());
assertEquals("[1, 2, 1, 2, 3, 4, 5]", writer.getWritten().toString());
assertEquals("[1, 3, 5]", writer.getCommitted().toString());
assertEquals("[1, 2, 1, 2, 3, 4, 5]", writer.getWritten().toString());
assertEquals("[1, 3, 5]", processor.getCommitted().toString());
// TODO: Fix this with BATCH-1259?
assertEquals("[1, 2, 3, 4, 5, 1, 2, 3, 4, 5]", processor.getProcessed()
.toString());
}
@SuppressWarnings("unchecked")

7
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
View File

@@ -540,10 +540,9 @@ public class FaultTolerantStepFactoryBeanTests {
assertEquals(1, stepExecution.getSkipCount());
assertEquals(2, stepExecution.getRollbackCount());
// 1,2,3,4,3,4,4 - two re-processing attempts until the item is
// identified and finally skipped on the third attempt
assertEquals(7, processor.getProcessed().size());
assertEquals("[1, 2, 3, 4, 3, 4, 4]", processor.getProcessed().toString());
// 1,2,3,4,3,4 - two re-processing attempts until the item is
// identified and finally skipped on the second attempt
assertEquals("[1, 2, 3, 4, 3, 4]", processor.getProcessed().toString());
assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step
.getName()));
}