From c230e9e0d3e927d94e55444b0d066ea189207b2f Mon Sep 17 00:00:00 2001 From: dsyer Date: Wed, 6 Aug 2008 10:44:13 +0000 Subject: [PATCH] OPEN - issue BATCH-753: Listener exception handling Add test for SkipListenerFailedException --- .../step/item/ItemSkipPolicyItemHandler.java | 323 --------------- .../step/item/SkipLimitStepFactoryBean.java | 5 +- .../SkipListenerFailedException.java | 2 +- .../batch/core/AbstractExceptionTests.java | 8 +- .../AbstractDoubleExceptionTests.java | 36 ++ .../StepListenerFailedExceptionTests.java | 36 ++ .../item/ItemSkipPolicyItemHandlerTests.java | 367 ------------------ ... => LimitCheckingItemSkipPolicyTests.java} | 19 +- .../batch/core/step/item/MockItemReader.java | 67 ---- .../item/SkipLimitStepFactoryBeanTests.java | 89 ++++- .../SkipListenerFailedExceptionTests.java | 35 ++ 11 files changed, 214 insertions(+), 773 deletions(-) delete mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java rename spring-batch-core/src/main/java/org/springframework/batch/core/step/{item => skip}/SkipListenerFailedException.java (96%) create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/listener/AbstractDoubleExceptionTests.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/listener/StepListenerFailedExceptionTests.java delete mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java rename spring-batch-core/src/test/java/org/springframework/batch/core/step/item/{SkipLimitReadFailurePolicyTests.java => LimitCheckingItemSkipPolicyTests.java} (85%) delete mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/step/item/MockItemReader.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/SkipListenerFailedExceptionTests.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java deleted file mode 100644 index f3e5496e6..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java +++ /dev/null @@ -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 extends SimpleItemHandler { - - /** - * 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 skippedExceptions = new HashMap(); - - 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 itemReader, ItemWriter 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()); - } - ((Set) 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) 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; - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java index 0a929286a..fa8804b12 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java @@ -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 extends SimpleStepFactoryBean { 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 extends SimpleStepFactoryBean { 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 extends SimpleStepFactoryBean { throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t); } } - contribution.incrementWriteSkipCount(); return null; } }); retryOperations.execute(retryCallback); } - } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipListenerFailedException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipListenerFailedException.java similarity index 96% rename from spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipListenerFailedException.java rename to spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipListenerFailedException.java index 02cb95948..0e7d8e745 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipListenerFailedException.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/SkipListenerFailedException.java @@ -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; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/AbstractExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/AbstractExceptionTests.java index 3fa611c9b..ee9b02847 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/AbstractExceptionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/AbstractExceptionTests.java @@ -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)); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/listener/AbstractDoubleExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/AbstractDoubleExceptionTests.java new file mode 100644 index 000000000..f91ab2299 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/AbstractDoubleExceptionTests.java @@ -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; + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/listener/StepListenerFailedExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/StepListenerFailedExceptionTests.java new file mode 100644 index 000000000..2e840e388 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/listener/StepListenerFailedExceptionTests.java @@ -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); + } + + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java deleted file mode 100644 index 1b50823bb..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java +++ /dev/null @@ -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 handler = new ItemSkipPolicyItemHandler(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 removed = (Set) 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 { - - final String[] values = { "1", "2", "3", "4", "5" }; - - Collection processed = new ArrayList(); - - 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 { - - boolean mutate = false; - - List written = new ArrayList(); - - 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 + "]"; - } - } -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitReadFailurePolicyTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/LimitCheckingItemSkipPolicyTests.java similarity index 85% rename from spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitReadFailurePolicyTests.java rename to spring-batch-core/src/test/java/org/springframework/batch/core/step/item/LimitCheckingItemSkipPolicyTests.java index 542fc1af4..2f68fea82 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitReadFailurePolicyTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/LimitCheckingItemSkipPolicyTests.java @@ -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> skippableExceptions = new ArrayList>(); skippableExceptions.add(FlatFileParseException.class); List> fatalExceptions = new ArrayList>(); @@ -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)); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/MockItemReader.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/MockItemReader.java deleted file mode 100644 index 6de928123..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/MockItemReader.java +++ /dev/null @@ -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 { - - 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 { - } - -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java index 2a99afd23..d6b06a037 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java @@ -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 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 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); } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/SkipListenerFailedExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/SkipListenerFailedExceptionTests.java new file mode 100644 index 000000000..f2e660a2f --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/SkipListenerFailedExceptionTests.java @@ -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); + } + +}