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

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

Browse Source 
DRY refactorings - moved common helper methods up to parent class
This commit is contained in:
robokaso
2008-10-30 13:50:45 +00:00
parent 22a014060d
commit 917418dd64
3 changed files with 92 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java
View File

@@ -1,15 +1,21 @@
package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.MulticasterBatchListener;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.core.AttributeAccessor;
/**
* Superclass for {@link Tasklet}s implementing variations on read-process-write
@@ -62,6 +68,7 @@ public abstract class AbstractItemOrientedTasklet<I, O> implements Tasklet {
}
/**
* Surrounds the read call with listener callbacks.
* @return item
* @throws Exception
*/
@@ -97,6 +104,7 @@ public abstract class AbstractItemOrientedTasklet<I, O> implements Tasklet {
}
/**
* Surrounds the actual write call with listener callbacks.
* @param items
* @throws Exception
*/
@@ -112,4 +120,77 @@ public abstract class AbstractItemOrientedTasklet<I, O> implements Tasklet {
}
}
/**
* Call all skip listeners in read-process-write order
* @param skippedReads read exceptions
* @param skippedInputs items and corresponding exceptions skipped in
* processing phase
* @param skippedOutputs items and corresponding exceptions skipped in write
* phase
*/
protected void callSkipListeners(final List<Exception> skippedReads, final Map<I, Exception> skippedInputs,
final Map<O, Exception> skippedOutputs) {
for (Exception e : skippedReads) {
try {
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
}
for (Entry<I, 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<O, Exception> skip : skippedOutputs.entrySet()) {
try {
listener.onSkipInWrite(skip.getKey(), skip.getValue());
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in skip listener", ex, skip.getValue());
}
}
}
/**
* Return a list stored in the attributes under the key. Create an empty
* list and store it if the list is not stored yet.
*/
protected static <T> List<T> getBufferedList(AttributeAccessor attributes, String key) {
List<T> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new ArrayList<T>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
List<T> casted = (List<T>) attributes.getAttribute(key);
buffer = casted;
}
return buffer;
}
/**
* Return a map of items to exceptions stored in the attributes under the
* key, Create an empty map and store it if the list is not stored yet.
*/
protected static <T> Map<T, Exception> getBufferedSkips(AttributeAccessor attributes, String key) {
Map<T, Exception> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new LinkedHashMap<T, Exception>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
Map<T, Exception> casted = (Map<T, Exception>) attributes.getAttribute(key);
buffer = casted;
}
return buffer;
}
}

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

@@ -17,16 +17,13 @@ package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
import org.springframework.batch.core.step.skip.NonSkippableReadException;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
@@ -105,12 +102,12 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
*/
public ExitStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception {
final List<T> inputs = getBuffer(attributes, INPUT_BUFFER_KEY);
final List<T> inputs = getBufferedList(attributes, INPUT_BUFFER_KEY);
final List<S> outputs = new ArrayList<S>();
ExitStatus result = ExitStatus.CONTINUABLE;
final List<Exception> skippedReads = getBuffer(attributes, SKIPPED_READS_KEY);
final List<Exception> skippedReads = getBufferedList(attributes, SKIPPED_READS_KEY);
if (inputs.isEmpty() && outputs.isEmpty()) {
@@ -134,41 +131,17 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
Map<T, Exception> skippedInputs = getSkippedBuffer(attributes, SKIPPED_INPUTS_KEY);
Map<T, Exception> skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY);
if (!inputs.isEmpty()) {
inputs.removeAll(skippedInputs.keySet());
process(contribution, inputs, outputs, skippedInputs);
}
Map<S, Exception> skippedOutputs = getSkippedBuffer(attributes, SKIPPED_OUTPUTS_KEY);
Map<S, Exception> skippedOutputs = getBufferedSkips(attributes, SKIPPED_OUTPUTS_KEY);
outputs.removeAll(skippedOutputs.keySet());
write(outputs, contribution, skippedOutputs);
for (Exception e : skippedReads) {
try {
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
}
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> skip : skippedOutputs.entrySet()) {
try {
listener.onSkipInWrite(skip.getKey(), skip.getValue());
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in skip listener", ex, skip.getValue());
}
}
callSkipListeners(skippedReads, skippedInputs, skippedOutputs);
// On successful completion clear the attributes to signal that there is
// no more processing
@@ -348,26 +321,4 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
private static <W> List<W> getBuffer(AttributeAccessor attributes, String key) {
if (!attributes.hasAttribute(key)) {
List<W> emptyList = new ArrayList<W>();
attributes.setAttribute(key, emptyList);
return emptyList;
}
@SuppressWarnings("unchecked")
List<W> resource = (List<W>) attributes.getAttribute(key);
return resource;
}
private static <E> Map<E, Exception> getSkippedBuffer(AttributeAccessor attributes, String key) {
if (!attributes.hasAttribute(key)) {
Map<E, Exception> emptyMap = new LinkedHashMap<E, Exception>();
attributes.setAttribute(key, emptyMap);
return emptyMap;
}
@SuppressWarnings("unchecked")
Map<E, Exception> resource = (Map<E, Exception>) attributes.getAttribute(key);
return resource;
}
}

