From af1729f2510159b9d9770dfddc12d24b350d615b Mon Sep 17 00:00:00 2001 From: lucasward Date: Mon, 11 Feb 2008 03:17:50 +0000 Subject: [PATCH] BATCH-220: Committed the read part of chunking, without adding it to the StepExecutor. It should be noted that I used 'Chunker' instead of 'ChunkReader', since there's already precedence in the java world, and it isn't reading chunk so much as making a chunk out of items. I have also added skipCount to StepExecution, since it's reasonable that this type of data A) should be persisted and B) the domain object is the most logical place for this type of state. --- .../simple/AlwaysSkipReadFailurePolicy.java | 33 ++++++ .../batch/execution/step/simple/Chunker.java | 41 +++++++ .../execution/step/simple/ItemChunker.java | 95 ++++++++++++++++ .../simple/SkipLimitExceededException.java | 39 +++++++ .../simple/SkipLimitReadFailurePolicy.java | 82 ++++++++++++++ .../step/simple/ItemChunkerTests.java | 101 ++++++++++++++++++ .../execution/step/simple/MockItemReader.java | 60 +++++++++++ .../SkipLimitReadFailurePolicyTests.java | 66 ++++++++++++ 8 files changed, 517 insertions(+) create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipReadFailurePolicy.java create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitExceededException.java create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicy.java create mode 100644 spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java create mode 100644 spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/MockItemReader.java create mode 100644 spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicyTests.java diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipReadFailurePolicy.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipReadFailurePolicy.java new file mode 100644 index 000000000..de35c6ea5 --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipReadFailurePolicy.java @@ -0,0 +1,33 @@ +/* + * Copyright 2006-2008 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.execution.step.simple; + +import org.springframework.batch.core.domain.ReadFailurePolicy; +import org.springframework.batch.core.domain.StepExecution; + +/** + * Implementation of the {@link ReadFailurePolicy} interface that + * will always return that reading should continue. + * + * @author Ben Hale + * @author Lucas Ward + */ +public class AlwaysSkipReadFailurePolicy implements ReadFailurePolicy { + + public boolean shouldContinue(Exception ex, StepExecution stepExecution) { + return true; + } +} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java new file mode 100644 index 000000000..45fc52acc --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java @@ -0,0 +1,41 @@ +/* + * Copyright 2006-2008 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.execution.step.simple; + +import org.springframework.batch.core.domain.Chunk; +import org.springframework.batch.io.exception.ReadFailureException; + + +/** + * Interface defining the contract for reading a chunk. This is most useful when + * implementing a 'chunk-oriented' approach to processing. Implementors of this + * class are expected to aggregate the output of an ItemReader into 'chunks'. + * + * @author Ben Hale + * @author Lucas Ward + */ +public interface Chunker { + + /** + * Read in a chunk, given the provided chunk size. + * + * @param chunkSize the number of items that should be read for this chunk. + * @return the {@link Chunk} that has been read. + * @throws IllegalArgumentException if chunkSize is less than zero. + */ + public Chunk read(int chunkSize) throws ReadFailureException; + +} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java new file mode 100644 index 000000000..374a4c8c9 --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java @@ -0,0 +1,95 @@ +/* + * 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.execution.step.simple; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.batch.core.domain.Chunk; +import org.springframework.batch.core.domain.ReadFailurePolicy; +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.io.exception.ReadFailureException; +import org.springframework.batch.item.ItemReader; +import org.springframework.util.Assert; + +/** + * Implementation of the {@link Chunker} interface that creates chunks from + * an {@link ItemReader} + * + * @author Ben Hale + * @author Lucas Ward + */ +public class ItemChunker implements Chunker { + + private final ItemReader itemReader; + private final StepExecution stepExecution; + + private long chunkCounter = 0; + + private ReadFailurePolicy readFailurePolicy = new AlwaysSkipReadFailurePolicy(); + + public ItemChunker(ItemReader itemReader, StepExecution stepExecution) { + this.itemReader = itemReader; + this.stepExecution = stepExecution; + } + + public void setReadFailurePolicy(ReadFailurePolicy readFailurePolicy) { + this.readFailurePolicy = readFailurePolicy; + } + + public Chunk read(int size) throws ReadFailureException { + Assert.isTrue(size > 0, "Chunk size must be greater than 0"); + + int counter = 0; + List items = new ArrayList(size); + + Object item; + while (counter < size) { + try { + item = itemReader.read(); + if (item == null) { + break; + } + items.add(item); + counter++; + } catch (Exception ex) { + if(!readFailurePolicy.shouldContinue(ex, stepExecution)){ + rethrow(ex); + } + } + } + + if (items.size() == 0) { + return null; + } + + return new Chunk(getChunkId(), items); + } + + private void rethrow(Exception ex){ + if(ex instanceof RuntimeException){ + throw (RuntimeException)ex; + } + else{ + throw new ReadFailureException("Error encountered while reading", ex); + } + } + + private synchronized Long getChunkId() { + return new Long(chunkCounter++); + } + +} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitExceededException.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitExceededException.java new file mode 100644 index 000000000..3e31a62ec --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitExceededException.java @@ -0,0 +1,39 @@ +/* + * 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.execution.step.simple; + +import org.springframework.batch.io.exception.BatchCriticalException; + +/** + * Exception indicating that the skip limit for a particular {@Step} has + * been exceeded. + * + * @author Ben Hale + * @author Lucas Ward + */ +public class SkipLimitExceededException extends BatchCriticalException { + + private final int skipLimit; + + public SkipLimitExceededException(int skipLimit, Exception ex) { + super("Skip limit of '" + skipLimit + "' exceeded", ex); + this.skipLimit = skipLimit; + } + + public int getSkipLimit() { + return skipLimit; + } +} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicy.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicy.java new file mode 100644 index 000000000..d6f1a4197 --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicy.java @@ -0,0 +1,82 @@ +/* + * 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.execution.step.simple; + +import java.io.FileNotFoundException; +import java.util.ArrayList; +import java.util.List; + +import org.springframework.batch.core.domain.ReadFailurePolicy; +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.io.exception.FlatFileParsingException; + +/** + *

