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

OPEN - issue BATCH-753: Listener exception handling

Browse Source 
Add test for SkipListenerFailedException
This commit is contained in:
dsyer
2008-08-06 10:44:13 +00:00
parent 05d7b6f036
commit c230e9e0d3
11 changed files with 214 additions and 773 deletions
Download Patch File Download Diff File Expand all files Collapse all files

323
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java
View File

@@ -1,323 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.UnexpectedJobExecutionException;
import org.springframework.batch.core.listener.CompositeSkipListener;
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.MarkFailedException;
import org.springframework.transaction.support.TransactionSynchronizationManager;
/**
* {@link ItemHandler} that implements skip behavior. It delegates to
* {@link #setItemSkipPolicy(ItemSkipPolicy)} to decide whether skip should be
* called or not.
*
* When exception is skipped on read it is *not* re-thrown (does not cause tx
* rollback). Skipped exception on write is re-thrown by default (causes tx
* rollback) unless the exception class is included in
* {@link #setDoNotRethrowExceptionClasses(Class[])}.
*
* If exception is thrown while reading the item, skip is called on the
* {@link ItemReader}. If exception is thrown while writing the item, skip is
* called on both {@link ItemReader} and {@link ItemWriter}.
*
* @author Dave Syer
* @author Robert Kasanicky
*/
public class ItemSkipPolicyItemHandler<T> extends SimpleItemHandler<T> {
/**
* Key for transaction resource that holds skipped keys until they can be
* removed
*/
private static final String TO_BE_REMOVED = ItemSkipPolicyItemHandler.class.getName() + ".TO_BE_REMOVED";
private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
private int skipCacheCapacity = 1024;
private Map<Object, Throwable> skippedExceptions = new HashMap<Object, Throwable>();
private Class<?>[] doNotRethrowExceptionClasses = new Class[] {};
private static final ItemKeyGenerator defaultItemKeyGenerator = new ItemKeyGenerator() {
public Object getKey(Object item) {
return item;
}
};
private ItemKeyGenerator itemKeyGenerator = defaultItemKeyGenerator;
private CompositeSkipListener listener = new CompositeSkipListener();
/**
* Register some {@link SkipListener}s with the handler. Each will get the
* callbacks in the order specified at the correct stage if a skip occurs.
*
* @param listeners
*/
public void setSkipListeners(SkipListener[] listeners) {
for (int i = 0; i < listeners.length; i++) {
registerSkipListener(listeners[i]);
}
}
/**
* Register a listener for callbacks at the appropriate stages in a skip
* process.
*
* @param listener a {@link SkipListener}
*/
public void registerSkipListener(SkipListener listener) {
this.listener.register(listener);
}
/**
* Public setter for the {@link ItemKeyGenerator}. Defaults to just return
* the item, and since it will be used before a write operation.
* Implementations must ensure that items always have the same key when they
* are read from the {@link ItemReader} (so if the item is mutable and the
* reader does any buffering the key generator might need to take care to
* only use data that do not change on write).
*
* @param itemKeyGenerator the {@link ItemKeyGenerator} to set. If null
* resets to default value.
*/
public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) {
this.itemKeyGenerator = (itemKeyGenerator == null) ? defaultItemKeyGenerator : itemKeyGenerator;
}
/**
* @param itemReader
* @param itemWriter
*/
public ItemSkipPolicyItemHandler(ItemReader<T> itemReader, ItemWriter<T> itemWriter) {
super(itemReader, itemWriter);
}
/**
* @param itemSkipPolicy
*/
public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) {
this.itemSkipPolicy = itemSkipPolicy;
}
/**
* Public setter for the capacity of the skipped item cache. If a large
* number of items are failing and not being recognized as skipped, it
* usually signals a problem with the key generation (often equals and
* hashCode in the item itself). So it is better to enforce a strict limit
* than have weird looking errors, where a skip limit is reached without
* anything being skipped.
*
* @param skipCacheCapacity the capacity to set
*/
public void setSkipCacheCapacity(int skipCacheCapacity) {
this.skipCacheCapacity = skipCacheCapacity;
}
/**
* Tries to read the item from the reader, in case of exception skip the
* item if the skip policy allows, otherwise re-throw.
*
* @param contribution current StepContribution holding skipped items count
* @return next item for processing
*/
protected T read(StepContribution contribution) throws Exception {
while (true) {
try {
T item = doRead();
Object key = itemKeyGenerator.getKey(item);
Throwable throwable = getSkippedException(key);
while (item != null && throwable != null) {
logger.debug("Skipping item on input, previously failed on output; key=[" + key + "]");
scheduleForRemoval(key);
item = doRead();
key = itemKeyGenerator.getKey(item);
throwable = getSkippedException(key);
}
return item;
}
catch (Exception e) {
try {
if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
// increment skip count and try again
contribution.incrementTemporaryReadSkipCount();
listener.onSkipInRead(e);
logger.debug("Skipping failed input", e);
}
else {
// re-throw only when the skip policy runs out of
// patience
throw e;
}
}
catch (SkipLimitExceededException ex) {
// we are headed for a abnormal ending so bake in the skip
// count
contribution.combineSkipCounts();
throw ex;
}
}
}
}
/**
* Tries to write the item using the writer. In case of exception consults
* skip policy before re-throwing the exception. The exception is always
* re-thrown, but if the item is seen again on read it will be skipped.
*
* @param item item to write
* @param contribution current StepContribution holding skipped items count
*/
protected void write(T item, StepContribution contribution) throws Exception {
doWriteWithSkip(item, contribution);
}
/**
* @param item
* @param contribution
* @throws Exception
*/
protected final void doWriteWithSkip(T item, StepContribution contribution) throws Exception {
// Get the key as early as possible, otherwise it might change in
// doWrite()
Object key = itemKeyGenerator.getKey(item);
try {
doWrite(item);
}
catch (Exception e) {
if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
contribution.incrementWriteSkipCount();
addSkippedException(key, e);
logger.debug("Added item to skip list; key=" + key);
listener.onSkipInWrite(item, e);
// return without re-throwing if exception shouldn't cause
// rollback
if (!shouldRethrow(e)) {
return;
}
}
// re-throw exception on write by default
throw e;
}
}
private boolean shouldRethrow(Exception e) {
for (int i = 0; i < doNotRethrowExceptionClasses.length; i++) {
if (doNotRethrowExceptionClasses[i].isAssignableFrom(e.getClass())) {
return false;
}
}
return true;
}
public void mark() throws MarkFailedException {
super.mark();
clearSkippedExceptions();
}
/**
* Synchronized setter for a skipped exception.
*/
private void addSkippedException(Object key, Throwable e) {
synchronized (skippedExceptions) {
if (skippedExceptions.size() >= skipCacheCapacity) {
throw new UnexpectedJobExecutionException(
"The cache of failed items to skipped unexpectedly reached its capacity ("
+ skipCacheCapacity
+ "). "
+ "This often indicates a problem with the key generation strategy, and/or a mistake in the implementation of hashCode and equals in the items being processed.");
}
skippedExceptions.put(key, e);
}
}
/**
* Schedule this key for removal from the skipped exception cache at the end
* of this transaction (only removed if business transaction is successful).
*
* @param key
*/
@SuppressWarnings("unchecked")
private void scheduleForRemoval(Object key) {
if (!TransactionSynchronizationManager.hasResource(TO_BE_REMOVED)) {
TransactionSynchronizationManager.bindResource(TO_BE_REMOVED, new HashSet<Object>());
}
((Set<Object>) TransactionSynchronizationManager.getResource(TO_BE_REMOVED)).add(key);
}
/**
* Clear the map of skipped exception corresponding to key.
* @param key the key to clear
*/
@SuppressWarnings("unchecked")
private void clearSkippedExceptions() {
if (!TransactionSynchronizationManager.hasResource(TO_BE_REMOVED)) {
return;
}
synchronized (skippedExceptions) {
for (Object key : ((Set<Object>) TransactionSynchronizationManager.getResource(TO_BE_REMOVED))) {
skippedExceptions.remove(key);
}
TransactionSynchronizationManager.unbindResource(TO_BE_REMOVED);
}
}
/**
* Synchronized getter for a skipped exception.
* @return the skippedExceptions
*/
private Throwable getSkippedException(Object key) {
synchronized (skippedExceptions) {
return (Throwable) skippedExceptions.get(key);
}
}
/**
* doNotRethrowExceptionClasses will not be re-thrown when skipped.
* @param doNotRethrowExceptionClasses empty by default
*/
public void setDoNotRethrowExceptionClasses(Class<?>[] doNotRethrowExceptionClasses) {
this.doNotRethrowExceptionClasses = doNotRethrowExceptionClasses;
}
}

