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

IN PROGRESS - BATCH-888: skip listeners should be called when chunk is about to commit

Browse Source 
onSkipInProcess(..) moved after write and before onSkipInWrite(..)
This commit is contained in:
robokaso
2008-10-27 11:12:33 +00:00
parent b5e04ef089
commit 26f6327e4b
2 changed files with 42 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java
View File

@@ -77,6 +77,8 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
private static final String SKIPPED_OUTPUTS_KEY = "SKIPPED_OUTPUTS_BUFFER_KEY";
private static final String SKIPPED_INPUTS_KEY = "SKIPPED_INPUTS_BUFFER_KEY";
public FaultTolerantChunkOrientedTasklet(ItemReader<? extends T> itemReader,
ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
RepeatOperations chunkOperations, RetryOperations retryTemplate,
@@ -104,7 +106,7 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
// TODO: check flags to see if these need to be saved or not (e.g. JMS
// not)
final Chunk<T> inputs = getBuffer(attributes, INPUT_BUFFER_KEY);
final List<T> inputs = getBuffer(attributes, INPUT_BUFFER_KEY);
final List<S> outputs = new ArrayList<S>();
ExitStatus result = ExitStatus.CONTINUABLE;
@@ -134,15 +136,35 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
Map<T, Exception> skippedInputs = getSkippedBuffer(attributes, SKIPPED_INPUTS_KEY);
if (!inputs.isEmpty()) {
process(contribution, inputs, outputs);
inputs.removeAll(skippedInputs.keySet());
process(contribution, inputs, outputs, skippedInputs);
}
Map<S, Exception> skippedOutputs = getSkippedOutputsBuffer(attributes);
Map<S, Exception> skippedOutputs = getSkippedBuffer(attributes, SKIPPED_OUTPUTS_KEY);
// TODO: make sure exceptions get handled by the appropriate handler
outputs.removeAll(skippedOutputs.keySet());
write(outputs, contribution, skippedOutputs);
for (Entry<T, Exception> skip : skippedInputs.entrySet()) {
try {
listener.onSkipInProcess(skip.getKey(), skip.getValue());
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, skip.getValue());
}
}
for (Entry<S, Exception> entry : skippedOutputs.entrySet()) {
try {
listener.onSkipInWrite(entry.getKey(), entry.getValue());
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in skip listener", ex, entry.getValue());
}
}
// On successful completion clear the attributes to signal that there is
// no more processing
if (outputs.isEmpty()) {
@@ -151,6 +173,7 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
inputs.clear();
outputs.clear();
skippedInputs.clear();
skippedOutputs.clear();
}
@@ -202,14 +225,12 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
/**
* Incorporate retry into the item processor stage.
*/
protected void process(final StepContribution contribution, final Chunk<T> inputs, final List<S> outputs)
throws Exception {
protected void process(final StepContribution contribution, final List<T> inputs, final List<S> outputs,
final Map<T, Exception> skippedInputs) throws Exception {
int filtered = 0;
for (final Chunk<T>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) {
final T item = iterator.next();
for (final T item : inputs) {
RetryCallback<S> retryCallback = new RetryCallback<S>() {
@@ -226,13 +247,8 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
Exception e = (Exception) context.getLastThrowable();
if (processSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
contribution.incrementProcessSkipCount();
iterator.remove(e);
try {
listener.onSkipInProcess(item, e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
skippedInputs.put(item, e);
return null;
}
else {
@@ -319,15 +335,6 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(skipped, rollbackClassifier));
for (Entry<S, Exception> entry : skipped.entrySet()) {
try {
listener.onSkipInWrite(entry.getKey(), entry.getValue());
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in skip listener", ex, entry.getValue());
}
}
chunk.clear();
}
@@ -337,23 +344,23 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
* @param inputBufferKey
* @return
*/
private static <W> Chunk<W> getBuffer(AttributeAccessor attributes, String key) {
private static <W> List<W> getBuffer(AttributeAccessor attributes, String key) {
if (!attributes.hasAttribute(key)) {
return new Chunk<W>();
return new ArrayList<W>();
}
@SuppressWarnings("unchecked")
Chunk<W> resource = (Chunk<W>) attributes.getAttribute(key);
List<W> resource = (List<W>) attributes.getAttribute(key);
return resource;
}
private Map<S, Exception> getSkippedOutputsBuffer(AttributeAccessor attributes) {
if (!attributes.hasAttribute(SKIPPED_OUTPUTS_KEY)) {
Map<S, Exception> result = new LinkedHashMap<S, Exception>();
attributes.setAttribute(SKIPPED_OUTPUTS_KEY, result);
private static <E> Map<E, Exception> getSkippedBuffer(AttributeAccessor attributes, String key) {
if (!attributes.hasAttribute(key)) {
Map<E, Exception> result = new LinkedHashMap<E, Exception>();
attributes.setAttribute(key, result);
return result;
}
@SuppressWarnings("unchecked")
Map<S, Exception> resource = (Map<S, Exception>) attributes.getAttribute(SKIPPED_OUTPUTS_KEY);
Map<E, Exception> resource = (Map<E, Exception>) attributes.getAttribute(key);
return resource;
}

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

@@ -249,12 +249,8 @@ public class FaultTolerantChunkOrientedTaskletTests {
}
assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY"));
@SuppressWarnings("unchecked")
Chunk<Integer> chunk = (Chunk<Integer>) attributes.getAttribute("INPUT_BUFFER_KEY");
// skip...
tasklet.execute(contribution, attributes);
assertEquals(1, chunk.getSkips().size());
assertEquals(3, contribution.getReadCount());
assertEquals(1, contribution.getProcessSkipCount());
@@ -277,7 +273,7 @@ public class FaultTolerantChunkOrientedTaskletTests {
StepContribution contribution = new StepExecution("foo", null).createStepContribution();
ChunkContext attributes = new ChunkContext();
// Count to 3: (try + skip + try)
// Count to 2: (try first + fail) + (skip first + try second + fail)
for (int i = 0; i < 2; i++) {
try {
tasklet.execute(contribution, attributes);
@@ -289,12 +285,11 @@ public class FaultTolerantChunkOrientedTaskletTests {
assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY"));
}
@SuppressWarnings("unchecked")
Chunk<Integer> chunk = (Chunk<Integer>) attributes.getAttribute("INPUT_BUFFER_KEY");
assertEquals(1, chunk.getSkips().size());
Map<Integer, Exception > skips = (Map<Integer, Exception>) attributes.getAttribute("SKIPPED_INPUTS_BUFFER_KEY");
assertEquals(1, skips.size());
// The last recovery for this chunk...
tasklet.execute(contribution, attributes);
assertEquals(2, chunk.getSkips().size());
attributes = new ChunkContext();
try {