diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java index f496ebfb7..4daa84edd 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java @@ -21,8 +21,10 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.domain.BatchStatus; import org.springframework.batch.core.domain.Chunk; -import org.springframework.batch.core.domain.ChunkResult; +import org.springframework.batch.core.domain.ChunkingResult; import org.springframework.batch.core.domain.Dechunker; +import org.springframework.batch.core.domain.DechunkingResult; +import org.springframework.batch.core.domain.ItemFailureLog; import org.springframework.batch.core.domain.JobInterruptedException; import org.springframework.batch.core.domain.StepContribution; import org.springframework.batch.core.domain.StepExecution; @@ -81,6 +83,8 @@ public class ChunkedStep extends AbstractStep { // default to checking current thread for interruption. private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy(); + + private ItemFailureLog failureLog = new DefaultItemFailureLog(); private StreamManager streamManager; @@ -107,6 +111,10 @@ public class ChunkedStep extends AbstractStep { public void setStreamManager(StreamManager streamManager) { this.streamManager = streamManager; } + + public void setFailureLog(ItemFailureLog failureLog) { + this.failureLog = failureLog; + } /** * Injected strategy for storage and retrieval of persistent step information. Mandatory property. @@ -222,10 +230,14 @@ public class ChunkedStep extends AbstractStep { //shouldn't have to create a chunker each time, I'll refactor the interface later Chunker chunker = new ItemChunker(itemReader, stepExecution); - final Chunk chunk = chunker.chunk(chunkSize); - if(chunk == null){ + ChunkingResult chunkingResult = chunker.chunk(chunkSize); + + if(chunkingResult == null){ return ExitStatus.FINISHED; } + + final Chunk chunk = chunkingResult.getChunk(); + failureLog.log(chunkingResult.getExceptions()); retryTemplate.execute(new RetryCallback(){ @@ -292,7 +304,8 @@ public class ChunkedStep extends AbstractStep { Dechunker dechunker = new ItemDechunker(itemWriter, stepExecution); - ChunkResult chunkResult = dechunker.dechunk(chunk); + DechunkingResult chunkResult = dechunker.dechunk(chunk); + failureLog.log(chunkResult.getExceptions()); // TODO: check that stepExecution can // aggregate these contributions if they 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 index b2c3aed86..971dd32b3 100644 --- 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 @@ -16,6 +16,7 @@ package org.springframework.batch.execution.step.simple; import org.springframework.batch.core.domain.Chunk; +import org.springframework.batch.core.domain.ChunkingResult; import org.springframework.batch.io.exception.ReadFailureException; @@ -36,6 +37,6 @@ public interface Chunker { * @return the {@link Chunk} that has been read. * @throws IllegalArgumentException if chunkSize is less than zero. */ - public Chunk chunk(int chunkSize) throws ReadFailureException; + public ChunkingResult chunk(int chunkSize) throws ReadFailureException; } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/DefaultItemFailureLog.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/DefaultItemFailureLog.java new file mode 100644 index 000000000..ed1a9f240 --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/DefaultItemFailureLog.java @@ -0,0 +1,56 @@ +/* + * 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.util.Iterator; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.domain.ItemFailureLog; + +/** + * Default implementation of the {@link ItemFailureLog} interface that + * writes all exceptions via commons logging. Since generics can't be + * used to ensure the list contains exceptions, any non exceptions will + * be logged out by calling toString on the object. + * + * @author Lucas Ward + * + */ +public class DefaultItemFailureLog implements ItemFailureLog { + + protected static final Log logger = LogFactory + .getLog(DefaultItemFailureLog.class); + + /* (non-Javadoc) + * @see org.springframework.batch.core.domain.ItemFailureLog#log(java.util.List) + */ + public void log(List exceptions) { + for(Iterator it = exceptions.iterator(); it.hasNext();){ + Object exception = it.next(); + try{ + Throwable t = (Throwable)exception; + logger.error("Error encountered during processing", t); + } + catch(Exception ex){ + logger.error("Invalid type for logging: [" + exception.toString() + "]"); + } + + } + } + +} 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 index 113cde6ee..419750a11 100644 --- 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 @@ -19,6 +19,7 @@ import java.util.ArrayList; import java.util.List; import org.springframework.batch.core.domain.Chunk; +import org.springframework.batch.core.domain.ChunkingResult; import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.ReadFailureException; @@ -50,11 +51,12 @@ public class ItemChunker implements Chunker { this.readFailurePolicy = readFailurePolicy; } - public Chunk chunk(int size) throws ReadFailureException { + public ChunkingResult chunk(int size) throws ReadFailureException { Assert.isTrue(size > 0, "Chunk size must be greater than 0"); int counter = 0; List items = new ArrayList(size); + List exceptions = new ArrayList(); Object item; while (counter < size) { @@ -66,6 +68,7 @@ public class ItemChunker implements Chunker { items.add(item); counter++; } catch (Exception ex) { + exceptions.add(ex); if(!readFailurePolicy.shouldSkip(ex, stepExecution)){ rethrow(ex); } @@ -76,7 +79,7 @@ public class ItemChunker implements Chunker { return null; } - return new Chunk(getChunkId(), items); + return new ChunkingResult(new Chunk(getChunkId(), items), exceptions); } private void rethrow(Exception ex){ diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java index c8edbd060..41d9f7f00 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java @@ -20,10 +20,11 @@ import java.util.Iterator; import java.util.List; import org.springframework.batch.core.domain.Chunk; -import org.springframework.batch.core.domain.ChunkResult; +import org.springframework.batch.core.domain.DechunkingResult; import org.springframework.batch.core.domain.Dechunker; import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.io.exception.WriteFailureException; import org.springframework.batch.item.ItemWriter; import org.springframework.util.Assert; @@ -48,7 +49,7 @@ public class ItemDechunker implements Dechunker { /* (non-Javadoc) * @see org.springframework.batch.core.domain.Dechunker#dechunk(org.springframework.batch.core.domain.Chunk) */ - public ChunkResult dechunk(Chunk chunk) throws Exception { + public DechunkingResult dechunk(Chunk chunk) throws Exception { Assert.notNull(chunk, "Chunk must not be null"); List skippedItems = new ArrayList(); @@ -61,7 +62,7 @@ public class ItemDechunker implements Dechunker { catch(Exception ex){ if(itemSkipPolicy.shouldSkip(ex, stepExecution)){ stepExecution.incrementSkipCount(); - skippedItems.add(item); + skippedItems.add(new WriteFailureException(ex, item)); } else{ rethrow(ex); @@ -69,7 +70,7 @@ public class ItemDechunker implements Dechunker { } } - return new ChunkResult(ChunkResult.SUCCESS, chunk.getId(), skippedItems); + return new DechunkingResult(true, chunk.getId(), skippedItems); } public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java index da279f4cb..eb224909d 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java @@ -18,7 +18,6 @@ package org.springframework.batch.execution.step.simple; import java.util.ArrayList; import java.util.Arrays; -import java.util.List; import junit.framework.TestCase; @@ -31,10 +30,6 @@ import org.springframework.batch.core.domain.JobSupport; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.execution.repository.SimpleJobRepository; -import org.springframework.batch.execution.repository.dao.MapJobDao; -import org.springframework.batch.execution.repository.dao.MapStepDao; -import org.springframework.batch.execution.repository.dao.StepDao; import org.springframework.batch.execution.scope.StepScope; import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.item.ExecutionAttributes; 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 index dadd3b7c7..3ccd3483f 100644 --- 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 @@ -17,7 +17,7 @@ package org.springframework.batch.execution.step.simple; import junit.framework.TestCase; -import org.springframework.batch.core.domain.Chunk; +import org.springframework.batch.core.domain.ChunkingResult; import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; @@ -54,15 +54,15 @@ public class ItemChunkerTests extends TestCase { public void testSizePositive() { MockItemReader itemReader = new MockItemReader(10); ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - Chunk chunk = chunkReader.chunk(10); - assertEquals(10, chunk.getItems().size()); + ChunkingResult chunkingResult = chunkReader.chunk(10); + assertEquals(10, chunkingResult.getChunk().getItems().size()); } public void testIncompleteChunk() { MockItemReader itemReader = new MockItemReader(5); ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - Chunk chunk = chunkReader.chunk(10); - assertEquals(5, chunk.getItems().size()); + ChunkingResult chunkingResult = chunkReader.chunk(10); + assertEquals(5, chunkingResult.getChunk().getItems().size()); } public void testPolicyNoContinue() { @@ -82,8 +82,8 @@ public class ItemChunkerTests extends TestCase { itemReader.setFail(true); ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(false)); - Chunk chunk = chunkReader.chunk(1); - assertEquals(1, chunk.getItems().size()); + ChunkingResult chunkingResult = chunkReader.chunk(1); + assertEquals(1,chunkingResult.getChunk().getItems().size()); } private class StubReadFailurePolicy implements ItemSkipPolicy { diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java index 407316660..994540b0b 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java @@ -20,8 +20,9 @@ import java.util.List; import org.easymock.MockControl; import org.springframework.batch.core.domain.Chunk; -import org.springframework.batch.core.domain.ChunkResult; +import org.springframework.batch.core.domain.DechunkingResult; import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.io.exception.WriteFailureException; import org.springframework.batch.item.ItemWriter; import junit.framework.TestCase; @@ -71,9 +72,12 @@ public class ItemDechunkerTests extends TestCase { itemWriter.write("2"); writerControl.setThrowable(new Exception()); writerControl.replay(); - ChunkResult result = dechunker.dechunk(chunk); + DechunkingResult result = dechunker.dechunk(chunk); writerControl.verify(); - assertEquals("2",result.getSkippedItems().get(0)); + List exceptions = result.getExceptions(); + assertEquals(1, exceptions.size()); + WriteFailureException exception = (WriteFailureException)exceptions.get(0); + assertEquals("2",exception.getItem()); }