diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java index dd400fe02..382c6bab6 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecution.java @@ -48,7 +48,7 @@ public class StepExecution extends Entity { private volatile int rollbackCount = 0; private volatile int readSkipCount = 0; - + private volatile int writeSkipCount = 0; private volatile Date startTime = new Date(System.currentTimeMillis()); @@ -234,38 +234,6 @@ public class StepExecution extends Entity { return null; } - /* - * (non-Javadoc) - * - * @see org.springframework.batch.container.common.domain.Entity#equals(java.lang.Object) - */ - public boolean equals(Object obj) { - - Object jobExecutionId = getJobExecutionId(); - if (jobExecutionId == null || !(obj instanceof StepExecution) || getId() == null) { - return super.equals(obj); - } - StepExecution other = (StepExecution) obj; - - return stepName.equals(other.getStepName()) && (jobExecutionId.equals(other.getJobExecutionId())); - } - - /* - * (non-Javadoc) - * - * @see org.springframework.batch.container.common.domain.Entity#hashCode() - */ - public int hashCode() { - Object jobExecutionId = getJobExecutionId(); - return super.hashCode() + 31 * (stepName != null ? stepName.hashCode() : 0) + 91 - * (jobExecutionId != null ? jobExecutionId.hashCode() : 0); - } - - public String toString() { - return super.toString() + ", name=" + stepName + ", itemCount=" + itemCount + ", commitCount=" - + commitCount + ", rollbackCount=" + rollbackCount; - } - /** * @param exitStatus */ @@ -401,8 +369,7 @@ public class StepExecution extends Entity { } /** - * Set the time when the StepExecution was last updated before - * persisting + * Set the time when the StepExecution was last updated before persisting * * @param lastUpdated */ @@ -410,5 +377,40 @@ public class StepExecution extends Entity { this.lastUpdated = lastUpdated; } - + /* + * (non-Javadoc) + * + * @see + * org.springframework.batch.container.common.domain.Entity#equals(java. + * lang.Object) + */ + public boolean equals(Object obj) { + + Object jobExecutionId = getJobExecutionId(); + if (jobExecutionId == null || !(obj instanceof StepExecution) || getId() == null) { + return super.equals(obj); + } + StepExecution other = (StepExecution) obj; + + return stepName.equals(other.getStepName()) && (jobExecutionId.equals(other.getJobExecutionId())); + } + + /* + * (non-Javadoc) + * + * @see org.springframework.batch.container.common.domain.Entity#hashCode() + */ + public int hashCode() { + Object jobExecutionId = getJobExecutionId(); + return super.hashCode() + 31 * (stepName != null ? stepName.hashCode() : 0) + 91 + * (jobExecutionId != null ? jobExecutionId.hashCode() : 0); + } + + public String toString() { + return super.toString() + + String.format(", name=%s, itemCount=%d, readSkipCount=%d, writeSkipCount=%d" + + ", commitCount=%d, rollbackCount=%d", stepName, itemCount, readSkipCount, writeSkipCount, + commitCount, rollbackCount); + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/StepHandlerStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/StepHandlerStep.java index c9db54897..81ddea5ef 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/StepHandlerStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/handler/StepHandlerStep.java @@ -260,6 +260,7 @@ public class StepHandlerStep extends AbstractStep { // Apply the contribution to the step // even if unsuccessful + logger.debug("Applying contribution: " + contribution); stepExecution.apply(contribution); } @@ -312,6 +313,7 @@ public class StepHandlerStep extends AbstractStep { } try { + logger.debug("Saving step execution after commit: " + stepExecution); getJobRepository().update(stepExecution); } catch (Exception e) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java index d575fad5e..85a499263 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java @@ -5,9 +5,13 @@ import java.util.Collections; import java.util.Iterator; import java.util.List; -import org.springframework.util.Assert; - /** + * Encapsulation of a list of items to be processed and possibly a list of + * failed items to be skipped. To mark an item as skipped clients should iterate + * over the chunk using the {@link #iterator()} method, and if there is a + * failure call {@link ChunkIterator#remove(Exception)} on the iterator. The + * skipped items are then available through the chunk. + * * @author Dave Syer * */ @@ -15,13 +19,7 @@ class Chunk implements Iterable { private List items = new ArrayList(); - private int current = 0; - - private int last = 0; - - private Exception exception = null; - - private boolean skipped = false; + private List> skips = new ArrayList>(); /** * Add the item to the chunk. @@ -29,7 +27,6 @@ class Chunk implements Iterable { */ public void add(W item) { items.add(item); - last++; } /** @@ -37,15 +34,20 @@ class Chunk implements Iterable { */ public void clear() { items.clear(); - last = 0; - current = 0; } /** * @return a copy of the items to be processed as an unmodifiable list */ public List getItems() { - return Collections.unmodifiableList(new ArrayList(items.subList(current, last))); + return Collections.unmodifiableList(new ArrayList(items)); + } + + /** + * @return a copy of the skips as an unmodifiable list + */ + public List> getSkips() { + return Collections.unmodifiableList(new ArrayList>(skips)); } /** @@ -59,92 +61,8 @@ class Chunk implements Iterable { * Get an unmodifiable iterator for the underlying items. * @see java.lang.Iterable#iterator() */ - public Iterator iterator() { - return getItems().iterator(); - } - - /** - * @return true if the chunk is ready for a retry attempt - */ - public boolean canRetry() { - return exception == null || canSkip(); - } - - /** - * Re-throw the last exception if there was one, and reset. Subsequent calls - * would do nothing until {@link #rethrow(Exception)} is called. - * - * @throws Exception if there is a last exception - */ - public void rethrow() throws Exception { - int size = items.size(); - Exception throwable = exception; - if (exception != null && !skipped) { - if (isComplete()) { - // we tried all items and there was no exception - exception = null; - } - else { - // we tried some but not all elements with no exception - current = last; - } - } - if (skipped) { - skipped = false; - } - last = size; // reset end point of scan - if (current == size) { - // we scanned all the elements - current = 0; - exception = null; - } - if (throwable != null) { - throw throwable; - } - } - - /** - * @return true if the current item slice includes all items - */ - public boolean isComplete() { - return current == 0 && last == items.size(); - } - - /** - * Get the skipped item and remove it from the backing list. - * @return the item that can be skipped - */ - public W getSkippedItem() { - Assert.state(canSkip(), "To remove a skipped item it has to be unique"); - W item = items.remove(current); - if (last > items.size()) { - last = items.size(); - } - skipped = true; - return item; - } - - /** - * @param e an exception to register and re-throw - * @throws Exception the exception passed in - */ - public void rethrow(Exception e) throws Exception { - exception = e; - // narrow the search for the failed item - last = current + (last - current) / 2; - // ... unless it would lead to processing no data - if (last==current) { - last = current + 1; - } - throw e; - } - - /** - * @return true if there is a single item waiting, so it can be identified - * and passed to listeners - */ - public boolean canSkip() { - return current == last - 1 && current < items.size(); + public ChunkIterator iterator() { + return new ChunkIterator(items); } /* @@ -154,7 +72,86 @@ class Chunk implements Iterable { */ @Override public String toString() { - return String.format("items=%s, canSkip=%s, current=%d, last=%d", items, canSkip(), current, last); + return String.format("[items=%s, skips=%s]", items, skips); + } + + /** + * Special iterator for a chunk providing the {@link #remove(Exception)} + * method for dynamically removing an item abd adding it to the skips. + * + * @author Dave Syer + * + */ + public class ChunkIterator implements Iterator { + + final private Iterator iterator; + + private W next; + + public ChunkIterator(List items) { + iterator = items.iterator(); + } + + public boolean hasNext() { + return iterator.hasNext(); + } + + public W next() { + next = iterator.next(); + return next; + } + + public void remove(Exception e) { + if (next != null) { + skips.add(new SkippedItem(next, e)); + } + iterator.remove(); + } + + public void remove() { + throw new UnsupportedOperationException("To remove an item you must provide an exception."); + } + + } + + /** + * Wrapper for a skipped item and its exception. + * + * @author Dave Syer + * + */ + public static class SkippedItem { + + final private Exception exception; + + final private T item; + + public SkippedItem(T item, Exception e) { + this.item = item; + this.exception = e; + } + + /** + * Public getter for the exception. + * @return the exception + */ + public Exception getException() { + return exception; + } + + /** + * Public getter for the item. + * @return the item + */ + public T getItem() { + return item; + } + + @Override + public String toString() { + return String.format("[exception=%s, item=%s]", exception, item); + } + } } \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java index b37830ba0..d0e1b58fe 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java @@ -108,6 +108,11 @@ public class ItemOrientedStepHandler implements StepHandler { return ExitStatus.CONTINUABLE; } }); + + // If there is no input we don't have to do anything more + if (inputs.isEmpty()) { + return result; + } storeInputs(attributes, inputs); @@ -141,7 +146,6 @@ public class ItemOrientedStepHandler implements StepHandler { clearAll(attributes); } - logger.info("Contribution: " + contribution); return result; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java deleted file mode 100644 index a192a96b0..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java +++ /dev/null @@ -1,33 +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.core.Step; -import org.springframework.batch.repeat.RepeatOperations; - -/** - * Factory bean for {@link Step} implementations allowing registration of - * listeners and also direct injection of the {@link RepeatOperations} needed at - * step and chunk level. - * - * @deprecated use the {@link SimpleStepFactoryBean} instead - * - * @author Dave Syer - * - */ -public class RepeatOperationsStepFactoryBean extends SimpleStepFactoryBean { - -} 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 b19838425..98dd34f31 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 @@ -2,6 +2,7 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -374,38 +375,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean * the next transaction automatically.
*/ @Override - protected void write(final Chunk chunk, StepContribution contribution) throws Exception { - - if (!chunk.canRetry()) { - logger.debug("Run items: " + chunk.getItems()); - runChunk(chunk, contribution); - } - else { - logger.debug(String.format("Retry items: %s", chunk.getItems())); - retryChunk(chunk, contribution); - } - chunk.rethrow(); - - chunk.clear(); - - } - - /** - * @param chunk - */ - private void runChunk(Chunk chunk, final StepContribution contribution) throws Exception { - try { - doWrite(chunk.getItems()); - } - catch (Exception e) { - chunk.rethrow(e); - } - } - - /** - * @param chunk - */ - private void retryChunk(final Chunk chunk, final StepContribution contribution) throws Exception { + protected void write(final Chunk chunk, final StepContribution contribution) throws Exception { RetryCallback retryCallback = new RetryCallback() { public Object doWithRetry(RetryContext context) throws Exception { @@ -422,36 +392,42 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean Exception t = (Exception) context.getLastThrowable(); - if (!chunk.canSkip()) { - throw t; + for (Chunk.ChunkIterator iterator = chunk.iterator(); iterator.hasNext();) { + S item = iterator.next(); + try { + doWrite(Collections.singletonList(item)); + } + catch (Exception e) { + if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) { + iterator.remove(e); + contribution.incrementWriteSkipCount(); + throw e; + } + else { + throw new RetryException("Non-skippable exception in recoverer", t); + } + } } - if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) { - contribution.incrementWriteSkipCount(); - S item = chunk.getSkippedItem(); - try { - getListener().onSkipInWrite(item, t); - return null; - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t); - } - } - else { - throw new RetryException("Non-skippable exception in recoverer", t); - } + return null; } }; - try { - retryOperations.execute(retryCallback, recoveryCallback, new RetryState(chunk)); - } - catch (Exception e) { - // only if the retry failed do we re-arrange the chunk - chunk.rethrow(e); + retryOperations.execute(retryCallback, recoveryCallback, new RetryState(chunk)); + + for (Chunk.SkippedItem skip : chunk.getSkips()) { + Exception exception = skip.getException(); + try { + getListener().onSkipInWrite(skip.getItem(), exception); + } + catch (RuntimeException e) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, exception); + } } + chunk.clear(); + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/AlmostStatefulRetryChunkTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/AlmostStatefulRetryChunkTests.java index c5d7662a0..26b62fb17 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/AlmostStatefulRetryChunkTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/AlmostStatefulRetryChunkTests.java @@ -37,10 +37,6 @@ import org.junit.runners.Parameterized.Parameters; @RunWith(Parameterized.class) public class AlmostStatefulRetryChunkTests { - private enum CallType { - RUN, RETRY; - } - private Log logger = LogFactory.getLog(getClass()); private final Chunk chunk; @@ -53,8 +49,6 @@ public class AlmostStatefulRetryChunkTests { private int count = 0; - private Object lastCallType; - public AlmostStatefulRetryChunkTests(String[] args, int limit) { chunk = new Chunk(); for (String string : args) { @@ -67,88 +61,78 @@ public class AlmostStatefulRetryChunkTests { public void testRetry() throws Exception { logger.debug("Starting simple scenario"); List items = new ArrayList(chunk.getItems()); + int before = items.size(); items.removeAll(Collections.singleton("fail")); boolean error = true; while (error && count++ < BACKSTOP_LIMIT) { try { - if (!chunk.canRetry()) { - // success - logger.debug("Run items: " + chunk.getItems()); - lastCallType = CallType.RUN; - runChunk(chunk); - } - else { - logger.debug(String.format("Retry (attempts=%d) items: %s", retryAttempts, chunk.getItems())); - lastCallType = CallType.RETRY; - try { - retryChunk(chunk); - } - catch (Exception e) { - chunk.rethrow(e); - } - - } - chunk.rethrow(); + statefulRetry(chunk); error = false; } catch (Exception e) { error = true; } } - logger.debug("Items: " + chunk.getItems()); + logger.debug("Chunk: " + chunk); assertTrue("Backstop reached. Probably an infinite loop...", count < BACKSTOP_LIMIT); - assertEquals(CallType.RETRY, lastCallType); assertFalse(chunk.getItems().contains("fail")); assertEquals(items, chunk.getItems()); + assertEquals(before-chunk.getItems().size(), chunk.getSkips().size()); } /** * @param chunk * @throws Exception */ - private void retryChunk(Chunk chunk) throws Exception { - try { - // N.B. a classic stateful retry goes straight to recovery here - doWrite(chunk); - retryAttempts = 0; - } - catch (Exception e) { - if (++retryAttempts > retryLimit) { - // recovery + private void statefulRetry(Chunk chunk) throws Exception { + if (retryAttempts <= retryLimit) { + try { + // N.B. a classic stateful retry goes straight to recovery here + logger.debug(String.format("Retry (attempts=%d) chunk: %s", retryAttempts, chunk)); + doWrite(chunk.getItems()); retryAttempts = 0; - if (chunk.canSkip()) { - chunk.getSkippedItem(); - } - else { - throw e; - } } - else { + catch (Exception e) { + retryAttempts++; // stateful retry always rethrow throw e; } } + else { + try { + logger.debug(String.format("Recover (attempts=%d) chunk: %s", retryAttempts, chunk)); + recover(chunk); + } + finally { + retryAttempts = 0; + } + } + // recovery + return; + + } + + /** + * @param chunk + * @throws Exception + */ + private void recover(Chunk chunk) throws Exception { + for (Chunk.ChunkIterator iterator = chunk.iterator(); iterator.hasNext();) { + String string = iterator.next(); + try { + doWrite(Collections.singletonList(string)); + } catch (Exception e) { + iterator.remove(e); + throw e; + } + } } /** * @param chunk * @throws Exception */ - private void runChunk(Chunk chunk) throws Exception { - try { - doWrite(chunk); - } - catch (Exception e) { - chunk.rethrow(e); - } - } - - /** - * @param chunk - * @throws Exception - */ - private void doWrite(Chunk chunk) throws Exception { - List items = chunk.getItems(); + private void doWrite(List items) throws Exception { if (items.contains("fail")) { throw new Exception(); } 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 01d19181a..62d30901c 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 @@ -93,7 +93,7 @@ public class SkipLimitStepFactoryBeanTests { // only write exception caused rollback, but more than once because it // has to go back and split the chunk up to isolate the failed item - assertEquals(3, stepExecution.getRollbackCount()); + assertEquals(2, stepExecution.getRollbackCount()); // writer did not skip "2" as it never made it to writer, only "4" did assertTrue(reader.processed.contains("4")); @@ -229,8 +229,8 @@ public class SkipLimitStepFactoryBeanTests { assertFalse(reader.processed.contains("2")); assertTrue(reader.processed.contains("4")); - // "1" was sent to writer but never comitted - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("")); + // only "1" was ever committed + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1")); assertEquals(expectedOutput, writer.written); } @@ -333,16 +333,14 @@ public class SkipLimitStepFactoryBeanTests { StepExecution stepExecution = jobExecution.createStepExecution(step); - // TODO: uncomment this! - // step.execute(stepExecution); - // assertEquals(4, stepExecution.getSkipCount()); - // assertEquals(3, stepExecution.getReadSkipCount()); - // assertEquals(1, stepExecution.getWriteSkipCount()); - // - // // skipped 2,3,4,5 - // List expectedOutput = - // Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6")); - // assertEquals(expectedOutput, writer.written); + step.execute(stepExecution); + assertEquals(4, stepExecution.getSkipCount()); + assertEquals(3, stepExecution.getReadSkipCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + + // skipped 2,3,4,5 + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6")); + assertEquals(expectedOutput, writer.written); } @@ -366,16 +364,14 @@ public class SkipLimitStepFactoryBeanTests { StepExecution stepExecution = jobExecution.createStepExecution(step); - // TODO: uncomment this! - // step.execute(stepExecution); - // assertEquals(4, stepExecution.getSkipCount()); - // assertEquals(2, stepExecution.getReadSkipCount()); - // assertEquals(2, stepExecution.getWriteSkipCount()); - // - // // skipped 2,3,4,5 - // List expectedOutput = - // Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6,7")); - // assertEquals(expectedOutput, writer.written); + step.execute(stepExecution); + assertEquals(4, stepExecution.getSkipCount()); + assertEquals(2, stepExecution.getReadSkipCount()); + assertEquals(2, stepExecution.getWriteSkipCount()); + + // skipped 2,3,4,5 + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6,7")); + assertEquals(expectedOutput, writer.written); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java index 968441611..e2bcfd051 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java @@ -52,7 +52,9 @@ import org.springframework.batch.retry.policy.MapRetryContextCache; import org.springframework.batch.retry.policy.RetryCacheCapacityExceededException; import org.springframework.batch.retry.policy.SimpleRetryPolicy; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.util.StringUtils; /** * @author Dave Syer @@ -70,6 +72,8 @@ public class StatefulRetryStepFactoryBeanTests { private List provided = new ArrayList(); + private List written = TransactionAwareProxyFactory.createTransactionalList(); + int count = 0; private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), @@ -137,8 +141,7 @@ public class StatefulRetryStepFactoryBeanTests { */ @Test public void testSuccessfulRetryWithReadFailure() throws Exception { - List items = Arrays.asList(new String[] { "a", "b", "c" }); - ItemReader provider = new ListItemReader(items) { + ItemReader provider = new ListItemReader(Arrays.asList("a", "b", "c")) { public String read() { String item = super.read(); provided.add(item); @@ -177,8 +180,7 @@ public class StatefulRetryStepFactoryBeanTests { } }); factory.setSkipLimit(2); - List items = Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" }); - ItemReader provider = new ListItemReader(items) { + ItemReader provider = new ListItemReader(Arrays.asList("a", "b", "c", "d", "e", "f")) { public String read() { String item = super.read(); count++; @@ -216,8 +218,7 @@ public class StatefulRetryStepFactoryBeanTests { } } }); factory.setSkipLimit(2); - List items = Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" }); - ItemReader provider = new ListItemReader(items) { + ItemReader provider = new ListItemReader(Arrays.asList("a", "b", "c", "d", "e", "f")) { public String read() { String item = super.read(); logger.debug("Read Called! Item: [" + item + "]"); @@ -231,6 +232,7 @@ public class StatefulRetryStepFactoryBeanTests { public void write(List item) throws Exception { logger.debug("Write Called! Item: [" + item + "]"); processed.addAll(item); + written.addAll(item); if (item.contains("b") || item.contains("d")) { throw new RuntimeException("Write error - planned but recoverable."); } @@ -253,10 +255,13 @@ public class StatefulRetryStepFactoryBeanTests { assertEquals(2, stepExecution.getSkipCount()); assertEquals(2, stepExecution.getWriteSkipCount()); + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,c,e,f")); + assertEquals(expectedOutput, written); + // [a, b, c, d, e, f, null] assertEquals(7, provided.size()); - // [a, b, b, b, b, b, c, d, d, d, d, d, e, f] - assertEquals(14, processed.size()); + // [a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f] + assertEquals(16, processed.size()); // [b, d] assertEquals(2, recovered.size()); } @@ -277,8 +282,7 @@ public class StatefulRetryStepFactoryBeanTests { } } }); factory.setSkipLimit(2); - List items = Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" }); - ItemReader provider = new ListItemReader(items) { + ItemReader provider = new ListItemReader(Arrays.asList("a", "b", "c", "d", "e", "f")) { public String read() { String item = super.read(); logger.debug("Read Called! Item: [" + item + "]"); @@ -292,6 +296,7 @@ public class StatefulRetryStepFactoryBeanTests { public void write(List item) throws Exception { logger.debug("Write Called! Item: [" + item + "]"); processed.addAll(item); + written.addAll(item); if (item.contains("b") || item.contains("d")) { throw new RuntimeException("Write error - planned but recoverable."); } @@ -314,10 +319,13 @@ public class StatefulRetryStepFactoryBeanTests { assertEquals(2, stepExecution.getSkipCount()); assertEquals(2, stepExecution.getWriteSkipCount()); + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,c,e,f")); + assertEquals(expectedOutput, written); + // [a, b, c, d, e, f, null] assertEquals(7, provided.size()); - // [a, b, c, a, b, c, b, b, b, b, b, c, a, c, d, e, f, d, d, d, d, e, f, e, f] - assertEquals(25, processed.size()); + // [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, a, c, d, e, f, d, e, f, d, e, f, d, e, f, d, e, f, d, e, f] + assertEquals(37, processed.size()); // [b, d] assertEquals(2, recovered.size()); } @@ -331,8 +339,7 @@ public class StatefulRetryStepFactoryBeanTests { }); factory.setRetryLimit(4); factory.setSkipLimit(0); - List items = Arrays.asList(new String[] { "b" }); - ItemReader provider = new ListItemReader(items) { + ItemReader provider = new ListItemReader(Arrays.asList("b")) { public String read() { String item = super.read(); provided.add(item); @@ -343,6 +350,7 @@ public class StatefulRetryStepFactoryBeanTests { ItemWriter itemWriter = new ItemWriter() { public void write(List item) throws Exception { processed.addAll(item); + written.addAll(item); logger.debug("Write Called! Item: [" + item + "]"); throw new RuntimeException("Write error - planned but retryable."); } @@ -360,11 +368,15 @@ public class StatefulRetryStepFactoryBeanTests { // expected } + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("")); + assertEquals(expectedOutput, written); + assertEquals(0, stepExecution.getSkipCount()); // [b] assertEquals(1, provided.size()); - // [b, b, b, b] - assertEquals(4, processed.size()); + // the failed items are tried one more time than the limit (TODO: maybe fix this?) + // [b, b, b, b, b] + assertEquals(5, processed.size()); // [] assertEquals(0, recovered.size()); assertEquals(1, stepExecution.getItemCount()); @@ -383,8 +395,7 @@ public class StatefulRetryStepFactoryBeanTests { factory.setRetryableExceptionClasses(new HashSet>()); factory.setSkipLimit(1); - List items = Arrays.asList(new String[] { "b" }); - ItemReader provider = new ListItemReader(items) { + ItemReader provider = new ListItemReader(Arrays.asList("b")) { public String read() { String item = super.read(); provided.add(item); @@ -395,6 +406,7 @@ public class StatefulRetryStepFactoryBeanTests { ItemWriter itemWriter = new ItemWriter() { public void write(List item) throws Exception { processed.addAll(item); + written.addAll(item); logger.debug("Write Called! Item: [" + item + "]"); throw new RuntimeException("Write error - planned but not skippable."); } @@ -414,11 +426,14 @@ public class StatefulRetryStepFactoryBeanTests { assertTrue("Wrong message: " + message, message.contains("Write error - planned but not skippable.")); } + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("")); + assertEquals(expectedOutput, written); + assertEquals(0, stepExecution.getSkipCount()); // [b] assertEquals(1, provided.size()); - // [b] - assertEquals(1, processed.size()); + // [b, b] + assertEquals(2, processed.size()); // [] assertEquals(0, recovered.size()); assertEquals(1, stepExecution.getItemCount()); @@ -428,8 +443,7 @@ public class StatefulRetryStepFactoryBeanTests { public void testRetryPolicy() throws Exception { factory.setRetryPolicy(new SimpleRetryPolicy(4)); factory.setSkipLimit(0); - List items = Arrays.asList(new String[] { "b" }); - ItemReader provider = new ListItemReader(items) { + ItemReader provider = new ListItemReader(Arrays.asList("b")) { public String read() { String item = super.read(); provided.add(item); @@ -440,6 +454,7 @@ public class StatefulRetryStepFactoryBeanTests { ItemWriter itemWriter = new ItemWriter() { public void write(List item) throws Exception { processed.addAll(item); + written.addAll(item); logger.debug("Write Called! Item: [" + item + "]"); throw new RuntimeException("Write error - planned but retryable."); } @@ -457,11 +472,14 @@ public class StatefulRetryStepFactoryBeanTests { // expected } + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("")); + assertEquals(expectedOutput, written); + assertEquals(0, stepExecution.getSkipCount()); // [b] assertEquals(1, provided.size()); - // [b, b, b, b] - assertEquals(4, processed.size()); + // [b, b, b, b, b] + assertEquals(5, processed.size()); // [] assertEquals(0, recovered.size()); assertEquals(1, stepExecution.getItemCount()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatelessRetryChunkTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatelessRetryChunkTests.java deleted file mode 100644 index d59bf8f24..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatelessRetryChunkTests.java +++ /dev/null @@ -1,159 +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 static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; -import org.junit.runners.Parameterized.Parameters; - -/** - * @author Dave Syer - * - */ -@RunWith(Parameterized.class) -public class StatelessRetryChunkTests { - - private Log logger = LogFactory.getLog(getClass()); - - private final Chunk chunk; - - private static final int BACKSTOP_LIMIT = 1000; - - private int count = 0; - - public StatelessRetryChunkTests(String[] args) { - chunk = new Chunk(); - for (String string : args) { - chunk.add(string); - } - } - - @Test - public void testRetry() throws Exception { - - logger.debug("Starting simple scenario"); - List items = new ArrayList(chunk.getItems()); - int before = items.size(); - - items.removeAll(Collections.singleton("fail")); - - int errors = 0; - - boolean error = true; - while (error && count++ < BACKSTOP_LIMIT) { - try { - // success - logger.debug("Run items: " + chunk.getItems()); - retryChunk(chunk); - error = false; - } - catch (SpecialException e) { - error = true; - } - catch (Exception e) { - errors++; - error = true; - } - } - - logger.debug("Items: " + chunk.getItems()); - - assertTrue("Backstop reached. Probably an infinite loop...", count < BACKSTOP_LIMIT); - assertFalse(chunk.getItems().contains("fail")); - assertEquals(items, chunk.getItems()); - - int after = chunk.getItems().size(); - logger.debug(String.format("Error count: %d, size before: %d, size after: %d", errors, before, after)); - - } - - /** - * @param chunk - * @throws Exception - */ - private void retryChunk(Chunk chunk) throws Exception { - boolean complete = chunk.isComplete(); - try { - doWrite(chunk); - } - catch (Exception e) { - if (chunk.canSkip()) { - chunk.getSkippedItem(); - } - else { - if (complete) { - chunk.rethrow(e); - } else { - chunk.rethrow(new SpecialException()); - } - } - } - chunk.rethrow(); - } - - /** - * @param chunk - * @throws Exception - */ - private void doWrite(Chunk chunk) throws Exception { - List items = chunk.getItems(); - if (items.contains("fail")) { - throw new Exception(); - } - } - - @Parameters - public static List data() { - List params = new ArrayList(); - params.add(new Object[] { new String[] { "foo" } }); - params.add(new Object[] { new String[] { "foo", "bar" } }); - params.add(new Object[] { new String[] { "foo", "bar", "spam" } }); - params.add(new Object[] { new String[] { "foo", "bar", "spam", "maps", "rab", "oof" } }); - params.add(new Object[] { new String[] { "fail" } }); - params.add(new Object[] { new String[] { "foo", "fail" } }); - params.add(new Object[] { new String[] { "fail", "bar" } }); - params.add(new Object[] { new String[] { "foo", "fail", "spam" } }); - params.add(new Object[] { new String[] { "fail", "bar", "spam" } }); - params.add(new Object[] { new String[] { "foo", "fail", "spam", "maps", "rab", "oof" } }); - params.add(new Object[] { new String[] { "foo", "fail", "spam", "fail", "rab", "oof" } }); - params.add(new Object[] { new String[] { "fail", "bar", "spam", "fail", "rab", "oof" } }); - params.add(new Object[] { new String[] { "foo", "fail", "fail", "fail", "rab", "oof" } }); - params.add(new Object[] { new String[] { "fail" } }); - params.add(new Object[] { new String[] { "foo", "fail", "fail", "fail", "rab", "oof" } }); - params.add(new Object[] { new String[] { "foo", "fail", "fail", "fail", "rab", "oof" } }); - return params; - } - - /** - * @author Dave Syer - * - */ - public class SpecialException extends Exception { - - } - -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java index 04d9e4c7e..329b74a9a 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java @@ -96,9 +96,8 @@ public class SimpleRetryPolicy implements RetryPolicy { * attempts so far is less than the limit. */ public boolean canRetry(RetryContext context) { - SimpleRetryContext simpleContext = ((SimpleRetryContext) context); - Throwable t = simpleContext.getLastThrowable(); - return (t == null || retryForException(t)) && simpleContext.getRetryCount() < maxAttempts; + Throwable t = context.getLastThrowable(); + return (t == null || retryForException(t)) && context.getRetryCount() < maxAttempts; } /**