diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchRetryTemplate.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchRetryTemplate.java index 8179ca874..f9641a31a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchRetryTemplate.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchRetryTemplate.java @@ -196,6 +196,8 @@ public class BatchRetryTemplate implements RetryOperations { private final RetryTemplate regular = new RetryTemplate(); + private RetryPolicy retryPolicy; + public T execute(RetryCallback retryCallback, Collection states) throws ExhaustedRetryException, Exception { RetryState batchState = new BatchRetryState(states); @@ -263,8 +265,13 @@ public class BatchRetryTemplate implements RetryOperations { } public void setRetryPolicy(RetryPolicy retryPolicy) { + this.retryPolicy = retryPolicy; delegate.setRetryPolicy(retryPolicy); regular.setRetryPolicy(retryPolicy); } + + public boolean canRetry(RetryContext context) { + return context==null ? true : retryPolicy.canRetry(context); + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java index e03a270cf..fb7d2270a 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java @@ -21,6 +21,7 @@ import java.util.Collections; import java.util.Iterator; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -213,10 +214,12 @@ public class FaultTolerantChunkProcessor extends : new ArrayList(cache.getItems()).iterator(); final AtomicInteger count = new AtomicInteger(0); + // final int scanLimit = processorTransactional && data.scanning() ? 1 : + // 0; + for (final Chunk.ChunkIterator iterator = inputs.iterator(); iterator .hasNext();) { - final int scanLimit = processorTransactional ? 1 : 0; final I item = iterator.next(); RetryCallback retryCallback = new RetryCallback() { @@ -227,18 +230,11 @@ public class FaultTolerantChunkProcessor extends count.incrementAndGet(); O cached = (cacheIterator != null && cacheIterator .hasNext()) ? cacheIterator.next() : null; - if (cached != null && count.get() > scanLimit) { - /* - * If there is a cached chunk then we must be - * scanning for errors in the writer, in which case - * only the first one will be written, and for the - * rest we need to fill in the output from the - * cache. - */ + if (cached != null && !processorTransactional) { output = cached; } else { output = doProcess(item); - if (!processorTransactional) { + if (!processorTransactional && !data.scanning()) { cache.add(output); } } @@ -309,6 +305,17 @@ public class FaultTolerantChunkProcessor extends outputs.add(output); } + /* + * We only want to process the first item if there is a scan for a + * failed item. + */ + if (data.scanning()) { + while (cacheIterator != null && cacheIterator.hasNext()) { + outputs.add(cacheIterator.next()); + } + // Only process the first item if scanning + break; + } } return outputs; @@ -319,10 +326,16 @@ public class FaultTolerantChunkProcessor extends protected void write(final StepContribution contribution, final Chunk inputs, final Chunk outputs) throws Exception { + @SuppressWarnings("unchecked") + final UserData data = (UserData) inputs.getUserData(); + final AtomicReference contextHolder = new AtomicReference(); + RetryCallback retryCallback = new RetryCallback() { public Object doWithRetry(RetryContext context) throws Exception { - if (!inputs.isBusy()) { + contextHolder.set(context); + + if (!data.scanning()) { chunkMonitor.setChunkSize(inputs.size()); try { doWrite(outputs.getItems()); @@ -409,6 +422,7 @@ public class FaultTolerantChunkProcessor extends } inputs.setBusy(true); + data.scanning(true); scan(contribution, inputs, outputs, chunkMonitor, true); return null; } @@ -418,8 +432,21 @@ public class FaultTolerantChunkProcessor extends if (logger.isDebugEnabled()) { logger.debug("Attempting to write: " + inputs); } - batchRetryTemplate.execute(retryCallback, recoveryCallback, - new DefaultRetryState(inputs, rollbackClassifier)); + try { + batchRetryTemplate.execute(retryCallback, recoveryCallback, + new DefaultRetryState(inputs, rollbackClassifier)); + } catch (Exception e) { + RetryContext context = contextHolder.get(); + if (!batchRetryTemplate.canRetry(context)) { + /* + * BATCH-1761: we need advanvce warning of the scan about to + * start in the next transaction, so we can change the + * processing behaviour. + */ + data.scanning(true); + } + throw e; + } } @@ -543,6 +570,9 @@ public class FaultTolerantChunkProcessor extends final Chunk inputs, final Chunk outputs, ChunkMonitor chunkMonitor, boolean recovery) throws Exception { + @SuppressWarnings("unchecked") + final UserData data = (UserData) inputs.getUserData(); + if (logger.isDebugEnabled()) { if (recovery) { logger.debug("Scanning for failed item on recovery from write: " @@ -552,6 +582,7 @@ public class FaultTolerantChunkProcessor extends } } if (outputs.isEmpty()) { + data.scanning(false); inputs.setBusy(false); return; } @@ -585,6 +616,7 @@ public class FaultTolerantChunkProcessor extends } chunkMonitor.incrementOffset(); if (outputs.isEmpty()) { + data.scanning(false); inputs.setBusy(false); chunkMonitor.resetOffset(); } @@ -596,6 +628,16 @@ public class FaultTolerantChunkProcessor extends private int filterCount = 0; + private boolean scanning; + + public boolean scanning() { + return scanning; + } + + public void scanning(boolean scanning) { + this.scanning = scanning; + } + public void incrementFilterCount() { filterCount++; } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java index 486168c44..82159d78b 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java @@ -75,13 +75,15 @@ public class FaultTolerantStepFactoryBeanRetryTests { private List provided = new ArrayList(); - private List written = TransactionAwareProxyFactory.createTransactionalList(); + private List written = TransactionAwareProxyFactory + .createTransactionalList(); int count = 0; boolean fail = false; - private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), + private SimpleJobRepository repository = new SimpleJobRepository( + new MapJobInstanceDao(), new MapJobExecutionDao(), new MapStepExecutionDao(), new MapExecutionContextDao()); JobExecution jobExecution; @@ -99,7 +101,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory = new FaultTolerantStepFactoryBean(); factory.setBeanName("step"); - factory.setItemReader(new ListItemReader(new ArrayList())); + factory.setItemReader(new ListItemReader( + new ArrayList())); factory.setItemWriter(writer); factory.setJobRepository(repository); factory.setTransactionManager(new ResourcelessTransactionManager()); @@ -108,8 +111,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setSkippableExceptionClasses(getExceptionMap(Exception.class)); - JobParameters jobParameters = new JobParametersBuilder().addString("statefulTest", "make_this_unique") - .toJobParameters(); + JobParameters jobParameters = new JobParametersBuilder().addString( + "statefulTest", "make_this_unique").toJobParameters(); jobExecution = repository.createJobExecution("job", jobParameters); jobExecution.setEndTime(new Date()); @@ -126,11 +129,13 @@ public class FaultTolerantStepFactoryBeanRetryTests { } @Test - public void testProcessAllItemsWhenErrorInWriterTransformation() throws Exception{ + public void testProcessAllItemsWhenErrorInWriterTransformation() + throws Exception { FaultTolerantStepFactoryBean factory = new FaultTolerantStepFactoryBean(); factory.setBeanName("step"); - factory.setItemReader(new ListItemReader(new ArrayList())); + factory.setItemReader(new ListItemReader( + new ArrayList())); factory.setJobRepository(repository); factory.setTransactionManager(new ResourcelessTransactionManager()); @SuppressWarnings("unchecked") @@ -145,37 +150,40 @@ public class FaultTolerantStepFactoryBeanRetryTests { } written.add(item); } - }}; - - ItemProcessor processor = new ItemProcessor() { - public Integer process(String item) throws Exception { - processed.add(item); - return Integer.parseInt(item); - } - }; - ItemReader reader = new ListItemReader(Arrays.asList("1", "2", "3")); - factory.setCommitInterval(3); - factory.setRetryLimit(3); - factory.setSkippableExceptionClasses(new HashMap, Boolean>()); - factory.setItemReader(reader); - factory.setItemProcessor(processor); - factory.setItemWriter(failingWriter); - Step step = (Step) factory.getObject(); - - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); - repository.add(stepExecution); - step.execute(stepExecution); -// System.out.println(stepExecution.getWriteCount()); -// System.out.println(processed.size()); -// System.out.println(processed); -// System.out.println(written); - assertEquals((1+3)*3, processed.size()); //(Initial try + retry limit)*item count + } + }; + + ItemProcessor processor = new ItemProcessor() { + public Integer process(String item) throws Exception { + processed.add(item); + return Integer.parseInt(item); + } + }; + ItemReader reader = new ListItemReader(Arrays.asList( + "1", "2", "3")); + factory.setCommitInterval(3); + factory.setRetryLimit(3); + factory.setSkippableExceptionClasses(new HashMap, Boolean>()); + factory.setItemReader(reader); + factory.setItemProcessor(processor); + factory.setItemWriter(failingWriter); + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); + repository.add(stepExecution); + step.execute(stepExecution); + System.out.println(stepExecution.getWriteCount()); + System.out.println(processed.size()); + System.out.println(processed); + System.out.println(written); + assertEquals((1 + 3) * 3, processed.size()); // (Initial try + retry limit)*item count } @Test public void testProcessAllItemsWhenErrorInWriter() throws Exception { final int RETRY_LIMIT = 3; - final List ITEM_LIST = Arrays.asList("a", " b", "c"); + final List ITEM_LIST = Arrays.asList("a", "b", "c"); ItemWriter failingWriter = new ItemWriter() { public void write(List data) throws Exception { int count = 0; @@ -206,16 +214,25 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setItemWriter(failingWriter); Step step = (Step) factory.getObject(); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); -// System.out.println(processed); - assertEquals(ExitStatus.COMPLETED.getExitCode(), stepExecution.getExitStatus().getExitCode()); - assertEquals((1 + RETRY_LIMIT) * ITEM_LIST.size(), processed.size()); + System.out.println(processed); + assertEquals(ExitStatus.COMPLETED.getExitCode(), stepExecution + .getExitStatus().getExitCode()); + /* + * Each chunk tried up to RETRY_LIMIT, then the scan processes 1 full + * chunk and fails on the scan, identfies the skip, and then + * re-processes the other n-1 items + */ + assertEquals(RETRY_LIMIT * ITEM_LIST.size() + ITEM_LIST.size() + + ITEM_LIST.size() - 1, processed.size()); } @Test - public void testNoItemsReprocessedWhenErrorInWriterAndProcessorNotTransactional() throws Exception{ + public void testNoItemsReprocessedWhenErrorInWriterAndProcessorNotTransactional() + throws Exception { ItemWriter failingWriter = new ItemWriter() { public void write(List data) throws Exception { int count = 0; @@ -225,34 +242,36 @@ public class FaultTolerantStepFactoryBeanRetryTests { } written.add(item); } - }}; - - ItemProcessor processor = new ItemProcessor() { - public String process(String item) throws Exception { - processed.add(item); - return item; - } - }; - ItemReader reader = new ListItemReader(Arrays.asList("a", "b", "c")); - factory.setProcessorTransactional(false); - factory.setCommitInterval(3); - factory.setRetryLimit(3); - factory.setSkippableExceptionClasses(new HashMap, Boolean>()); - factory.setItemReader(reader); - factory.setItemProcessor(processor); - factory.setItemWriter(failingWriter); - Step step = (Step) factory.getObject(); - - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); - repository.add(stepExecution); - step.execute(stepExecution); - System.out.println(stepExecution.getWriteCount()); - System.out.println(processed.size()); - System.out.println(processed); - System.out.println(written); - assertEquals(3, processed.size()); //(Initial try + retry limit)*item count + } + }; + + ItemProcessor processor = new ItemProcessor() { + public String process(String item) throws Exception { + processed.add(item); + return item; + } + }; + ItemReader reader = new ListItemReader(Arrays.asList( + "a", "b", "c")); + factory.setProcessorTransactional(false); + factory.setCommitInterval(3); + factory.setRetryLimit(3); + factory.setSkippableExceptionClasses(new HashMap, Boolean>()); + factory.setItemReader(reader); + factory.setItemProcessor(processor); + factory.setItemWriter(failingWriter); + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); + repository.add(stepExecution); + step.execute(stepExecution); + // System.out.println(stepExecution.getWriteCount()); + // System.out.println(processed.size()); + // System.out.println(processed); + // System.out.println(written); + assertEquals(3, processed.size()); // Initial try only, then cached } - /** * N.B. this doesn't really test retry, since the retry is only on write @@ -265,13 +284,15 @@ public class FaultTolerantStepFactoryBeanRetryTests { @SuppressWarnings("unchecked") @Test public void testSuccessfulRetryWithReadFailure() throws Exception { - ItemReader provider = new ListItemReader(Arrays.asList("a", "b", "c")) { + ItemReader provider = new ListItemReader(Arrays.asList( + "a", "b", "c")) { public String read() { String item = super.read(); provided.add(item); count++; if (count == 2) { - throw new RuntimeException("Temporary error - retry for success."); + throw new RuntimeException( + "Temporary error - retry for success."); } return item; } @@ -281,7 +302,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setSkippableExceptionClasses(getExceptionMap()); Step step = (Step) factory.getObject(); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); @@ -313,7 +335,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { @Override protected void doOpen() throws Exception { - reader = new ListItemReader(Arrays.asList("a", "b", "c", "d", "e", "f")); + reader = new ListItemReader(Arrays.asList("a", "b", + "c", "d", "e", "f")); } @Override @@ -338,7 +361,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { Step step = (Step) factory.getObject(); fail = true; - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); @@ -362,12 +386,14 @@ public class FaultTolerantStepFactoryBeanRetryTests { public void testSkipAndRetry() throws Exception { factory.setSkipLimit(2); - ItemReader provider = new ListItemReader(Arrays.asList("a", "b", "c", "d", "e", "f")) { + ItemReader provider = new ListItemReader(Arrays.asList( + "a", "b", "c", "d", "e", "f")) { public String read() { String item = super.read(); count++; if ("b".equals(item) || "d".equals(item)) { - throw new RuntimeException("Read error - planned but skippable."); + throw new RuntimeException( + "Read error - planned but skippable."); } return item; } @@ -376,7 +402,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setRetryLimit(10); Step step = (Step) factory.getObject(); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); @@ -393,11 +420,13 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setListeners(new StepListener[] { new SkipListenerSupport() { public void onSkipInWrite(String item, Throwable t) { recovered.add(item); - assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); + assertTrue(TransactionSynchronizationManager + .isActualTransactionActive()); } } }); factory.setSkipLimit(2); - ItemReader provider = new ListItemReader(Arrays.asList("a", "b", "c", "d", "e", "f")) { + 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 + "]"); @@ -413,7 +442,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { processed.addAll(item); written.addAll(item); if (item.contains("b") || item.contains("d")) { - throw new RuntimeException("Write error - planned but recoverable."); + throw new RuntimeException( + "Write error - planned but recoverable."); } } }; @@ -423,7 +453,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setRetryableExceptionClasses(getExceptionMap(RuntimeException.class)); AbstractStep step = (AbstractStep) factory.getObject(); step.setName("mytest"); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); @@ -431,27 +462,32 @@ public class FaultTolerantStepFactoryBeanRetryTests { assertEquals(2, stepExecution.getSkipCount()); assertEquals(2, stepExecution.getWriteSkipCount()); - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,c,e,f")); + List expectedOutput = Arrays.asList(StringUtils + .commaDelimitedListToStringArray("a,c,e,f")); assertEquals(expectedOutput, written); assertEquals("[a, b, c, d, e, f, null]", provided.toString()); - assertEquals("[a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f]", processed.toString()); + assertEquals("[a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f]", + processed.toString()); assertEquals("[b, d]", recovered.toString()); } @SuppressWarnings("unchecked") @Test - public void testSkipAndRetryWithWriteFailureAndNonTrivialCommitInterval() throws Exception { + public void testSkipAndRetryWithWriteFailureAndNonTrivialCommitInterval() + throws Exception { factory.setCommitInterval(3); factory.setListeners(new StepListener[] { new SkipListenerSupport() { public void onSkipInWrite(String item, Throwable t) { recovered.add(item); - assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); + assertTrue(TransactionSynchronizationManager + .isActualTransactionActive()); } } }); factory.setSkipLimit(2); - ItemReader provider = new ListItemReader(Arrays.asList("a", "b", "c", "d", "e", "f")) { + 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 + "]"); @@ -467,7 +503,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { processed.addAll(item); written.addAll(item); if (item.contains("b") || item.contains("d")) { - throw new RuntimeException("Write error - planned but recoverable."); + throw new RuntimeException( + "Write error - planned but recoverable."); } } }; @@ -477,7 +514,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setRetryableExceptionClasses(getExceptionMap(RuntimeException.class)); AbstractStep step = (AbstractStep) factory.getObject(); step.setName("mytest"); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); @@ -485,7 +523,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { assertEquals(2, stepExecution.getSkipCount()); assertEquals(2, stepExecution.getWriteSkipCount()); - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,c,e,f")); + List expectedOutput = Arrays.asList(StringUtils + .commaDelimitedListToStringArray("a,c,e,f")); assertEquals(expectedOutput, written); // [a, b, c, d, e, f, null] @@ -503,7 +542,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setRetryLimit(4); factory.setSkipLimit(0); - ItemReader provider = new ListItemReader(Arrays.asList("b")) { + ItemReader provider = new ListItemReader( + Arrays.asList("b")) { public String read() { String item = super.read(); provided.add(item); @@ -516,19 +556,22 @@ public class FaultTolerantStepFactoryBeanRetryTests { processed.addAll(item); written.addAll(item); logger.debug("Write Called! Item: [" + item + "]"); - throw new RuntimeException("Write error - planned but retryable."); + throw new RuntimeException( + "Write error - planned but retryable."); } }; factory.setItemReader(provider); factory.setItemWriter(itemWriter); Step step = (Step) factory.getObject(); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("")); + List expectedOutput = Arrays.asList(StringUtils + .commaDelimitedListToStringArray("")); assertEquals(expectedOutput, written); assertEquals(0, stepExecution.getSkipCount()); @@ -552,7 +595,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setRetryableExceptionClasses(getExceptionMap()); factory.setSkipLimit(1); - ItemReader provider = new ListItemReader(Arrays.asList("b")) { + ItemReader provider = new ListItemReader( + Arrays.asList("b")) { public String read() { String item = super.read(); provided.add(item); @@ -565,20 +609,25 @@ public class FaultTolerantStepFactoryBeanRetryTests { processed.addAll(item); written.addAll(item); logger.debug("Write Called! Item: [" + item + "]"); - throw new RuntimeException("Write error - planned but not skippable."); + throw new RuntimeException( + "Write error - planned but not skippable."); } }; factory.setItemReader(provider); factory.setItemWriter(itemWriter); Step step = (Step) factory.getObject(); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); - String message = stepExecution.getFailureExceptions().get(0).getMessage(); - assertTrue("Wrong message: " + message, message.contains("Write error - planned but not skippable.")); + String message = stepExecution.getFailureExceptions().get(0) + .getMessage(); + assertTrue("Wrong message: " + message, + message.contains("Write error - planned but not skippable.")); - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("")); + List expectedOutput = Arrays.asList(StringUtils + .commaDelimitedListToStringArray("")); assertEquals(expectedOutput, written); assertEquals(0, stepExecution.getSkipCount()); @@ -594,9 +643,11 @@ public class FaultTolerantStepFactoryBeanRetryTests { @Test public void testRetryPolicy() throws Exception { factory.setRetryPolicy(new SimpleRetryPolicy(4, Collections - ., Boolean> singletonMap(Exception.class, true))); + ., Boolean> singletonMap( + Exception.class, true))); factory.setSkipLimit(0); - ItemReader provider = new ListItemReader(Arrays.asList("b")) { + ItemReader provider = new ListItemReader( + Arrays.asList("b")) { public String read() { String item = super.read(); provided.add(item); @@ -609,19 +660,22 @@ public class FaultTolerantStepFactoryBeanRetryTests { processed.addAll(item); written.addAll(item); logger.debug("Write Called! Item: [" + item + "]"); - throw new RuntimeException("Write error - planned but retryable."); + throw new RuntimeException( + "Write error - planned but retryable."); } }; factory.setItemReader(provider); factory.setItemWriter(itemWriter); AbstractStep step = (AbstractStep) factory.getObject(); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("")); + List expectedOutput = Arrays.asList(StringUtils + .commaDelimitedListToStringArray("")); assertEquals(expectedOutput, written); assertEquals(0, stepExecution.getSkipCount()); @@ -657,14 +711,16 @@ public class FaultTolerantStepFactoryBeanRetryTests { public void write(List item) throws Exception { processed.addAll(item); logger.debug("Write Called! Item: [" + item + "]"); - throw new RuntimeException("Write error - planned but retryable."); + throw new RuntimeException( + "Write error - planned but retryable."); } }; factory.setItemReader(provider); factory.setItemWriter(itemWriter); AbstractStep step = (AbstractStep) factory.getObject(); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + StepExecution stepExecution = new StepExecution(step.getName(), + jobExecution); repository.add(stepExecution); step.execute(stepExecution); assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); @@ -680,7 +736,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { assertEquals(0, recovered.size()); } - private Map, Boolean> getExceptionMap(Class... args) { + private Map, Boolean> getExceptionMap( + Class... args) { Map, Boolean> map = new HashMap, Boolean>(); for (Class arg : args) { map.put(arg, true); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java index 707114515..a94e23fe6 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java @@ -434,6 +434,7 @@ public class FaultTolerantStepFactoryBeanRollbackTests { assertEquals("[1, 3, 5]", writer.getWritten().toString()); assertEquals("[1, 3, 5]", writer.getCommitted().toString()); + // If non-transactional, we should only process each item once assertEquals("[1, 2, 3, 4, 5]", processor.getProcessed().toString()); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java index 8016b05f0..96b308e10 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java @@ -717,9 +717,9 @@ public class FaultTolerantStepFactoryBeanTests { assertEquals(1, stepExecution.getSkipCount()); assertEquals(2, stepExecution.getRollbackCount()); - // 1,2,3,4,3,4 - two re-processing attempts until the item is + // 1,2,3,4,3,4,4 - two re-processing attempts until the item is // identified and finally skipped on the second attempt - assertEquals("[1, 2, 3, 4, 3, 4]", processor.getProcessed().toString()); + assertEquals("[1, 2, 3, 4, 3, 4, 4]", processor.getProcessed().toString()); assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step .getName())); }