diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java index dbe795e51..af45821b3 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java @@ -177,6 +177,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw */ public final void execute(StepExecution stepExecution) throws JobInterruptedException, UnexpectedJobExecutionException { + + logger.debug("Executing: id="+stepExecution.getId()); stepExecution.setStartTime(new Date()); stepExecution.setStatus(BatchStatus.STARTED); getJobRepository().update(stepExecution); @@ -204,7 +206,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw } stepExecution.setStatus(BatchStatus.COMPLETED); - logger.debug("Step execution success: " + stepExecution); + logger.debug("Step execution success: id=" + stepExecution.getId()); } catch (Throwable e) { logger.error("Encountered an error executing the step: " + e.getClass() + ": " + e.getMessage(), 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 8b239ed20..730910d1d 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 @@ -26,8 +26,8 @@ import java.util.List; * 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. + * failure call {@link ChunkIterator#remove(Exception)} on the iterator. The + * skipped items are then available through the chunk. * * @author Dave Syer * @@ -39,25 +39,27 @@ public class Chunk implements Iterable { private List> skips = new ArrayList>(); private List errors = new ArrayList(); - + private Object userData; private boolean end; + private boolean busy; + public Chunk() { - this(null,null); + this(null, null); } public Chunk(Collection items) { - this(items,null); + this(items, null); } - + public Chunk(Collection items, List> skips) { super(); - if (items!=null) { + if (items != null) { this.items = new ArrayList(items); } - if (skips!=null) { + if (skips != null) { this.skips = new ArrayList>(skips); } } @@ -132,14 +134,50 @@ public class Chunk implements Iterable { return items.size(); } + /** + * Flag to indicate if the source data is exhausted. + * + * @return true if there is no more data to process + */ public boolean isEnd() { return end; } + /** + * Set the flag to say that this chunk represents an end of stream (there is + * no more data to process). + */ public void setEnd() { this.end = true; } + /** + * Query the chunk to see if anyone has registered an interest in keeping a + * reference to it. + * + * @return the busy flag + */ + public boolean isBusy() { + return busy; + } + + /** + * Register an interest in the chunk to prevent it from being cleaned up + * before the flag is reset to false. + * + * @param busy the flag to set + */ + public void setBusy(boolean busy) { + this.busy = busy; + } + + /** + * Clear only the skips list. + */ + public void clearSkips() { + skips.clear(); + } + public Object getUserData() { return userData; } @@ -159,9 +197,8 @@ public class Chunk implements Iterable { } /** - * Special iterator for a chunk providing the - * {@link #remove(Exception)} method for dynamically removing an - * item and adding it to the skips. + * Special iterator for a chunk providing the {@link #remove(Exception)} + * method for dynamically removing an item and adding it to the skips. * * @author Dave Syer * diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java new file mode 100644 index 000000000..7c1e45068 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java @@ -0,0 +1,150 @@ +/* + * 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.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemStreamException; + +/** + * Manage the offset data between the last successful commit and updates made to + * an input chunk. Only works with single threaded steps because it has to use a + * {@link ThreadLocal} to manage the state and co-ordinate between the caller + * and the wrapped {@link ItemStream}. + * + * @author Dave Syer + * + */ +class ChunkMonitor implements ItemStream { + + private Log logger = LogFactory.getLog(getClass()); + + public static class ChunkMonitorData { + public int offset; + + public int chunkSize; + + public ChunkMonitorData(int offset, int chunkSize) { + this.offset = offset; + this.chunkSize = chunkSize; + } + } + + private static final String OFFSET = ChunkMonitor.class.getName() + ".OFFSET"; + + private ItemStream stream; + + private ThreadLocal holder = new ThreadLocal(); + { + // For testing purposes, make an instance of the offset data + // available: + holder.set(new ChunkMonitorData(0, 0)); + } + + private ItemReader reader; + + /** + * @param stream the stream to set + */ + public void setItemStream(ItemStream stream) { + this.stream = stream; + } + + /** + * @param reader the reader to set + */ + public void setItemReader(ItemReader reader) { + this.reader = reader; + } + + public void incrementOffset() { + ChunkMonitorData data = getData(); + data.offset ++; + if (data.offset >= data.chunkSize) { + resetOffset(); + } + } + + public int getOffset() { + return getData().offset; + } + + public void resetOffset() { + getData().offset = 0; + } + + public void setChunkSize(int chunkSize) { + getData().chunkSize = chunkSize; + resetOffset(); + } + + public void close() throws ItemStreamException { + holder.set(new ChunkMonitorData(0,0)); + if (stream != null) { + stream.close(); + } + } + + public void open(ExecutionContext executionContext) throws ItemStreamException { + if (stream != null) { + stream.open(executionContext); + ChunkMonitorData data = new ChunkMonitorData(executionContext.getInt(OFFSET, 0), 0); + holder.set(data); + if (reader == null) { + logger.warn("No ItemReader set (must be concurrent step), so ignoring offset data."); + return; + } + for (int i = 0; i < data.offset; i++) { + try { + reader.read(); + } + catch (Exception e) { + throw new ItemStreamException("Could not position reader with offset: " + data.offset, e); + } + } + } + } + + public void update(ExecutionContext executionContext) throws ItemStreamException { + if (stream != null) { + ChunkMonitorData data = getData(); + if (data.offset == 0) { + // Only call the underlying update method if we are on a chunk + // boundary + stream.update(executionContext); + } + else { + executionContext.putInt(OFFSET, data.offset); + } + } + } + + private ChunkMonitorData getData() { + ChunkMonitorData data = holder.get(); + if (data==null) { + if (stream!=null) { + logger.warn("ItemStream was opened in a different thread. Restart data could be compromised."); + } + data = new ChunkMonitorData(0,0); + holder.set(data); + } + return data; + } + +} \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java index b9182c877..15c73e373 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java @@ -48,7 +48,7 @@ public class ChunkOrientedTasklet implements Tasklet { /** * Flag to indicate that items should be buffered once read. Defaults to * true, which is appropriate for forward-only, non-transactional item - * readers. Main (or only) use case for setting this flag to false is a + * readers. Main (or only) use case for setting this flag to true is a * transactional JMS item reader. * * @param buffering @@ -69,10 +69,18 @@ public class ChunkOrientedTasklet implements Tasklet { } chunkProcessor.process(contribution, inputs); + chunkProvider.postProcess(contribution, inputs); + + // Allow a message coming back from the processor to say that we + // are not done yet + if (inputs.isBusy()) { + // TODO: update ExecutionContext with an offset if the + // ItemReader was stateful + return RepeatStatus.CONTINUABLE; + } chunkContext.removeAttribute(INPUTS_KEY); chunkContext.setComplete(); - chunkProvider.postProcess(contribution, inputs); if (inputs.isEnd()) { contribution.setExitStatus(ExitStatus.COMPLETED); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java.svntmp b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java.svntmp new file mode 100644 index 000000000..15c73e373 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java.svntmp @@ -0,0 +1,92 @@ +/* + * 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.ExitStatus; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.scope.context.ChunkContext; +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.repeat.RepeatStatus; + +/** + * A {@link Tasklet} implementing variations on read-process-write item + * handling. + * + * @author Dave Syer + * + * @param input item type + */ +public class ChunkOrientedTasklet implements Tasklet { + + private static final String INPUTS_KEY = "INPUTS"; + + private final ChunkProcessor chunkProcessor; + + private final ChunkProvider chunkProvider; + + private boolean buffering = true; + + public ChunkOrientedTasklet(ChunkProvider chunkProvider, ChunkProcessor chunkProcessor) { + this.chunkProvider = chunkProvider; + this.chunkProcessor = chunkProcessor; + } + + /** + * Flag to indicate that items should be buffered once read. Defaults to + * true, which is appropriate for forward-only, non-transactional item + * readers. Main (or only) use case for setting this flag to true is a + * transactional JMS item reader. + * + * @param buffering + */ + public void setBuffering(boolean buffering) { + this.buffering = buffering; + } + + public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { + + @SuppressWarnings("unchecked") + Chunk inputs = (Chunk) chunkContext.getAttribute(INPUTS_KEY); + if (inputs == null) { + inputs = chunkProvider.provide(contribution); + if (buffering) { + chunkContext.setAttribute(INPUTS_KEY, inputs); + } + } + + chunkProcessor.process(contribution, inputs); + chunkProvider.postProcess(contribution, inputs); + + // Allow a message coming back from the processor to say that we + // are not done yet + if (inputs.isBusy()) { + // TODO: update ExecutionContext with an offset if the + // ItemReader was stateful + return RepeatStatus.CONTINUABLE; + } + + chunkContext.removeAttribute(INPUTS_KEY); + chunkContext.setComplete(); + if (inputs.isEnd()) { + contribution.setExitStatus(ExitStatus.COMPLETED); + } + + return RepeatStatus.continueIf(!inputs.isEnd()); + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/KeyGenerator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/KeyGenerator.java new file mode 100644 index 000000000..116c1eb37 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/KeyGenerator.java @@ -0,0 +1,26 @@ +/* + * 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; + +/** + * @author Dave Syer + * + */ +public interface KeyGenerator { + + Object getKey(Object item); + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/OffsetItemReader.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/OffsetItemReader.java new file mode 100644 index 000000000..dc4d1a237 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/OffsetItemReader.java @@ -0,0 +1,56 @@ +package org.springframework.batch.core.step.item; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; + +/** + * Convenience wrapper for an ItemReader that keeps track of how many items + * successfully processed. + */ +class OffsetItemReader implements ItemReader, ItemStream { + + private static final String OFFSET_KEY = FaultTolerantStepFactoryBean.class.getName()+".OFFSET_KEY"; + private final ItemReader itemReader; + private int offset; + + /** + * @param itemReader + */ + public OffsetItemReader(ItemReader itemReader) { + this.itemReader = itemReader; + } + + public T read() throws Exception, UnexpectedInputException, ParseException { + for (int i=0; i streams = getStreams("s1", taskletElementParentAttributeParserTestsContext); assertEquals(2, streams.size()); boolean c = false; - boolean d = false; for (ItemStream o : streams) { if (o instanceof CompositeItemStream) { c = true; } - else if (o instanceof TestReader) { - d = true; - } } assertTrue(c); - assertTrue(d); } @Test diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java index 1832c81e5..60c80744d 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java @@ -12,11 +12,11 @@ import org.springframework.batch.item.ParseException; import org.springframework.batch.item.UnexpectedInputException; public class TestReader extends AbstractTestComponent implements ItemReader, ItemStream { - + private boolean opened = false; List items = null; - + { List l = new ArrayList(); l.add("Item *** 1 ***"); @@ -31,28 +31,26 @@ public class TestReader extends AbstractTestComponent implements ItemReader 0) { - String item = items.remove(0); - return item; + synchronized (items) { + if (items.size() > 0) { + String item = items.remove(0); + return item; + } } return null; } - public void close() - throws ItemStreamException { + public void close() throws ItemStreamException { } - public void open(ExecutionContext executionContext) - throws ItemStreamException { + public void open(ExecutionContext executionContext) throws ItemStreamException { opened = true; } - public void update(ExecutionContext executionContext) - throws ItemStreamException { + public void update(ExecutionContext executionContext) throws ItemStreamException { } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java new file mode 100644 index 000000000..4296897f6 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java @@ -0,0 +1,137 @@ +/* + * 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.assertTrue; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.ItemStreamSupport; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; + +/** + * @author Dave Syer + * + */ +public class ChunkMonitorTests { + + /** + * + */ + private static final int CHUNK_SIZE = 5; + + private ChunkMonitor monitor = new ChunkMonitor(); + + private int count = 0; + + private boolean closed = false; + + @Before + public void setUp() { + monitor.setItemReader(new ItemReader() { + public String read() throws Exception, UnexpectedInputException, ParseException { + return "" + (count++); + } + }); + monitor.setItemStream(new ItemStreamSupport() { + @Override + public void close() throws ItemStreamException { + closed = true; + } + }); + monitor.setChunkSize(CHUNK_SIZE); + } + + @Test + public void testIncrementOffset() { + assertEquals(0, monitor.getOffset()); + monitor.incrementOffset(); + assertEquals(1, monitor.getOffset()); + } + + @Test + public void testResetOffsetManually() { + monitor.incrementOffset(); + monitor.resetOffset(); + assertEquals(0, monitor.getOffset()); + } + + @Test + public void testResetOffsetAutomatically() { + for (int i = 0; i < CHUNK_SIZE; i++) { + monitor.incrementOffset(); + } + assertEquals(0, monitor.getOffset()); + } + + @Test + public void testClose() { + monitor.incrementOffset(); + monitor.close(); + assertTrue(closed); + assertEquals(0, monitor.getOffset()); + } + + @Test + public void testOpen() { + ExecutionContext executionContext = new ExecutionContext(); + executionContext.putInt(ChunkMonitor.class.getName() + ".OFFSET", 2); + monitor.open(executionContext); + assertEquals(2, count); + } + + @Test + public void testOpenWithNullReader() { + monitor.setItemReader(null); + ExecutionContext executionContext = new ExecutionContext(); + monitor.open(executionContext); + assertEquals(0, monitor.getOffset()); + } + + @Test(expected = ItemStreamException.class) + public void testOpenWithErrorInReader() { + monitor.setItemReader(new ItemReader() { + public String read() throws Exception, UnexpectedInputException, ParseException { + throw new IllegalStateException("Expected"); + } + }); + ExecutionContext executionContext = new ExecutionContext(); + executionContext.putInt(ChunkMonitor.class.getName() + ".OFFSET", 2); + monitor.open(executionContext); + } + + @Test + public void testUpdateOnBoundary() { + monitor.resetOffset(); + ExecutionContext executionContext = new ExecutionContext(); + monitor.update(executionContext); + assertEquals(0, executionContext.size()); + } + + @Test + public void testUpdateVanilla() { + monitor.incrementOffset(); + ExecutionContext executionContext = new ExecutionContext(); + monitor.update(executionContext); + assertEquals(1, executionContext.size()); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java index ce4751282..966bbd8ca 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java @@ -100,6 +100,8 @@ public class FaultTolerantChunkProcessorTests { catch (RuntimeException e) { assertEquals("Planned failure!", e.getMessage()); } + processor.process(contribution, chunk); + assertEquals(2, chunk.getItems().size()); try { processor.process(contribution, chunk); fail(); @@ -107,11 +109,12 @@ public class FaultTolerantChunkProcessorTests { catch (RuntimeException e) { assertEquals("Planned failure!", e.getMessage()); } - assertEquals(2, chunk.getItems().size()); + assertEquals(1, chunk.getItems().size()); processor.process(contribution, chunk); + assertEquals(0, chunk.getItems().size()); // foo is written twice because the failure is detected on the second // attempt when throttling - assertEquals("[foo, foo, bar]", list.toString()); + assertEquals("[foo, bar]", list.toString()); // but the after listener is only called once, which is important assertEquals(2, after.size()); } @@ -122,7 +125,7 @@ public class FaultTolerantChunkProcessorTests { processor = new FaultTolerantChunkProcessor(new PassThroughItemProcessor(), new ItemWriter() { public void write(List items) throws Exception { - // Fail is there is more than one item + // Fail if there is more than one item if (items.size() > 1) { throw new RuntimeException("Planned failure!"); } @@ -145,6 +148,7 @@ public class FaultTolerantChunkProcessorTests { assertEquals("Planned failure!", e.getMessage()); } processor.process(contribution, chunk); + processor.process(contribution, chunk); assertEquals("[foo, bar]", list.toString()); assertEquals("[foo, bar]", after.toString()); 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 0d923a734..27ebdf349 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 @@ -42,8 +42,11 @@ import org.springframework.batch.core.repository.dao.MapJobInstanceDao; import org.springframework.batch.core.repository.dao.MapStepExecutionDao; import org.springframework.batch.core.repository.support.SimpleJobRepository; import org.springframework.batch.core.step.AbstractStep; +import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader; import org.springframework.batch.item.support.ListItemReader; import org.springframework.batch.retry.policy.MapRetryContextCache; import org.springframework.batch.retry.policy.SimpleRetryPolicy; @@ -72,12 +75,14 @@ public class FaultTolerantStepFactoryBeanRetryTests { int count = 0; + boolean fail = false; + private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), new MapStepExecutionDao(), new MapExecutionContextDao()); JobExecution jobExecution; - private ItemWriter processor = new ItemWriter() { + private ItemWriter writer = new ItemWriter() { public void write(List data) throws Exception { processed.addAll(data); } @@ -98,7 +103,7 @@ public class FaultTolerantStepFactoryBeanRetryTests { factory.setBeanName("step"); factory.setItemReader(new ListItemReader(new ArrayList())); - factory.setItemWriter(processor); + factory.setItemWriter(writer); factory.setJobRepository(repository); factory.setTransactionManager(new ResourcelessTransactionManager()); factory.setRetryableExceptionClasses(new HashSet>() { @@ -167,9 +172,70 @@ public class FaultTolerantStepFactoryBeanRetryTests { assertEquals(0, stepExecution.getReadSkipCount()); } + @Test + public void testRestartAfterFailedWrite() throws Exception { + + factory.setSkipLimit(0); + factory.setCommitInterval(3); + AbstractItemCountingItemStreamItemReader reader = new AbstractItemCountingItemStreamItemReader() { + + private ItemReader reader; + + @Override + protected void doClose() throws Exception { + reader = null; + } + + @Override + protected void doOpen() throws Exception { + reader = new ListItemReader(Arrays.asList("a", "b", "c", "d", "e", "f")); + } + + @Override + protected String doRead() throws Exception { + return reader.read(); + } + + }; + // Need to set name or else reader will fail to open + reader.setName("foo"); + factory.setItemReader(reader); + factory.setStreams(new ItemStream[] { reader }); + factory.setItemWriter(new ItemWriter() { + public void write(List items) throws Exception { + if (fail && items.contains("e")) { + throw new RuntimeException("Planned failure"); + } + processed.addAll(items); + } + }); + factory.setRetryLimit(0); + Step step = (Step) factory.getObject(); + + fail = true; + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + repository.add(stepExecution); + step.execute(stepExecution); + + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + assertEquals(4, stepExecution.getWriteCount()); + assertEquals(6, stepExecution.getReadCount()); + + fail = false; + ExecutionContext executionContext = stepExecution.getExecutionContext(); + stepExecution = new StepExecution(step.getName(), jobExecution); + stepExecution.setExecutionContext(executionContext); + repository.add(stepExecution); + step.execute(stepExecution); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getWriteCount()); + assertEquals(2, stepExecution.getReadCount()); + } + @Test public void testSkipAndRetry() throws Exception { - + factory.setSkipLimit(2); ItemReader provider = new ListItemReader(Arrays.asList("a", "b", "c", "d", "e", "f")) { public String read() { @@ -198,7 +264,7 @@ public class FaultTolerantStepFactoryBeanRetryTests { @Test public void testSkipAndRetryWithWriteFailure() throws Exception { - factory.setListeners(new StepListener[] { new SkipListenerSupport() { + factory.setListeners(new StepListener[] { new SkipListenerSupport() { public void onSkipInWrite(String item, Throwable t) { recovered.add(item); assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); @@ -258,7 +324,7 @@ public class FaultTolerantStepFactoryBeanRetryTests { public void testSkipAndRetryWithWriteFailureAndNonTrivialCommitInterval() throws Exception { factory.setCommitInterval(3); - factory.setListeners(new StepListener[] { new SkipListenerSupport() { + factory.setListeners(new StepListener[] { new SkipListenerSupport() { public void onSkipInWrite(String item, Throwable t) { recovered.add(item); assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); @@ -308,16 +374,17 @@ public class FaultTolerantStepFactoryBeanRetryTests { // [a, b, c, d, e, f, null] assertEquals(7, provided.size()); - // [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, a, c, d, e, f, d, + // [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, d, e, f, d, // e, f, d, e, f, d, e, f, d, e, f, d, e, f] - assertEquals(37, processed.size()); + System.err.println(processed); + assertEquals(36, processed.size()); // [b, d] assertEquals(2, recovered.size()); } @Test public void testRetryWithNoSkip() throws Exception { - + factory.setRetryLimit(4); factory.setSkipLimit(0); ItemReader provider = new ListItemReader(Arrays.asList("b")) { @@ -343,8 +410,8 @@ public class FaultTolerantStepFactoryBeanRetryTests { StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); repository.add(stepExecution); step.execute(stepExecution); - assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); - + assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("")); assertEquals(expectedOutput, written); @@ -440,7 +507,7 @@ public class FaultTolerantStepFactoryBeanRetryTests { repository.add(stepExecution); step.execute(stepExecution); assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); - + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("")); assertEquals(expectedOutput, written); @@ -489,7 +556,7 @@ public class FaultTolerantStepFactoryBeanRetryTests { repository.add(stepExecution); step.execute(stepExecution); assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); - + // We added a bogus cache so no items are actually skipped // because they aren't recognised as eligible assertEquals(0, stepExecution.getSkipCount()); 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 55b39a146..1ca1884ff 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 @@ -30,14 +30,18 @@ import org.springframework.batch.core.StepListener; import org.springframework.batch.core.listener.SkipListenerSupport; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.ItemStreamReader; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.ParseException; import org.springframework.batch.item.UnexpectedInputException; import org.springframework.batch.item.support.ListItemReader; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.util.StringUtils; @@ -66,7 +70,11 @@ public class FaultTolerantStepFactoryBeanTests { private List processed = new ArrayList(); - protected int count; + private int count; + + private boolean opened = false; + + private boolean closed = false; private Collection NO_FAILURES = Collections.emptyList(); @@ -380,9 +388,10 @@ public class FaultTolerantStepFactoryBeanTests { // listeners are called only once chunk is about to commit, so // listener failure does not affect other statistics - assertEquals(3, stepExecution.getSkipCount()); assertEquals(2, stepExecution.getReadSkipCount()); - assertEquals(1, stepExecution.getWriteSkipCount()); + // but we didn't get as far as the write skip in the scan: + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(2, stepExecution.getSkipCount()); assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step .getName())); } @@ -599,10 +608,10 @@ public class FaultTolerantStepFactoryBeanTests { assertEquals(1, stepExecution.getSkipCount()); assertEquals(2, stepExecution.getRollbackCount()); - // 1,2,3,4,3,4,3 - two re-processing attempts until the item is - // identified and skipped + // 1,2,3,4,3,4,4 - two re-processing attempts until the item is + // identified and finally skipped on the third attempt assertEquals(7, processed.size()); - assertEquals("[1, 2, 3, 4, 3, 4, 3]", processed.toString()); + assertEquals("[1, 2, 3, 4, 3, 4, 4]", processed.toString()); assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step .getName())); @@ -683,8 +692,42 @@ public class FaultTolerantStepFactoryBeanTests { } } - private static class SkipProcessorStub implements ItemProcessor { + /** + * Check ItemStream is opened + */ + @Test + public void testItemStreamOpenedEvenWithTaskExecutor() throws Exception { + ItemStreamReader reader = new ItemStreamReader() { + public void close() throws ItemStreamException { + closed = true; + } + + public void open(ExecutionContext executionContext) throws ItemStreamException { + opened = true; + } + + public void update(ExecutionContext executionContext) throws ItemStreamException { + } + + public String read() throws Exception, UnexpectedInputException, ParseException { + return null; + } + }; + + factory.setItemReader(reader); + factory.setTaskExecutor(new ConcurrentTaskExecutor()); + + Step step = (Step) factory.getObject(); + + step.execute(stepExecution); + + assertTrue(opened); + assertTrue(closed); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + } + + private static class SkipProcessorStub implements ItemProcessor { private final Collection failures; private boolean rollback = false; diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java index cbef14b8d..904a5b754 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java @@ -126,16 +126,31 @@ public class TaskletStepTests { @Test public void testStepExecutor() throws Exception { - JobExecution jobExecutionContext = new JobExecution(jobInstance); StepExecution stepExecution = new StepExecution(step.getName(), jobExecutionContext); - step.execute(stepExecution); assertEquals(1, processed.size()); assertEquals(1, stepExecution.getReadCount()); assertEquals(1, stepExecution.getCommitCount()); } + @Test + public void testEmptyReader() throws Exception { + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step.getName(), jobExecutionContext); + step = getStep(new String[0]); + step.setTasklet(new TestingChunkOrientedTasklet(getReader(new String[0]), itemWriter, + new RepeatTemplate())); + step.setStepOperations(new RepeatTemplate()); + step.execute(stepExecution); + assertEquals(0, processed.size()); + assertEquals(0, stepExecution.getReadCount()); + // Commit after end of data detected (this leads to the commit count + // being one greater than people expect if the commit interval is + // commensurate with the total number of items).h + assertEquals(1, stepExecution.getCommitCount()); + } + /** * StepExecution should be updated after every chunk commit. */ diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/common-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/common-context.xml index db7711ebd..b0bf021f6 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/common-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/common-context.xml @@ -34,9 +34,9 @@ - - - - - + + + + + \ No newline at end of file diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java index e7a0d1dfd..56bb2981f 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/common/StagingItemReader.java @@ -95,7 +95,7 @@ public class StagingItemReader implements ItemReader read() throws DataAccessException { if (!initialized) { - throw new ReaderNotOpenException("ItemStream must be open before it can be read."); + throw new ReaderNotOpenException("Reader must be open before it can be used."); } Long id = null; diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java index a342bbc05..f51e8f586 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/SkipSampleFunctionalTests.java @@ -130,11 +130,11 @@ public class SkipSampleFunctionalTests { // // Launch 1 // - long id1 = this.launchJobWithIncrementer(); - Map execution1 = this.getJobExecution(id1); + long id1 = launchJobWithIncrementer(); + Map execution1 = this.getJobExecutionAsMap(id1); assertEquals("COMPLETED", execution1.get("STATUS")); - this.validateLaunchWithSkips(id1); + validateLaunchWithSkips(id1); // // Clear the data @@ -144,11 +144,11 @@ public class SkipSampleFunctionalTests { // // Launch 2 // - long id2 = this.launchJobWithIncrementer(); - Map execution2 = this.getJobExecution(id2); + long id2 = launchJobWithIncrementer(); + Map execution2 = getJobExecutionAsMap(id2); assertEquals("COMPLETED", execution2.get("STATUS")); - this.validateLaunchWithoutSkips(id2); + validateLaunchWithoutSkips(id2); // // Make sure that the launches were separate executions and separate @@ -176,8 +176,8 @@ public class SkipSampleFunctionalTests { assertEquals(new BigDecimal("340.45"), tradeWriter.getTotalPrice()); - Map step1Execution = this.getStepExecution(jobExecutionId, "step1"); - assertEquals(new Long(3), step1Execution.get("COMMIT_COUNT")); + Map step1Execution = getStepExecutionAsMap(jobExecutionId, "step1"); + assertEquals(new Long(4), step1Execution.get("COMMIT_COUNT")); assertEquals(new Long(8), step1Execution.get("READ_COUNT")); assertEquals(new Long(7), step1Execution.get("WRITE_COUNT")); } @@ -195,12 +195,12 @@ public class SkipSampleFunctionalTests { assertEquals(new BigDecimal("270.75"), tradeWriter.getTotalPrice()); } - private Map getJobExecution(long jobExecutionId) { + private Map getJobExecutionAsMap(long jobExecutionId) { return simpleJdbcTemplate.queryForMap("SELECT * from BATCH_JOB_EXECUTION where JOB_EXECUTION_ID = ?", jobExecutionId); } - private Map getStepExecution(long jobExecutionId, String stepName) { + private Map getStepExecutionAsMap(long jobExecutionId, String stepName) { return simpleJdbcTemplate.queryForMap( "SELECT * from BATCH_STEP_EXECUTION where JOB_EXECUTION_ID = ? and STEP_NAME = ?", jobExecutionId, stepName); diff --git a/src/site/apt/stateful.apt b/src/site/apt/stateful.apt new file mode 100644 index 000000000..3d484f61c --- /dev/null +++ b/src/site/apt/stateful.apt @@ -0,0 +1,161 @@ + ------ + Spring Batch - State and Thread Safety + ------ + Dave Syer + ------ + March 2009 + +State and Thread Safety in Spring Batch + + A stateless component is thread safe, but sometimes not practical (you need to store some state). A stateful component can be thread safe, if its contract is clearly explained to and met by its clients. Spring Batch has a lot of stateful components, which by and large are not capable of being used in a thread safe manner, but that doesn't have to be the case for ever. + + Components with private non-final fields are not necessarily stateful in practice - Spring components often have fields that are injected or initialized after the object is created. The working definition of "stateless" for the present purposes is a component with fields that do not change after initialization, which has the usual Spring lifecycle meaning (i.e. once "released into the wild" with <<>, or the equivalent). + + Conversely, even components with only final fields are not necessarily stateless. They can have the appearance of statelessness (and thread safety), but if they mutate their final fields, then they are stateful by association. It is not always possible to tell from the implementation of a component whether it is stateful by association, since it depends entirely on the implementation of its fields, whose concrete type may not even be known at compile time. + +* Variants of State + + There are two reasons why a Batch component might need to be stateful: rollback and restart. + + * : to support rollback after a transaction, a component might need to detect the rollback (and potentially the start of the original transaction) and rewind to its former state. This is a single-process (JVM) pattern. Normally a transaction is managed on a single thread as well, but in principle there might be multiple threads using the same component, which inevitably leads to problems. + + * : to support restart of a failed job execution, a component needs to be able to re-hydrate its former state from a previour execution. This is often a multi-process (JVM) pattern, and needs to work between processes even if it isn't always invoked that way in practice. To support this requirement Spring Batch uses the <<>> calback methods and its <<>> to store state. The framework deals with the state storage and re-hydration, and components only need to provide snapshots of their state through the <<>> interface. + +* ItemReaders + + The most common case in Batch where stateful components are necessary are the item readers, whose job is to provide instances of business data for processing. At a high level there are three variants, driven by the needs of their client (usually the framework) for rollback and restart. + + * . A fully transactional reader has its rollback and restart state managed entirely by an external system (middleware) that is driven from a transaction in the batch system. After a rollback items are returned to the middleware, and represented in a subsequent transaction. There is no essential difference between rollback and retry: as far as the middleware is concerned they are just failed reads. The user has to tell Spring Batch explicitly if one of these readers is being used so it can take into account the expected re-presentation of items in a new transaction. Just about the only example of this is the <<>>. + + * . A stateless component doesn't need to do anything special to provide restartability, which is quite a valuable feature. It is also thread safe (by construction), which is at least equally valuable if not more. Stateless <<>> are rare in practice: in the Spring Batch source code the only one is the <<>> from the samples. + + * . Part of the contract for <<>> implementations in Batch 2.0 is that <<>> do not need to manage state for rollback because each item is only ever read once (and buffered for use on rollback internally). They do, however still need state if they want to provide restartability, which all the framework implementations do. A stateful component has to work hard to be thread safe, but it is not ruled out in principle. In fact none of the stateful <<>> implementations in Spring Batch is thread safe as of 2.0, but they are all restartable (which is more to the point). + + Because these different categories of <<>> behave differently with respect to rollback and restart, they have to be recognised and treated differently by the framework. If a simple fail fast <<>> is used, then any error leads to immediate failure, so rollback is not important in that case, but restart always is. If a fault tolerant <<>> is used, then rollback and restart have to be handled, and differently for each category of reader. To recognise a reader the framework needs a flag to be set by the user (see <<>>). To recognise a or reader the framework uses the <<>> interface (if present assume ). + +Rollback and Restart with Skips + +* Spring Batch 2.0.0.RC1 + +** Retry and Skip with a Transactional Reader + + Since the middleware will simply re-present failed items, there is nothing for the framework to do. We can use this example to establish some notation. Suppose five items are read in one transaction and there is a deterministic failure while writing the 3rd one. If skips are allowed, but not retries, the process looks like this: + ++--- +[1,2,3,4,5; (1,2,3,4,5)]* +[1,2,3,4,5; (1),(2),(3)]* +[1,2,3,4,5; (1,2,4,5)] +... ++--- + + where parentheses denote a write operation, brackets ([]) represent a transaction, and an asterisk (*) denotes a rollback. In words we have: + + * Read items 1 through 5, then write them as a chunk, encounter an error and rollback. + + * Read items 1 through 5 again and write them individually, scanning for errors. Encounter the error on item 3, then rollback having identified item 3 as skippable + + * Read the 5 items again and skip item 3 on writing the chunk. + + To achieve this, the items from a failed chunk need to be intrinsically identifiable, so that when they show up again the system can throttle back and scan for the error. In completely general terms this is not a well-defined problem - tere is no generic identifier for the items that can always be used. Spring Batch by default uses the items themselves as the identifier in this case, leading to possible problems if their identity (equals and hashCode) are not properly defined. We might be able to pick an identifier in some special cases, like in the JMS case the message ID will be unique, in which case we would need to provide a <<>> implementation. (This is uncommon enough that in Spring Batch 2.0.0 there is no way to do it using the XML namespace, but you can do it using <<>>.) + + There is also the issue of storing the identifiers, waiting for all the failed items to be seen again. There is no guarantee that the failed items will ever come back to this consumer, so we have to store the identifiers potentially indefinitely, possibly leading to memory leaks. To help alleviate this problem (at the risk of misidentifying a failed item as a new one) Spring Batch provides the <<>> which allows cached values to be garabage collected if memory is under pressure. + + With a retry limit of 1, the same execution would look like this + ++--- +[1,2,3,4,5; (1,2,3,4,5)]* +[1,2,3,4,5; (1,2,3,4,5)]* +[1,2,3,4,5; (1),(2),(3)]* +[1,2,3,4,5; (1,2,4,5)] +... ++--- + + (just an extra iteration where the chunk is given a chance to succeed before the error scan starts). + +** Retry and Skip with a Non-Transactional Reader + + In this case there is no middleware so the <<>> has to buffer the items between rollbacks, but otherwise the process looks like very similar. The internal name for the buffer is a <<>>. With no retry: + ++--- +[1,2,3,4,5; (1,2,3,4,5)]* +[; (1),(2),(3)]* +[; (1,2,4,5)] +... ++--- + + and with retry: + ++--- +[1,2,3,4,5; (1,2,3,4,5)]* +[; (1,2,3,4,5)]* +[; (1),(2),(3)]* +[; (1,2,4,5)] +... ++--- + + In the case of the non-Transactional reader there is no problem with item identifiers: the <<>> can be used to identify the failed items when they are re-processed. + +* A More Efficient Approach + + It would be more efficient if we didn't end up processing each item more than twice in the simple case of skip with no retry. + +** Skip with a Transactional Reader + ++--- +[1,2,3,4,5; (1,2,3,4,5)]* +[1; (1)] +[2; (2)] +[3; (3)]* +[4; (4)] +[5; (5)] +... ++--- + + This is difficult to achieve with the <<>> because it requires communication between the <<>> and <<>> about the failed items: the <<>> has to stop reading when it encounters an item from a previously failed chunk. This feature is not yet implemented in Spring Batch (as of 2.0.0.RC2). + +** Skip with a Stateful Reader + ++--- +[1,2,3,4,5; (1,2,3,4,5)]* +[; (1)] +[; (2)] +[; (3)]* +[; (4)] +[; (5)] +... ++--- + + This processing and rollback plan is not difficult to achieve with the <<>> but it hard to get the restart data properly aligned. The main issue is that the <<>> gets the <<>> callback before every commit. It is going to think that all 5 items have been committed after the first (and every) commit, which is wrong, so any subsequent non-skippable failure will lead to a restart on item 6 with all intermediate items lost. So after a fatal error on item 2 the restart would like this + ++--- +[1,2,3,4,5; (1,2,3,4,5)]* +[; (1)] +[; (2)]* + + + +[6,7,8,9,10; (6,7,8,9,10)] +... ++--- + + and items (2,3,4,5) are ommitted with no record of attempting to process them. + + To fix this we have to mask the <<>> callback in the <<>> and only pass it through if we know we have successfully finished a whole chunk. In the partial chunk transactions, we need to update the <<>> with some offset data that would prevent the intermediate values being processed on restart. + ++--- +[1,2,3,4,5; (1,2,3,4,5)]* +[; (1)] +[; (2)]* + + + +[2,3,4,5,6; (2,3,4,5,6)]* +[; (2)] +[; (3)]* +[; (4)] +[; (5)] +[; (6)] +... ++--- + + This can be implemented by storing the offset within a chunk separately from the read count. In the example above, the offset would be 2 when the fatal exception happened, and the read count would be 0 up to the point where the first chunk successfully commits. Caveat: only works if the <<>> is single-threaded (which it has to be for all the existing Stateful readers). The implementation via <<>> assumes that the step is single threaded if it finds that the <<>> is an <<>>, and the <<>> is a <<>>. If the <<>> is concurrent then warnings are logged, and the offset is not stored )the assumption is that the reader is not restartable so it won't care.