5
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
View File

@@ -11,6 +11,7 @@ import org.springframework.batch.core.listener.CompositeSkipListener;
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
@@ -378,6 +379,7 @@ public class SkipLimitStepFactoryBean<T> extends SimpleStepFactoryBean<T> {
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
contribution.combineSkipCounts();
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
logger.debug("Skipping failed input", e);
@@ -421,6 +423,7 @@ public class SkipLimitStepFactoryBean<T> extends SimpleStepFactoryBean<T> {
public Object recover(RetryContext context) {
Throwable t = context.getLastThrowable();
if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) {
contribution.incrementWriteSkipCount();
try {
listener.onSkipInWrite(item, t);
}
@@ -428,13 +431,11 @@ public class SkipLimitStepFactoryBean<T> extends SimpleStepFactoryBean<T> {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t);
}
}
contribution.incrementWriteSkipCount();
return null;
}
});
retryOperations.execute(retryCallback);
}
}
}

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

@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
package org.springframework.batch.core.step.skip;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.UnexpectedJobExecutionException;

8
spring-batch-core/src/test/java/org/springframework/batch/core/AbstractExceptionTests.java
View File

@@ -16,15 +16,19 @@
package org.springframework.batch.core;
import junit.framework.TestCase;
import static org.junit.Assert.assertEquals;
public abstract class AbstractExceptionTests extends TestCase {
import org.junit.Test;
public abstract class AbstractExceptionTests {
@Test
public void testExceptionString() throws Exception {
Exception exception = getException("foo");
assertEquals("foo", exception.getMessage());
}
@Test
public void testExceptionStringThrowable() throws Exception {
Exception exception = getException("foo", new IllegalStateException());
assertEquals("foo", exception.getMessage().substring(0, 3));

36
spring-batch-core/src/test/java/org/springframework/batch/core/listener/AbstractDoubleExceptionTests.java Normal file
View File

@@ -0,0 +1,36 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.listener;
import static org.junit.Assert.assertEquals;
import org.junit.Test;
/**
* @author Dave Syer
*
*/
public abstract class AbstractDoubleExceptionTests {
@Test
public void testExceptionStringThrowable() throws Exception {
Exception exception = getException("foo", new IllegalStateException(), new RuntimeException("bar"));
assertEquals("foo", exception.getMessage().substring(0, 3));
}
public abstract Exception getException(String msg, RuntimeException cause, Throwable e) throws Exception;
}

36
spring-batch-core/src/test/java/org/springframework/batch/core/listener/StepListenerFailedExceptionTests.java Normal file
View File

@@ -0,0 +1,36 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.listener;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
/**
* @author Dave Syer
*
*/
public class StepListenerFailedExceptionTests extends AbstractDoubleExceptionTests {
/* (non-Javadoc)
* @see org.springframework.batch.core.listener.AbstractDoubleExceptionTests#getException(java.lang.String, java.lang.Throwable, java.lang.Throwable)
*/
@Override
public Exception getException(String msg, RuntimeException cause, Throwable e) throws Exception {
return new SkipListenerFailedException(msg, cause, e);
}
}

367
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java
View File

@@ -1,367 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import junit.framework.TestCase;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.SkipListener;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.UnexpectedJobExecutionException;
import org.springframework.batch.core.listener.SkipListenerSupport;
import org.springframework.batch.core.step.StepSupport;
import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy;
import org.springframework.batch.item.ClearFailedException;
import org.springframework.batch.item.FlushFailedException;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.MarkFailedException;
import org.springframework.batch.item.NoWorkFoundException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.ResetFailedException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.transaction.support.TransactionSynchronizationManager;
/**
* @author Dave Syer
*
*/
public class ItemSkipPolicyItemHandlerTests extends TestCase {
private final SkipWriterStub writer = new SkipWriterStub();
private ItemSkipPolicyItemHandler<Holder> handler = new ItemSkipPolicyItemHandler<Holder>(new SkipReaderStub(), writer);
private StepContribution contribution = new StepContribution(new JobExecution(new JobInstance(new Long(11),
new JobParameters(), "jobName")).createStepExecution(new StepSupport("foo")));
protected void tearDown() throws Exception {
// remove the resource if it exists
handler.mark();
}
public void testReadWithNoSkip() throws Exception {
assertEquals(new Holder("1"), handler.read(contribution));
try {
handler.read(contribution);
fail("Expected SkippableException");
}
catch (SkippableException e) {
// expected
}
assertEquals(0, contribution.getSkipCount());
assertEquals(new Holder("3"), handler.read(contribution));
}
public void testReadWithSkip() throws Exception {
handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
assertEquals(new Holder("1"), handler.read(contribution));
assertEquals(new Holder("3"), handler.read(contribution));
contribution.combineSkipCounts();
assertEquals(1, contribution.getReadSkipCount());
assertEquals(new Holder("4"), handler.read(contribution));
}
public void testWriteWithNoSkip() throws Exception {
handler.write(new Holder("3"), contribution);
try {
handler.write(new Holder("4"), contribution);
fail("Expected SkippableException");
}
catch (SkippableException e) {
// expected
}
assertEquals(0, contribution.getSkipCount());
}
public void testHandleWithSkip() throws Exception {
handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
handler.handle(contribution);
handler.handle(contribution);
contribution.combineSkipCounts();
assertEquals(1, contribution.getSkipCount());
// 2 is skipped so 3 was last one processed and now we are at 4
try {
handler.handle(contribution);
fail("Expected SkippableException");
}
catch (SkippableException e) {
// expected
}
assertEquals(3, contribution.getItemCount());
assertEquals(2, contribution.getSkipCount());
// No "4" because it was skipped on write
assertEquals(new Holder("5"), handler.read(contribution));
}
@SuppressWarnings("unchecked")
public void testWriteWithSkipAfterMark() throws Exception {
handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
try {
handler.write(new Holder("4"), contribution);
fail("Expected SkippableException");
}
catch (SkippableException e) {
// expected
}
handler.handle(contribution);
handler.handle(contribution);
contribution.combineSkipCounts();
assertEquals(2, contribution.getSkipCount());
// 2 is skipped so 3 was last one processed and now we are at 4, which
// was previously skipped
handler.handle(contribution);
assertEquals(null, handler.read(contribution));
assertEquals(3, contribution.getItemCount());
assertEquals(2, contribution.getSkipCount());
assertEquals(1, TransactionSynchronizationManager.getResourceMap().size());
Set<Object> removed = (Set<Object>) TransactionSynchronizationManager.getResourceMap().values().iterator().next();
// one skipped item was detected on read
assertEquals(1, removed.size());
// mark() should remove the set of removed keys
handler.mark();
assertEquals(0, TransactionSynchronizationManager.getResourceMap().size());
}
public void testWriteWithSkipAndItemKeyGenerator() throws Exception {
handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
handler.setItemKeyGenerator(new ItemKeyGenerator() {
public Object getKey(Object item) {
return ((Holder) item).value;
}
});
handler.write(new Holder("3"), contribution);
try {
handler.write(new Holder("4"), contribution);
fail("Expected SkippableException");
}
catch (SkippableException e) {
// expected
}
assertEquals(1, contribution.getSkipCount());
assertEquals(new Holder("1"), handler.read(contribution));
assertEquals(new Holder("3"), handler.read(contribution));
contribution.combineSkipCounts();
assertEquals(2, contribution.getSkipCount());
// No "4" because it was skipped on write
assertEquals(new Holder("5"), handler.read(contribution));
}
public void testWriteWithSkipWhenMutating() throws Exception {
handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
writer.mutate = true;
handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
handler.handle(contribution);
handler.handle(contribution);
contribution.combineSkipCounts();
assertEquals(1, contribution.getSkipCount());
// 2 is skipped so 3 was last one processed and now we are at 4
try {
handler.handle(contribution);
fail("Expected SkippableException");
}
catch (SkippableException e) {
// expected
}
assertEquals(3, contribution.getItemCount());
assertEquals(2, contribution.getSkipCount());
// No "4" because it was skipped on write, even though it is mutating
// its key
assertEquals(new Holder("5"), handler.read(contribution));
}
public void testWriteWithSkipCapacityBreached() throws Exception {
handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
handler.setSkipCacheCapacity(0);
handler.handle(contribution);
handler.handle(contribution);
contribution.combineSkipCounts();
assertEquals(1, contribution.getSkipCount());
// 2 is skipped so 3 was last one processed and now we are at 4
try {
handler.handle(contribution);
fail("Expected UnexpectedJobExecutionException");
}
catch (UnexpectedJobExecutionException e) {
// expected
String message = e.getMessage();
assertTrue("Message does not contain 'capacity': " + message, message.indexOf("capacity") >= 0);
}
assertEquals(2, contribution.getSkipCount());
// No "4" because it was skipped on write, even though it is mutating
// its key
assertEquals(new Holder("5"), handler.read(contribution));
}
/**
* Skippable write exceptions are not re-thrown when included in the
* {@link ItemSkipPolicyItemHandler#setDoNotRethrowExceptionClasses(Class[])}
*/
public void testWriteWithSkipAndDoNotRethrow() throws Exception {
handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
handler.setDoNotRethrowExceptionClasses(new Class[] { SkippableException.class });
handler.handle(contribution);
handler.handle(contribution);
contribution.combineSkipCounts();
assertEquals(1, contribution.getSkipCount());
// skippable exception thrown in writer at this point, but it won't be
// re-thrown
handler.handle(contribution);
assertEquals(3, contribution.getItemCount());
assertEquals(2, contribution.getSkipCount());
// No "4" because it was skipped on write, even though it is mutating
// its key
assertEquals(new Holder("5"), handler.read(contribution));
}
/**
* {@link SkipListener#onSkipInWrite(Object, Throwable)} should be called
* also if the exception is not re-thrown (does not cause rollback).
*/
public void testHandleWithSkipAndListeners() throws Exception {
class SkipOnWriteListener extends SkipListenerSupport {
boolean called = false;
public void onSkipInWrite(Object item, Throwable t) {
called = true;
}
}
;
SkipOnWriteListener skipOnWriteListener = new SkipOnWriteListener();
handler.setSkipListeners(new SkipListener[] { skipOnWriteListener });
handler.setDoNotRethrowExceptionClasses(new Class[] { SkippableException.class });
handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy());
for (int i = 0; i < 5; i++) {
handler.handle(contribution);
}
assertTrue("onSkipInWrite should be called although the exception is not re-thrown", skipOnWriteListener.called);
}
/**
* Simple item reader that supports skip functionality.
*/
private static class SkipReaderStub implements ItemReader<Holder> {
final String[] values = { "1", "2", "3", "4", "5" };
Collection<Object> processed = new ArrayList<Object>();
int counter = -1;
int marked = 0;
public Holder read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
counter++;
if (counter == 1) {
throw new SkippableException("exception in reader");
}
if (counter >= values.length) {
return null;
}
Holder item = new Holder(values[counter]);
processed.add(item);
return item;
}
public void mark() throws MarkFailedException {
marked = counter;
}
public void reset() throws ResetFailedException {
counter = marked;
}
}
/**
* Simple item writer that supports skip functionality.
*/
private static class SkipWriterStub implements ItemWriter<Holder> {
boolean mutate = false;
List<Object> written = new ArrayList<Object>();
int flushIndex = -1;
public void clear() throws ClearFailedException {
for (int i = flushIndex + 1; i < written.size(); i++) {
written.remove(i);
}
}
public void flush() throws FlushFailedException {
flushIndex = written.size() - 1;
}
public void write(Holder item) throws Exception {
String value = item.value;
written.add(item);
if (mutate) {
item.value = "done";
}
if (value.equals("4")) {
throw new SkippableException("exception in writer");
}
}
}
private static class SkippableException extends Exception {
public SkippableException(String message) {
super(message);
}
}
private static class Holder {
private String value = null;
public Holder(String value) {
super();
this.value = value;
}
public boolean equals(Object obj) {
return obj instanceof Holder && value.equals(((Holder) obj).value);
}
public int hashCode() {
return value.hashCode();
}
public String toString() {
return "[holder:" + value + "]";
}
}
}

19
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitReadFailurePolicyTests.java → spring-batch-core/src/test/java/org/springframework/batch/core/step/item/LimitCheckingItemSkipPolicyTests.java
View File

@@ -15,27 +15,31 @@
*/
package org.springframework.batch.core.step.item;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.FileNotFoundException;
import java.util.ArrayList;
import java.util.List;
import junit.framework.TestCase;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.item.file.FlatFileParseException;
/**
* @author Lucas Ward
* @author Dave Syer
*
*/
public class SkipLimitReadFailurePolicyTests extends TestCase {
public class LimitCheckingItemSkipPolicyTests {
private LimitCheckingItemSkipPolicy failurePolicy;
protected void setUp() throws Exception {
super.setUp();
@Before
public void setUp() throws Exception {
List<Class<?>> skippableExceptions = new ArrayList<Class<?>>();
skippableExceptions.add(FlatFileParseException.class);
List<Class<?>> fatalExceptions = new ArrayList<Class<?>>();
@@ -43,6 +47,7 @@ public class SkipLimitReadFailurePolicyTests extends TestCase {
failurePolicy = new LimitCheckingItemSkipPolicy(1, skippableExceptions, fatalExceptions);
}
@Test
public void testLimitExceed(){
try{
failurePolicy.shouldSkip(new FlatFileParseException("", ""), 2);
@@ -53,10 +58,12 @@ public class SkipLimitReadFailurePolicyTests extends TestCase {
}
}
@Test
public void testNonSkippableException(){
assertFalse(failurePolicy.shouldSkip(new FileNotFoundException(), 2));
}
@Test
public void testSkip(){
assertTrue(failurePolicy.shouldSkip(new FlatFileParseException("",""), 0));
}

67
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/MockItemReader.java
View File

@@ -1,67 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.MarkFailedException;
import org.springframework.batch.item.ResetFailedException;
public class MockItemReader implements ItemReader<String> {
private final int returnItemCount;
private int returnedItemCount;
private boolean fail = false;
public MockItemReader() {
this(-1);
}
public MockItemReader(int returnItemCount) {
this.returnItemCount = returnItemCount;
}
public void setFail(boolean fail) {
this.fail = fail;
}
public void close() {
}
public String read() {
if(fail) {
fail = false;
throw new RuntimeException();
}
if (returnItemCount < 0 || returnedItemCount < returnItemCount) {
return String.valueOf(returnedItemCount++);
}
return null;
}
public Object getKey(Object item) {
return null;
}
public void mark() throws MarkFailedException {
}
public void reset() throws ResetFailedException {
}
}

89
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
View File

@@ -14,9 +14,12 @@ import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.SkipListenerSupport;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.core.step.JobRepositorySupport;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.item.ClearFailedException;
import org.springframework.batch.item.FlushFailedException;
import org.springframework.batch.item.ItemReader;
@@ -91,7 +94,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,3,5"));
assertEquals(expectedOutput, writer.written);
assertEquals(4, stepExecution.getItemCount().intValue());
}
@@ -117,7 +120,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
// no rollbacks
assertEquals(0, stepExecution.getRollbackCount().intValue());
assertEquals(4, stepExecution.getItemCount().intValue());
}
@@ -208,14 +211,90 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
assertFalse(reader.processed.contains("2"));
assertTrue(reader.processed.contains("4"));
// failure on "5" tripped the skip limit but "4" failed on write and was skipped and
// RepeatSynchronizationManager.setCompleteOnly() was called in the retry policy to
// failure on "5" tripped the skip limit but "4" failed on write and was
// skipped and
// RepeatSynchronizationManager.setCompleteOnly() was called in the
// retry policy to
// aggressively commit after a recovery ("1" was written at that point)
List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1"));
assertEquals(expectedOutput, writer.written);
}
/**
* Check items causing errors are skipped as expected.
*/
@SuppressWarnings("unchecked")
public void testSkipListenerFailsOnRead() throws Exception {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
.commaDelimitedListToSet("2,3,5"));
factory.setSkipLimit(3);
factory.setItemReader(reader);
factory.setListeners(new StepListener[] { new SkipListenerSupport() {
@Override
public void onSkipInRead(Throwable t) {
throw new RuntimeException("oops");
}
} });
factory.setSkippableExceptionClasses(new Class[] { Exception.class });
AbstractStep step = (AbstractStep) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
try {
step.execute(stepExecution);
fail("Expected SkipListenerFailedException.");
}
catch (SkipListenerFailedException e) {
assertEquals("oops", e.getCause().getMessage());
}
assertEquals(1, stepExecution.getSkipCount());
assertEquals(1, stepExecution.getReadSkipCount().intValue());
assertEquals(0, stepExecution.getWriteSkipCount().intValue());
}
/**
* Check items causing errors are skipped as expected.
*/
@SuppressWarnings("unchecked")
public void testSkipListenerFailsOnWrite() throws Exception {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
.commaDelimitedListToSet("2,3,5"));
factory.setSkipLimit(3);
factory.setItemReader(reader);
factory.setListeners(new StepListener[] { new SkipListenerSupport() {
@Override
public void onSkipInWrite(Object item, Throwable t) {
throw new RuntimeException("oops");
}
} });
factory.setSkippableExceptionClasses(new Class[] { Exception.class });
AbstractStep step = (AbstractStep) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
try {
step.execute(stepExecution);
fail("Expected SkipListenerFailedException.");
}
catch (SkipListenerFailedException e) {
assertEquals("oops", e.getCause().getMessage());
}
assertEquals(1, stepExecution.getSkipCount());
assertEquals(0, stepExecution.getReadSkipCount().intValue());
assertEquals(1, stepExecution.getWriteSkipCount().intValue());
}
/**
* Check items causing errors are skipped as expected.
*/
@@ -383,7 +462,7 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
public void clear() throws ClearFailedException {
for (int i = flushIndex + 1; i < written.size(); i++) {
written.remove(written.size()-1);
written.remove(written.size() - 1);
}
}

35
spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/SkipListenerFailedExceptionTests.java Normal file
View File

@@ -0,0 +1,35 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.skip;
import org.springframework.batch.core.listener.AbstractDoubleExceptionTests;
/**
* @author Dave Syer
*
*/
public class SkipListenerFailedExceptionTests extends AbstractDoubleExceptionTests {
/* (non-Javadoc)
* @see org.springframework.batch.core.listener.AbstractDoubleExceptionTests#getException(java.lang.String, java.lang.RuntimeException, java.lang.Throwable)
*/
@Override
public Exception getException(String msg, RuntimeException cause, Throwable e) throws Exception {
return new SkipListenerFailedException(msg, cause, e);
}
}