70
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java
View File

@@ -3,15 +3,12 @@ package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
@@ -76,40 +73,7 @@ public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends Abstrac
this.processSkipPolicy = processSkipPolicy;
}
private static <T> List<T> getBufferList(AttributeAccessor attributes, String key) {
List<T> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new ArrayList<T>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
List<T> casted = (List<T>) attributes.getAttribute(key);
buffer = casted;
}
return buffer;
}
/**
*
* @param <T> buffer type
* @param attributes used to store the state of the tasklet
* @param key the key buffer is stored under in the attributes
* @return newly created or existing buffer stored under the given key
*/
private static <T> Map<T, Exception> getBuffer(AttributeAccessor attributes, String key) {
Map<T, Exception> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new LinkedHashMap<T, Exception>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
Map<T, Exception> casted = (Map<T, Exception>) attributes.getAttribute(key);
buffer = casted;
}
return buffer;
}
/**
* Read-process-write a list of items. Uses fault-tolerant read, process and
@@ -119,7 +83,7 @@ public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends Abstrac
ExitStatus result = ExitStatus.CONTINUABLE;
final List<I> inputs = new ArrayList<I>();
final List<Exception> skippedReads = getBufferList(attributes, SKIPPED_READS_KEY);
final List<Exception> skippedReads = getBufferedList(attributes, SKIPPED_READS_KEY);
result = repeatOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
@@ -135,7 +99,7 @@ public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends Abstrac
});
// filter inputs marked for skipping
final Map<I, Exception> skippedInputs = getBuffer(attributes, SKIPPED_INPUTS_KEY);
final Map<I, Exception> skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY);
inputs.removeAll(skippedInputs.keySet());
// If there is no input we don't have to do anything more
@@ -147,35 +111,13 @@ public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends Abstrac
process(contribution, inputs, outputs, skippedInputs);
// filter outputs marked for skipping
final Map<O, Exception> skippedOutputs = getBuffer(attributes, SKIPPED_OUTPUTS_KEY);
final Map<O, Exception> skippedOutputs = getBufferedSkips(attributes, SKIPPED_OUTPUTS_KEY);
outputs.removeAll(skippedOutputs.keySet());
write(contribution, outputs, skippedOutputs);
for (Exception e : skippedReads) {
try {
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
}
for (Entry<I, 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<O, Exception> skip : skippedOutputs.entrySet()) {
try {
listener.onSkipInWrite(skip.getKey(), skip.getValue());
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, skip.getValue());
}
}
callSkipListeners(skippedReads, skippedInputs, skippedOutputs);
return result;
}