{@link ReadFailurePolicy} that determines whether or not reading should + * continue based upon how many items have been skipped. This is extremely + * useful behavior, as it allows you to skip records, but will throw a + * {@link SkipLimitExceededException} if a set limit has been exceeded. For example, + * it is generally advisable to skip {@link FlatFileParsingException}s, however, if + * the vast majority of records are causing exceptions, the file is likely bad.

+ * + *

Furthermore, it is also likely that you only want to skip certain exceptions. + * {@link FlatFileParsingException} is a good example of an exception you will likely + * want to skip, but a {@link FileNotFoundException} should cause immediate termination + * of the {@link Step}. Because it would be impossible for a general purpose policy to + * determine all the types of exceptions that should be skipped from those that shouldn't, + * a list must be passed in, with all of the exceptions that are 'skippable'

+ * + * @author Ben Hale + * @author Lucas Ward + */ +public class SkipLimitReadFailurePolicy implements ReadFailurePolicy { + + private final int skipLimit; + + private final List skippableExceptions; + + public SkipLimitReadFailurePolicy(int skipLimit) { + this(skipLimit, new ArrayList(0)); + } + + public SkipLimitReadFailurePolicy(int skipLimit, List skippableExceptions) { + this.skipLimit = skipLimit; + this.skippableExceptions = skippableExceptions; + } + + /** + * Given the provided exception and StepExecution, determine whether or not + * processing should continue for the given exception. If the exception + * is not within the list of 'skippable exceptions', false will be returned. + * If the exception is within the list, and {@link StepExecution} skipCount + * is greater than the skipLimit, then a {@link SkipLimitExceededException} + * will be thrown. + */ + public boolean shouldContinue(Exception ex, StepExecution stepExecution){ + if(skippableExceptions.contains(ex.getClass())){ + if(stepExecution.getSkipCount() < skipLimit){ + stepExecution.incrementSkipCount(); + return true; + } + else{ + throw new SkipLimitExceededException(skipLimit, ex); + } + } + else{ + return false; + } + } + +} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java new file mode 100644 index 000000000..2fabab361 --- /dev/null +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java @@ -0,0 +1,101 @@ +/* + * 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.execution.step.simple; + +import junit.framework.TestCase; + +import org.springframework.batch.core.domain.Chunk; +import org.springframework.batch.core.domain.ReadFailurePolicy; +import org.springframework.batch.core.domain.StepExecution; + +public class ItemChunkerTests extends TestCase { + + StepExecution stepExecution; + + protected void setUp() throws Exception { + super.setUp(); + + stepExecution = new StepExecution(null,null); + } + + public void testSizeNegative() { + try { + MockItemReader itemReader = new MockItemReader(10); + ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); + chunkReader.read(-1); + fail(); + } catch (IllegalArgumentException e) { + } + } + + public void testSizeZero() { + try { + MockItemReader itemReader = new MockItemReader(10); + ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); + chunkReader.read(0); + fail(); + } catch (IllegalArgumentException e) { + } + } + + public void testSizePositive() { + MockItemReader itemReader = new MockItemReader(10); + ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); + Chunk chunk = chunkReader.read(10); + assertEquals(10, chunk.getItems().size()); + } + + public void testIncompleteChunk() { + MockItemReader itemReader = new MockItemReader(5); + ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); + Chunk chunk = chunkReader.read(10); + assertEquals(5, chunk.getItems().size()); + } + + public void testPolicyNoContinue() { + MockItemReader itemReader = new MockItemReader(1); + itemReader.setFail(true); + ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); + chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(true)); + try { + chunkReader.read(10); + fail(); + } catch (RuntimeException e) { + } + } + + public void testPolicyContinueWithFailure() { + MockItemReader itemReader = new MockItemReader(1); + itemReader.setFail(true); + ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); + chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(false)); + Chunk chunk = chunkReader.read(1); + assertEquals(1, chunk.getItems().size()); + } + + private class StubReadFailurePolicy implements ReadFailurePolicy { + + private final boolean fail; + + public StubReadFailurePolicy(boolean fail) { + this.fail = fail; + } + + public boolean shouldContinue(Exception ex, StepExecution stepExecution) { + return !fail; + } + } +} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/MockItemReader.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/MockItemReader.java new file mode 100644 index 000000000..6cfd958a1 --- /dev/null +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/MockItemReader.java @@ -0,0 +1,60 @@ +/* + * 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.execution.step.simple; + +import org.springframework.batch.item.ItemReader; + +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 Object read() { + if(fail) { + fail = false; + throw new RuntimeException(); + } + + if (returnItemCount < 0 || returnedItemCount < returnItemCount) { + return String.valueOf(returnedItemCount++); + } + return null; + } + + public Object getKey(Object item) { + // TODO Auto-generated method stub + return null; + } + +} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicyTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicyTests.java new file mode 100644 index 000000000..58482439c --- /dev/null +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicyTests.java @@ -0,0 +1,66 @@ +/* + * Copyright 2006-2008 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.execution.step.simple; + +import java.io.FileNotFoundException; +import java.util.ArrayList; +import java.util.List; + +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.io.exception.FlatFileParsingException; + +import junit.framework.TestCase; + +/** + * @author Lucas Ward + * + */ +public class SkipLimitReadFailurePolicyTests extends TestCase { + + SkipLimitReadFailurePolicy failurePolicy; + StepExecution stepExecution; + + protected void setUp() throws Exception { + super.setUp(); + + List skippableExceptions = new ArrayList(); + skippableExceptions.add(FlatFileParsingException.class); + + failurePolicy = new SkipLimitReadFailurePolicy(1, skippableExceptions); + stepExecution = new StepExecution(null, null); + stepExecution.setSkipCount(2); + } + + public void testLimitExceed(){ + try{ + failurePolicy.shouldContinue(new FlatFileParsingException("", ""), stepExecution); + fail(); + } + catch(SkipLimitExceededException ex){ + //expected + } + } + + public void testNonSkippableException(){ + assertFalse(failurePolicy.shouldContinue(new FileNotFoundException(), stepExecution)); + } + + public void testSkip(){ + stepExecution.setSkipCount(0); + assertTrue(failurePolicy.shouldContinue(new FlatFileParsingException("",""), stepExecution)); + } + +}