diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java index 2425d2611..7e6af16e6 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java @@ -21,8 +21,10 @@ import org.springframework.batch.classify.Classifier; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NonSkippableReadException; +import org.springframework.batch.core.step.skip.SkipException; import org.springframework.batch.core.step.skip.SkipListenerFailedException; import org.springframework.batch.core.step.skip.SkipPolicy; +import org.springframework.batch.core.step.skip.SkipPolicyFailedException; import org.springframework.batch.item.ItemReader; import org.springframework.batch.repeat.RepeatOperations; @@ -34,13 +36,30 @@ import org.springframework.batch.repeat.RepeatOperations; */ public class FaultTolerantChunkProvider extends SimpleChunkProvider { + /** + * Hard limit for number of read skips in the same chunk. Should be + * sufficiently high that it is only encountered in a runaway step where all + * items are skipped before the chunk can complete (leading to a potential + * heap memory problem). + */ + public static final int DEFAULT_MAX_SKIPS_ON_READ = 100; + private SkipPolicy skipPolicy = new LimitCheckingItemSkipPolicy(); private Classifier rollbackClassifier = new BinaryExceptionClassifier(true); + private int maxSkipsOnRead = DEFAULT_MAX_SKIPS_ON_READ; + public FaultTolerantChunkProvider(ItemReader itemReader, RepeatOperations repeatOperations) { super(itemReader, repeatOperations); } + + /** + * @param maxSkipsOnRead the maximum number of skips on read + */ + public void setMaxSkipsOnRead(int maxSkipsOnRead) { + this.maxSkipsOnRead = maxSkipsOnRead; + } /** * The policy that determines whether exceptions can be skipped on read. @@ -74,6 +93,10 @@ public class FaultTolerantChunkProvider extends SimpleChunkProvider { contribution.incrementReadSkipCount(); chunk.skip(e); + if (chunk.getErrors().size() >= maxSkipsOnRead) { + throw new SkipOverflowException("Too many skips on read"); + } + logger.debug("Skipping failed input", e); } else { @@ -110,8 +133,11 @@ public class FaultTolerantChunkProvider extends SimpleChunkProvider { try { return policy.shouldSkip(e, skipCount); } + catch (SkipException ex) { + throw ex; + } catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in SkipPolicy.", ex, e); + throw new SkipPolicyFailedException("Fatal exception in SkipPolicy.", ex, e); } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java index 4b0c54bd8..0b8e4bf0e 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java @@ -386,6 +386,7 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean chunkProvider = new FaultTolerantChunkProvider(getItemReader(), getChunkOperations()); + chunkProvider.setMaxSkipsOnRead(Math.max(getCommitInterval(), FaultTolerantChunkProvider.DEFAULT_MAX_SKIPS_ON_READ)); chunkProvider.setSkipPolicy(readSkipPolicy); chunkProvider.setRollbackClassifier(getRollbackClassifier()); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java index d3b18a87c..7a92205f6 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java @@ -30,8 +30,8 @@ import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.RepeatStatus; /** - * Simple implementation of the ChunkProvider interface that does basic - * chunk providing from an {@link ItemReader}. + * Simple implementation of the ChunkProvider interface that does basic chunk + * providing from an {@link ItemReader}. * * @author Dave Syer * @see ChunkOrientedTasklet @@ -71,7 +71,7 @@ public class SimpleChunkProvider implements ChunkProvider { public void registerListener(StepListener listener) { this.listener.register(listener); } - + /** * @return the listener */ @@ -103,7 +103,15 @@ public class SimpleChunkProvider implements ChunkProvider { repeatOperations.iterate(new RepeatCallback() { public RepeatStatus doInIteration(final RepeatContext context) throws Exception { - I item = read(contribution, inputs); + I item = null; + try { + item = read(contribution, inputs); + } + catch (SkipOverflowException e) { + // read() tells us about an excess of skips by throwing an + // exception + return RepeatStatus.FINISHED; + } if (item == null) { inputs.setEnd(); return RepeatStatus.FINISHED; @@ -118,12 +126,25 @@ public class SimpleChunkProvider implements ChunkProvider { return inputs; } - + public void postProcess(StepContribution contribution, Chunk chunk) { // do nothing } - protected I read(StepContribution contribution, Chunk chunk) throws Exception { + /** + * Delegates to {@link #doRead()}. Subclasses can add additional behaviour + * (e.g. exception handling). + * + * @param contribution the current step execution contribution + * @param chunk the current chunk + * @return a new item for processing + * + * @throws SkipOverflowException if specifically the chunk is accumulating + * too much data (e.g. skips) and it wants to force a commit. + * + * @throws Exception if there is a generic issue + */ + protected I read(StepContribution contribution, Chunk chunk) throws SkipOverflowException, Exception { return doRead(); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java index caaa04847..709edd04c 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java @@ -366,6 +366,15 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { public void setCommitInterval(int commitInterval) { this.commitInterval = commitInterval; } + + /** + * Accessor for commit interval if needed in sub classes. + * + * @return the commit interval + */ + protected int getCommitInterval() { + return commitInterval; + } /** * Public setter for the {@link CompletionPolicy} applying to the chunk diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipOverflowException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipOverflowException.java new file mode 100755 index 000000000..f818c47ab --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipOverflowException.java @@ -0,0 +1,34 @@ +/* + * Copyright 2006-2009 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.skip.SkipException; + +/** + * @author Dave Syer + * + */ +public class SkipOverflowException extends SkipException { + + /** + * @param msg the message for the user + */ + public SkipOverflowException(String msg) { + super(msg); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index cd3bf823d..7f3ae6acb 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -15,7 +15,6 @@ */ package org.springframework.batch.core.step.tasklet; -import java.lang.reflect.Field; import java.util.concurrent.Semaphore; import org.apache.commons.logging.Log; @@ -53,7 +52,6 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; -import org.springframework.util.ReflectionUtils; /** * Simple implementation of executing the step as a call to a {@link Tasklet}, @@ -456,7 +454,6 @@ public class TaskletStep extends AbstractStep { target.setRollbackCount(source.getRollbackCount()); } - } /** diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProviderTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProviderTests.java new file mode 100755 index 000000000..5c2f30e47 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProviderTests.java @@ -0,0 +1,53 @@ +package org.springframework.batch.core.step.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.Arrays; +import java.util.Collections; + +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.repeat.support.RepeatTemplate; + +public class FaultTolerantChunkProviderTests { + + private FaultTolerantChunkProvider provider; + + private StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution( + new JobInstance(123L, new JobParameters(), "job")))); + + @Test + public void testProvide() throws Exception { + provider = new FaultTolerantChunkProvider(new ListItemReader(Arrays.asList("foo", "bar")), + new RepeatTemplate()); + Chunk chunk = provider.provide(contribution); + assertNotNull(chunk); + assertEquals(2, chunk.getItems().size()); + } + + @Test + public void testProvideWithOverflow() throws Exception { + provider = new FaultTolerantChunkProvider(new ItemReader() { + public String read() throws Exception, UnexpectedInputException, ParseException { + throw new RuntimeException("Planned"); + } + }, new RepeatTemplate()); + provider.setSkipPolicy(new LimitCheckingItemSkipPolicy(Integer.MAX_VALUE, Collections.,Boolean>singletonMap(Exception.class, Boolean.TRUE))); + provider.setMaxSkipsOnRead(10); + Chunk chunk = null; + chunk = provider.provide(contribution); + assertNotNull(chunk); + assertEquals(0, chunk.getItems().size()); + assertEquals(10, chunk.getErrors().size()); + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProviderTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProviderTests.java index de2fa3715..3b2aaa026 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProviderTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProviderTests.java @@ -5,7 +5,6 @@ import static org.junit.Assert.assertNotNull; import java.util.Arrays; -import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; @@ -22,17 +21,31 @@ public class SimpleChunkProviderTests { private StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution( new JobInstance(123L, new JobParameters(), "job")))); - @Before - public void setUp() { - provider = new SimpleChunkProvider(new ListItemReader(Arrays.asList("foo", "bar")), - new RepeatTemplate()); - } - @Test public void testProvide() throws Exception { + provider = new SimpleChunkProvider(new ListItemReader(Arrays.asList("foo", "bar")), + new RepeatTemplate()); Chunk chunk = provider.provide(contribution); assertNotNull(chunk); assertEquals(2, chunk.getItems().size()); } + @Test + public void testProvideWithOverflow() throws Exception { + provider = new SimpleChunkProvider(new ListItemReader(Arrays.asList("foo", "bar")), + new RepeatTemplate()) { + @Override + protected String read(StepContribution contribution, Chunk chunk) throws SkipOverflowException, + Exception { + chunk.skip(new RuntimeException("Planned")); + throw new SkipOverflowException("Overflow"); + } + }; + Chunk chunk = null; + chunk = provider.provide(contribution); + assertNotNull(chunk); + assertEquals(0, chunk.getItems().size()); + assertEquals(1, chunk.getErrors().size()); + } + }