diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Chunk.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Chunk.java deleted file mode 100644 index d0d4c11c9..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Chunk.java +++ /dev/null @@ -1,64 +0,0 @@ -/* - * 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.core.domain; - -import java.io.Serializable; -import java.util.ArrayList; -import java.util.List; - -/** - *

A 'chunk' of items, that will be committed together. It is expected that a - * chunk may be serialized, especially if using a queue to dispatch chunks to - * various {@link ChunkProcessor}s.

- * - *

A note on serialization: A chunk is only as serializable as as the items - * it contains, and no validation is performed to ensure they are serialiable. - * Therefore, it is expected that constructors of this class will ensure this - * condition if they intend to use the chunk in such a way that will require - * serialization.

- * - * @author Ben Hale - * @author Lucas Ward - */ -public class Chunk implements Serializable { - - private final Long id; - - private final List items; - - public Chunk(Long id, List items) { - this.items = items; - this.id = id; - } - - /** - * Get the list of items for this chunk. - * - * @return items. - */ - public List getItems() { - return new ArrayList(items); - } - - /** - * Get the chunk id. - * - * @return id of this chunk. - */ - public Long getId() { - return id; - } -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Chunker.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Chunker.java deleted file mode 100644 index fa7f85566..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Chunker.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * 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.core.domain; - -import org.springframework.batch.io.exception.ReadFailureException; -import org.springframework.batch.item.ItemStream; - - -/** - * 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 extends ItemStream { - - /** - * Read in a chunk, given the provided chunk size for the given StepExecution. - * - * @param chunkSize the number of items that should be read for this chunk. - * @param StepExecution the stepExecution the current chunk is being processed within. - * @return the {@link Chunk} that has been read. - * @throws IllegalArgumentException if chunkSize is less than zero. - */ - public ChunkingResult chunk(int chunkSize, StepContribution stepContribution) throws ReadFailureException; - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/ChunkingResult.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/ChunkingResult.java deleted file mode 100644 index f7dfcad62..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/ChunkingResult.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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.core.domain; - -import java.util.List; - -/** - * Value object representing the result of chunking. - * - * @author Lucas Ward - */ -public class ChunkingResult { - - private final Chunk chunk; - - private final List exceptions; - - public ChunkingResult(Chunk chunk, List exceptions) { - this.chunk = chunk; - this.exceptions = exceptions; - } - - public Chunk getChunk() { - return chunk; - } - - public List getExceptions() { - return exceptions; - } -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Dechunker.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Dechunker.java deleted file mode 100644 index efde754bd..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/Dechunker.java +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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.core.domain; - -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; - -/** - * Interface defining the contract for dechunking a {@link Chunk}. Similar - * to the differences between an {@link ItemReader} and an {@link ItemWriter}, - * a dechunker is the polar oposite of a {@link Chunker}. A {@link Chunker} creates - * a chunk from a stream of items, wheras a Dechunker removes each item from the - * {@link Chunk} and adds it to an outgoing stream of items. - * - * @author Lucas Ward - */ -public interface Dechunker extends ItemStream{ - - /** - * Dechunk the provided chunk. In general, this will be done by delegating to an - * {@link ItemWriter} for each item in the {@link Chunk}, however, it's purely at the - * discretion of various implementations. - * - * @param chunk to be 'dechunked' - * @param stepContribution the chunk belongs to. - * @return a Dechunking result detailing whether or not dechunking was successful, and - * if any items were skipped. - * @throws Exception, specifically throws IllegalArgumentException if either the chunk - * or StepExecution is null. - */ - DechunkingResult dechunk(Chunk chunk, StepContribution stepContribution) throws Exception; -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/DechunkingResult.java b/spring-batch-core/src/main/java/org/springframework/batch/core/domain/DechunkingResult.java deleted file mode 100644 index 417b0770d..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/domain/DechunkingResult.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * 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.domain; - -import java.util.ArrayList; -import java.util.List; - -/** - * Value object representing the result of dechunking a {@link Chunk}. It contains - * the id of the chunk that was processed, a list of exceptions, and whether or not - * the chunk was successful. - * - * @author Ben Hale - * @author Lucas Ward - */ -public class DechunkingResult { - - private final Long chunkId; - - private final List exceptions; - - private final boolean successful; - - public DechunkingResult(boolean successful, Long chunkId, List exceptions) { - this.chunkId = chunkId; - this.exceptions = exceptions; - this.successful = successful; - } - - public Long getChunkId() { - return chunkId; - } - - public List getExceptions() { - return new ArrayList(exceptions); - } - - public boolean isSuccessful() { - return successful; - } -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/ChunkTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/ChunkTests.java deleted file mode 100644 index 0bc2b85c2..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/ChunkTests.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * 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.core.domain; - -import java.util.ArrayList; -import java.util.List; - -import junit.framework.TestCase; - -/** - * @author Lucas Ward - * - */ -public class ChunkTests extends TestCase { - - List items; - Chunk chunk; - - protected void setUp() throws Exception { - super.setUp(); - - items = new ArrayList(); - items.add("1"); - items.add("2"); - } - - - public void testCreateChunk(){ - - chunk = new Chunk(new Long(1), items); - assertEquals(items, chunk.getItems()); - } - - public void testImmutability(){ - - chunk = new Chunk(new Long(1), items); - List testItems = chunk.getItems(); - testItems.add("3"); - assertEquals(2, chunk.getItems().size()); - } -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/ChunkingResultTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/domain/ChunkingResultTests.java deleted file mode 100644 index ce5b4e164..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/domain/ChunkingResultTests.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * 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.core.domain; - -import java.util.ArrayList; -import java.util.List; - -import junit.framework.TestCase; - -/** - * @author Lucas Ward - * - */ -public class ChunkingResultTests extends TestCase { - - private Chunk chunk; - private List exceptions; - - public void testChunk(){ - - chunk = new Chunk(new Long(1), new ArrayList()); - exceptions = new ArrayList(); - - ChunkingResult result = new ChunkingResult(chunk,exceptions); - - assertEquals(chunk,result.getChunk()); - assertEquals(exceptions,result.getExceptions()); - } -} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ChunkedStep.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ChunkedStep.java deleted file mode 100644 index a75321f6d..000000000 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ChunkedStep.java +++ /dev/null @@ -1,548 +0,0 @@ -/* - * 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; - -import java.util.Date; -import java.util.Iterator; - -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.Chunker; -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.ItemFailureHandler; -import org.springframework.batch.core.domain.ItemSkipPolicy; -import org.springframework.batch.core.domain.JobInstance; -import org.springframework.batch.core.domain.JobInterruptedException; -import org.springframework.batch.core.domain.Step; -import org.springframework.batch.core.domain.StepContribution; -import org.springframework.batch.core.domain.StepExecution; -import org.springframework.batch.core.domain.StepSupport; -import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier; -import org.springframework.batch.execution.scope.SimpleStepContext; -import org.springframework.batch.execution.scope.StepContext; -import org.springframework.batch.execution.scope.StepScope; -import org.springframework.batch.execution.scope.StepSynchronizationManager; -import org.springframework.batch.execution.step.support.DefaultItemFailureHandler; -import org.springframework.batch.execution.step.support.ItemChunker; -import org.springframework.batch.execution.step.support.ItemDechunker; -import org.springframework.batch.execution.step.support.SimpleExitStatusExceptionClassifier; -import org.springframework.batch.execution.step.support.StepInterruptionPolicy; -import org.springframework.batch.execution.step.support.ThreadStepInterruptionPolicy; -import org.springframework.batch.io.exception.BatchCriticalException; -import org.springframework.batch.io.exception.WriteFailureException; -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.exception.ResetFailedException; -import org.springframework.batch.item.stream.SimpleStreamManager; -import org.springframework.batch.item.stream.StreamManager; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatCallback; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatListener; -import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.retry.RetryCallback; -import org.springframework.batch.retry.RetryContext; -import org.springframework.batch.retry.RetryPolicy; -import org.springframework.batch.retry.policy.NeverRetryPolicy; -import org.springframework.batch.retry.support.RetryTemplate; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.transaction.TransactionStatus; -import org.springframework.util.Assert; - -/** - *

- * Implementation of the {@link Step} interface that deals with input and output - * as 'chunks'. Reading is delegated to a {@link Chunker} that will read in a - * {@link Chunk} of items for processing. The number of items per chunks is - * configurable as the chunk size. Once the chunk has been read, any errors - * encountered while reading (usually skipped unless configured not to) will be - * logged out via the {@link ItemFailureHandler}. The chunk will then be - * 'dechunked', which in most scenarios will mean delegating to an - * {@link ItemWriter} by writing out one chunk at a time. The transaction - * boundary is around this process. If any errors are encountered, the - * dechunking process will error out, leaving the decision for retrying the - * chunk up to a {@link RepeatTemplate}. This template is configurable, - * allowing for the number of retries and how long to wait between retries - * (backoff) to be set. Once dechunking has been finished, any errors not fatal - * to the chunk (usually because the error didn't invalidate the transaction) - * will also be written out via the {@link ItemFailureHandler} - *

- * - *

- * Clients can use {@link RepeatListener}s in the step operations to intercept - * or listen to the iteration on a step-wide basis, for instance to get a - * callback when the step is complete. The open and close methods of could - * easily be done with AOP, however, notifications in between complete chunks - * (before and after) can be quite useful - *

- * - *

- * Repository Usage: The {@link JobRepository} is used extensively to store - * metadata about the run such as when the {@link StepExecution} was started, or - * the commit count. - *

- * - *

- * Interruption: At various times while processing, the step will check to see - * if it has been interrupted by calling the {@link StepInterruptionPolicy}. - * This policy could check if thread.isInterupted() is true, or - * RepeatContext.isTerminateOnly() is set. It could even be a check to see if a - * 'stop file' has been added to a particular directory. If the step should - * finish, a {@link JobInterruptedException} is thrown, and the step will clean - * up, set the status of the {@link StepExecution} to 'STOPPED' and rethrow. - * ExitStatusClassification: Any number of fatal errors could be thrown during - * processing. In general, the framework must remain fairly dumb as to what - * error code these exceptions should translate to. By default it's a fairly - * generic 'FATAL_EXECUTION'. However, this may be insufficient for many - * scenarios. If an enterprise scheduler is used to kick off a batch job, the - * exit code is the only means of communication as to what action must be taken. - * It may also be the only result that many batch operators see as well. - * Therefore, an {@link ExitStatusExceptionClassifier} may be used to classify - * an exception to a particular exit code. - *

- * - * @author Dave Syer - * @author Lucas Ward - * @author Ben Hale - */ -public class ChunkedStep extends StepSupport implements InitializingBean { - - private static final Log logger = LogFactory.getLog(ChunkedStep.class); - - private RepeatOperations stepOperations = new RepeatTemplate(); - - private JobRepository jobRepository; - - // default to simple exception classification. - private ExitStatusExceptionClassifier exceptionClassifier = new SimpleExitStatusExceptionClassifier(); - - // default to checking current thread for interruption. - private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy(); - - private ItemFailureHandler failureLog = new DefaultItemFailureHandler(); - - private StreamManager streamManager; - - private ItemReader itemReader; - - private Chunker chunker; - - private ItemWriter itemWriter; - - private Dechunker dechunker; - - private ItemSkipPolicy itemSkipPolicy; - - private RetryTemplate retryTemplate = new RetryTemplate(); - - private RetryPolicy retryPolicy = new NeverRetryPolicy(); - - private int chunkSize; - - public void setChunkSize(int chunkSize) { - this.chunkSize = chunkSize; - } - - /** - * Public setter for the {@link StreamManager}. This will be used to create - * the {@link StepContext}, and hence any component that is a - * {@link ItemStream} and in step scope will be registered with the service. - * The {@link StepContext} is then a source of aggregate statistics for the - * step. - * - * @param streamManager - * the {@link StreamManager} to set. Default is a - * {@link SimpleStreamManager}. - */ - public void setStreamManager(StreamManager streamManager) { - this.streamManager = streamManager; - } - - public void setFailureLog(ItemFailureHandler failureLog) { - this.failureLog = failureLog; - } - - /** - * Injected strategy for storage and retrieval of persistent step - * information. Mandatory property. - * - * @param jobRepository - */ - public void setJobRepository(JobRepository jobRepository) { - this.jobRepository = jobRepository; - } - - /** - * The {@link RepeatOperations} to use for the outer loop of the batch - * processing. Should be set up by the caller through a factory. Defaults to - * a plain {@link RepeatTemplate}. - * - * @param stepOperations - * a {@link RepeatOperations} instance. - */ - public void setStepOperations(RepeatOperations stepOperations) { - this.stepOperations = stepOperations; - } - - /** - * Setter for the {@link StepInterruptionPolicy}. The policy is used to - * check whether an external request has been made to interrupt the job - * execution. - * - * @param interruptionPolicy - * a {@link StepInterruptionPolicy} - */ - public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) { - this.interruptionPolicy = interruptionPolicy; - } - - /** - * Setter for the {@link ExitStatusExceptionClassifier} that will be used to - * classify any exception that causes a job to fail. - * - * @param exceptionClassifier - */ - public void setExceptionClassifier( - ExitStatusExceptionClassifier exceptionClassifier) { - this.exceptionClassifier = exceptionClassifier; - } - - /** - * @param itemReader - */ - public void setItemReader(ItemReader itemReader) { - this.itemReader = itemReader; - } - - public void setRetryPolicy(RetryPolicy retryPolicy) { - this.retryPolicy = retryPolicy; - } - - /** - * @param itemWriter - */ - public void setItemWriter(ItemWriter itemWriter) { - this.itemWriter = itemWriter; - } - - public void setChunker(Chunker chunker) { - this.chunker = chunker; - } - - public void setDechunker(Dechunker dechunker) { - this.dechunker = dechunker; - } - - /** - * Set the skip policy. If set, it will be used for both reading and - * writing. - * - * @param itemSkipPolicy - */ - public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { - this.itemSkipPolicy = itemSkipPolicy; - } - - /** - * Check mandatory properties (reader and writer). - * - * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() - */ - public void afterPropertiesSet() throws Exception { - // This is currently a little bit funky, I don't want to require a - // chunker or - // dechunker to be wired in, since the developer should really only be - // wiring in a - // ItemReader and ItemWriter, a namespace should take care of the issue - // though. - if (chunker == null) { - chunker = new ItemChunker(itemReader); - if (itemSkipPolicy != null) { - ((ItemChunker) chunker).setItemSkipPolicy(itemSkipPolicy); - } - } - - if (dechunker == null) { - dechunker = new ItemDechunker(itemWriter); - if (itemSkipPolicy != null) { - ((ItemChunker) dechunker).setItemSkipPolicy(itemSkipPolicy); - } - } - - Assert.notNull(jobRepository, "JobRepository must not be null"); - - this.retryTemplate.setRetryPolicy(retryPolicy); - } - - /** - * Process the step and update its context so that progress can be monitored - * by the caller. The step is broken down into chunks, each one executing in - * a transaction. The step and its execution and execution context are all - * given an up to date {@link BatchStatus}, and the {@link JobRepository} - * is used to store the result. Various reporting information are also added - * to the current context (the {@link RepeatContext} governing the step - * execution, which would normally be available to the caller somehow - * through the step's {@link StepContext}.
- * - * @throws JobInterruptedException - * if the step or a chunk is interrupted - * @throws RuntimeException - * if there is an exception during a chunk execution - * @see StepExecutor#execute(StepExecution) - */ - public void execute(final StepExecution stepExecution) - throws BatchCriticalException, JobInterruptedException { - - JobInstance jobInstance = stepExecution.getJobExecution().getJobInstance(); - String stepName = stepExecution.getStepName(); - StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, this); - - boolean isRestart = jobRepository.getStepExecutionCount(jobInstance, this) > 0 ? true : false; - - ExitStatus status = ExitStatus.FAILED; - - final int chunkSize = this.chunkSize; - try { - stepExecution.setStartTime(new Date(System.currentTimeMillis())); - // We need to save the step execution right away, before we start - // using its ID. It would be better to make the creation atomic in - // the caller. - updateStatus(stepExecution, BatchStatus.STARTED); - - StepContext parentStepContext = StepSynchronizationManager - .getContext(); - final StepContext stepContext = new SimpleStepContext( - stepExecution, parentStepContext, streamManager); - StepSynchronizationManager.register(stepContext); - possiblyRegisterStreams(stepExecution); - // Add the job identifier so that it can be used to identify - // the conversation in StepScope - stepContext.setAttribute(StepScope.ID_KEY, stepExecution - .getJobExecution().getId()); - - final boolean saveExecutionContext = isSaveExecutionContext(); - - streamManager.open(stepExecution); - - if (saveExecutionContext && isRestart - && lastStepExecution != null) { - stepExecution.setExecutionContext(lastStepExecution - .getExecutionContext()); - streamManager.restoreFrom(stepExecution, stepExecution - .getExecutionContext()); - } - - status = stepOperations.iterate(new RepeatCallback() { - - public ExitStatus doInIteration(final RepeatContext context) - throws Exception { - - // Before starting a new transaction, check for - // interruption. - interruptionPolicy.checkInterrupted(context); - - final StepContribution contribution = stepExecution - .createStepContribution(); - - ChunkingResult chunkingResult = chunker.chunk(chunkSize, - contribution); - - if (chunkingResult == null) { - return ExitStatus.FINISHED; - } - - final Chunk chunk = chunkingResult.getChunk(); - for(Iterator it = chunkingResult.getExceptions().iterator();it.hasNext();){ - failureLog.handleReadFailure((Exception)it.next()); - } - - ExitStatus result = (ExitStatus) retryTemplate - .execute(new RetryCallback() { - - public Object doWithRetry(RetryContext context) - throws Throwable { - return processChunk(contribution, stepExecution, - stepContext, chunk); - } - }); - - // Check for interruption after transaction as well, so that - // the interrupted exception is correctly propagated up to - // caller - interruptionPolicy.checkInterrupted(context); - - return result; - - } - }); - - updateStatus(stepExecution, BatchStatus.COMPLETED); - } catch (RuntimeException e) { - - // classify exception so an exit code can be stored. - status = exceptionClassifier.classifyForExitCode(e); - if (e.getCause() instanceof JobInterruptedException) { - updateStatus(stepExecution, BatchStatus.STOPPED); - throw (JobInterruptedException) e.getCause(); - } else if (e instanceof ResetFailedException) { - updateStatus(stepExecution, BatchStatus.UNKNOWN); - throw (ResetFailedException) e; - } else { - updateStatus(stepExecution, BatchStatus.FAILED); - throw e; - } - - } finally { - stepExecution.setExitStatus(status); - stepExecution.setEndTime(new Date(System.currentTimeMillis())); - try { - jobRepository.saveOrUpdate(stepExecution); - streamManager.close(stepExecution); - } catch (Exception e) { - logger - .error( - "Failed to update step execution: probably fatal, so there is already an exception on the stack.", - e); - } finally { - // clear any registered synchronizations - StepSynchronizationManager.close(); - - } - } - - } - - /** - * - */ - private void possiblyRegisterStreams(Object key) { - - streamManager.register(key, chunker); - streamManager.register(key, dechunker); - } - - /** - * Execute a bunch of identical business logic operations all within a - * transaction. - * - * @param stepExecution - * the current execution in which to process the chunk in. - * @param chunk - * to be processed. - * @param stepContext - * the current step context. - * @return true if there is more data to process. - */ - ExitStatus processChunk(final StepContribution contribution,StepExecution stepExecution, - StepContext stepContext, final Chunk chunk) { - - TransactionStatus transaction = streamManager - .getTransaction(stepExecution); - - try { - - DechunkingResult dechunkingResult = dechunker.dechunk(chunk, - contribution); - for(Iterator it = dechunkingResult.getExceptions().iterator(); it.hasNext();){ - WriteFailureException exception = (WriteFailureException)it.next(); - failureLog.handleWriteFailure(exception.getItem(), (Exception)exception.getCause()); - } - - // TODO: check that stepExecution can - // aggregate these contributions if they - // come in asynchronously. - ExecutionContext statistics = streamManager - .getExecutionContext(stepExecution); - contribution.setExecutionContext(statistics); - contribution.incrementCommitCount(); - - // If the step operations are asynchronous then we need - // to synchronize changes to the step execution (at a - // minimum). - synchronized (stepExecution) { - - // Apply the contribution to the step - // only if chunk was successful - stepExecution.apply(contribution); - - if (isSaveExecutionContext()) { - stepExecution.setExecutionContext(statistics); - } - jobRepository.saveOrUpdate(stepExecution); - - } - - streamManager.commit(transaction); - - return ExitStatus.CONTINUABLE; - - } catch (Throwable t) { - /* - * Any exception thrown within the transaction template will - * automatically cause the transaction to rollback. We need to - * include exceptions during an attempted commit (e.g. Hibernate - * flush) so this catch block comes outside the transaction. - */ - synchronized (stepExecution) { - stepExecution.rollback(); - } - try { - streamManager.rollback(transaction); - } catch (ResetFailedException e) { - // The original Throwable cause is in danger of - // being lost here, so we log the reset - // failure and re-throw with cause of the rollback. - logger - .error( - "Encountered reset error on rollback: " - + "one of the streams may be in an inconsistent state, " - + "so this step should not proceed", e); - throw new ResetFailedException( - "Encountered reset error on rollback. " - + "Consult logs for the cause of the reet failure. " - + "The cause of the original rollback is incuded here.", - t); - } - if (t instanceof RuntimeException) { - throw (RuntimeException) t; - } else { - throw new RuntimeException(t); - } - } - - } - - /* - * Convenience method to update the status in all relevant places. - * - * @param stepInstance the current step @param stepExecution the current - * stepExecution @param status the status to set - */ - private void updateStatus(StepExecution stepExecution, BatchStatus status) { - stepExecution.setStatus(status); - jobRepository.saveOrUpdate(stepExecution); - } -} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemChunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemChunker.java deleted file mode 100644 index 061f1c992..000000000 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemChunker.java +++ /dev/null @@ -1,131 +0,0 @@ -/* - * 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.support; - -import java.util.ArrayList; -import java.util.List; - -import org.springframework.batch.core.domain.Chunk; -import org.springframework.batch.core.domain.Chunker; -import org.springframework.batch.core.domain.ChunkingResult; -import org.springframework.batch.core.domain.ItemSkipPolicy; -import org.springframework.batch.core.domain.StepContribution; -import org.springframework.batch.io.exception.ReadFailureException; -import org.springframework.batch.item.ExecutionContext; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.exception.StreamException; -import org.springframework.util.Assert; - -/** - * Implementation of the {@link Chunker} interface that creates chunks from - * an {@link ItemReader}. It does not buffer chunks. If - * the underlying reader has been rolled back, and - * - * @author Ben Hale - * @author Lucas Ward - */ -public class ItemChunker implements Chunker { - - private final ItemReader itemReader; - private long chunkCounter = 0; - - private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); - - public ItemChunker(ItemReader itemReader) { - Assert.notNull(itemReader, "ItemReader must not be null"); - this.itemReader = itemReader; - } - - public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { - this.itemSkipPolicy = itemSkipPolicy; - } - - public ChunkingResult chunk(int size, StepContribution stepContribution) throws ReadFailureException { - Assert.isTrue(size > 0, "Chunk size must be greater than 0"); - - itemReader.mark(); - int counter = 0; - List items = new ArrayList(size); - List exceptions = new ArrayList(); - - Object item; - while (counter < size) { - try { - item = itemReader.read(); - if (item == null) { - break; - } - items.add(item); - counter++; - } catch (Exception ex) { - exceptions.add(ex); - if(!itemSkipPolicy.shouldSkip(ex, stepContribution)){ - itemReader.reset(); - rethrow(ex); - } - } - } - - if (items.size() == 0) { - return null; - } - - return new ChunkingResult(new Chunk(getChunkId(), items), exceptions); - } - - 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++); - } - - //These methods are temporary hacks until something can be done - //about the ItemStream interface. - public void close() throws StreamException { - if(itemReader instanceof ItemStream){ - ((ItemStream)itemReader).close(); - } - } - - public void open() throws StreamException { - if(itemReader instanceof ItemStream){ - ((ItemStream)itemReader).open(); - } - } - - public void restoreFrom(ExecutionContext context) { - if(itemReader instanceof ItemStream){ - ((ItemStream)itemReader).restoreFrom(context); - } - } - - public ExecutionContext getExecutionContext() { - if(itemReader instanceof ItemStream){ - return ((ItemStream)itemReader).getExecutionContext(); - } - - return new ExecutionContext(); - } - -} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemDechunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemDechunker.java deleted file mode 100644 index e5cc044ac..000000000 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemDechunker.java +++ /dev/null @@ -1,121 +0,0 @@ -/* - * 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.support; - -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; - -import org.springframework.batch.core.domain.Chunk; -import org.springframework.batch.core.domain.Dechunker; -import org.springframework.batch.core.domain.DechunkingResult; -import org.springframework.batch.core.domain.ItemSkipPolicy; -import org.springframework.batch.core.domain.StepContribution; -import org.springframework.batch.io.exception.WriteFailureException; -import org.springframework.batch.item.ExecutionContext; -import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.exception.StreamException; -import org.springframework.util.Assert; - -/** - * Implementation of the {@link Dechunker} interface that passes items to an - * {@link ItemWriter} one at a time. - * - * @author Lucas Ward - * - */ -public class ItemDechunker implements Dechunker { - - private final ItemWriter itemWriter; - private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); - - public ItemDechunker(ItemWriter itemWriter) { - this.itemWriter = itemWriter; - } - - /* (non-Javadoc) - * @see org.springframework.batch.core.domain.Dechunker#dechunk(org.springframework.batch.core.domain.Chunk) - */ - public DechunkingResult dechunk(Chunk chunk, StepContribution stepContribution) throws Exception { - - Assert.notNull(chunk, "Chunk must not be null"); - Assert.notNull(stepContribution, "StepExecution must not be null"); - List skippedItems = new ArrayList(); - for(Iterator it = chunk.getItems().iterator(); it.hasNext();){ - - Object item = it.next(); - try{ - itemWriter.write(item); - } - catch(Exception ex){ - if(itemSkipPolicy.shouldSkip(ex, stepContribution)){ - stepContribution.incrementSkipCount(); - skippedItems.add(new WriteFailureException(ex, item)); - } - else{ - itemWriter.clear(); - rethrow(ex); - } - } - } - - itemWriter.flush(); - return new DechunkingResult(true, chunk.getId(), skippedItems); - } - - public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { - this.itemSkipPolicy = itemSkipPolicy; - } - - private void rethrow(Exception ex){ - if(ex instanceof RuntimeException){ - throw (RuntimeException)ex; - } - else{ - throw new RuntimeException("Error encountered while dechunking", ex); - } - } - - //This a hack until the ItemStream interface can be updated. - public void close() throws StreamException { - if(itemWriter instanceof ItemStream){ - ((ItemStream)itemWriter).close(); - } - } - - public void open() throws StreamException { - if(itemWriter instanceof ItemStream){ - ((ItemStream)itemWriter).open(); - } - } - - public void restoreFrom(ExecutionContext context) { - if(itemWriter instanceof ItemStream){ - ((ItemStream)itemWriter).restoreFrom(context); - } - } - - public ExecutionContext getExecutionContext() { - if(itemWriter instanceof ItemStream){ - return ((ItemStream)itemWriter).getExecutionContext(); - } - else{ - return new ExecutionContext(); - } - } - -} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ChunkedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ChunkedStepTests.java deleted file mode 100644 index 17947af69..000000000 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ChunkedStepTests.java +++ /dev/null @@ -1,563 +0,0 @@ -/* - * 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; - -import java.util.ArrayList; -import java.util.Arrays; - -import junit.framework.TestCase; - -import org.easymock.MockControl; -import org.springframework.batch.core.domain.BatchStatus; -import org.springframework.batch.core.domain.JobExecution; -import org.springframework.batch.core.domain.JobInstance; -import org.springframework.batch.core.domain.JobParameters; -import org.springframework.batch.core.domain.JobSupport; -import org.springframework.batch.core.domain.StepExecution; -import org.springframework.batch.core.domain.StepSupport; -import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.execution.scope.StepScope; -import org.springframework.batch.execution.scope.StepSynchronizationManager; -import org.springframework.batch.execution.step.support.ItemChunker; -import org.springframework.batch.execution.step.support.ItemDechunker; -import org.springframework.batch.execution.step.support.JobRepositorySupport; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.reader.AbstractItemReader; -import org.springframework.batch.item.reader.ListItemReader; -import org.springframework.batch.item.stream.SimpleStreamManager; -import org.springframework.batch.item.writer.AbstractItemWriter; -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.interceptor.RepeatListenerSupport; -import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.support.transaction.ResourcelessTransactionManager; - -/** - * - * - * @author Lucas Ward - * - */ -public class ChunkedStepTests extends TestCase { - - ArrayList processed = new ArrayList(); - - ItemWriter processor = new AbstractItemWriter() { - public void write(Object data) throws Exception { - processed.add((String) data); - } - }; - - private ChunkedStep chunkedStep; - - private JobInstance jobInstance; - - private ResourcelessTransactionManager transactionManager; - - private JobExecution jobExecutionContext; - private StepExecution stepExecution; - - private ItemReader getReader(String[] args) { - return new ListItemReader(Arrays.asList(args)); - } - - - - private ChunkedStep getStep(String[] strings) throws Exception { - ChunkedStep step = new ChunkedStep(); - step.setItemWriter(processor); - step.setItemReader(getReader(strings)); - step.setJobRepository(new JobRepositorySupport()); - step.setStreamManager(new SimpleStreamManager(transactionManager)); - step.afterPropertiesSet(); - return step; - } - - protected void setUp() throws Exception { - transactionManager = new ResourcelessTransactionManager(); - chunkedStep = getStep(new String[] { "foo", "bar", "spam" }); - - // Only process one item: - chunkedStep.setChunkSize(1); - - jobInstance = new JobInstance(new Long(0), new JobParameters(), new JobSupport("FOO")); - SimpleStreamManager streamManager = new SimpleStreamManager(transactionManager); - streamManager.setUseClassNameAsPrefix(false); - chunkedStep.setStreamManager(streamManager); - chunkedStep.setJobRepository(new JobRepositorySupport()); - - jobExecutionContext = new JobExecution(jobInstance); - stepExecution = new StepExecution(new StepSupport("testStep"), jobExecutionContext); - } - - public void testStepExecutor() throws Exception { - - chunkedStep.execute(stepExecution); - assertEquals(3, processed.size()); - assertEquals(stepExecution.getStatus(), BatchStatus.COMPLETED); - } - - public void testStepContextInitialized() throws Exception { - - final JobExecution jobExecution = new JobExecution(jobInstance); - final StepExecution stepExecution = new StepExecution(new StepSupport("testStep"), jobExecution); - - chunkedStep.setChunker(new ItemChunker(new AbstractItemReader() { - int counter = 0; - public Object read() throws Exception { - assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); - if(counter++ < 2){ - return "foo"; - } - else{ - return null; - } - } - })); - - chunkedStep.execute(stepExecution); - assertEquals(2, processed.size()); - } - - public void testStepContextInitializedBeforeTasklet() throws Exception { - - RepeatTemplate template = new RepeatTemplate(); - - // Only process one chunk: - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - chunkedStep.setStepOperations(template); - - final JobExecution jobExecution = new JobExecution(jobInstance); - jobExecution.setId(new Long(1)); - final StepExecution stepExecution = new StepExecution(new StepSupport("testStep"), jobExecution); - - template.setListener(new RepeatListenerSupport() { - public void open(RepeatContext context) { - assertNotNull(StepSynchronizationManager.getContext().getStepExecution()); - assertEquals(stepExecution, StepSynchronizationManager.getContext().getStepExecution()); - // StepScope can obtain id information.... - assertNotNull(StepSynchronizationManager.getContext().getAttribute(StepScope.ID_KEY)); - } - }); - - chunkedStep.execute(stepExecution); - assertEquals(1, processed.size()); - - } - - public void testRepository() throws Exception { - - MockControl repoControl = MockControl.createControl(JobRepository.class); - JobRepository repository = (JobRepository)repoControl.getMock(); - chunkedStep.setJobRepository(repository); - -// StepInstance step = new StepInstance(new Long(1)); -// JobExecution jobExecutionContext = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); - - repository.getLastStepExecution(jobInstance, chunkedStep); - repoControl.setReturnValue(stepExecution); - repository.getStepExecutionCount(jobInstance, chunkedStep); - repoControl.setReturnValue(0); - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repository.saveOrUpdate(stepExecution); - repoControl.replay(); - chunkedStep.execute(stepExecution); - assertEquals(3, processed.size()); - repoControl.verify(); - assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); - } - - //ReadFailures (meaning an item couldn't be skipped) should cause the job to - //fail. - public void testReadFailure() { - - ItemReader itemReader = new AbstractItemReader() { - int counter = 0; - public Object read() throws Exception { - - counter++; - - if (counter > 1) { - throw new RuntimeException(); - } - - return "foo"; - } - - }; - - chunkedStep.setChunker(new ItemChunker(itemReader)); - - try { - chunkedStep.execute(stepExecution); - fail(); - } - catch (Exception ex) { - assertEquals( 1, processed.size()); - assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); - } - - } - - public void testWriterFailure(){ - - ItemWriter itemWriter = new AbstractItemWriter(){ - public void write(Object item) throws Exception { - throw new RuntimeException(); - } - }; - - chunkedStep.setDechunker(new ItemDechunker(itemWriter)); - - try{ - chunkedStep.execute(stepExecution); - fail(); - } - catch(Exception ex){ - //it should rollback three times since that's default behavior for a retry template. - assertEquals(new Integer(1), stepExecution.getRollbackCount()); - } - } - - public void testExitCodeDefaultClassification() throws Exception { - - ItemReader itemReader = new AbstractItemReader() { - int counter = 0; - public Object read() throws Exception { - counter++; - - if (counter == 1) { - throw new RuntimeException(); - } - - return ExitStatus.CONTINUABLE; - } - - }; - - chunkedStep.setItemReader(itemReader); - - try { - chunkedStep.execute(stepExecution); - } - catch (Exception ex) { - ExitStatus status = stepExecution.getExitStatus(); - assertEquals("FATAL_EXCEPTION", status.getExitCode()); - assertFalse(status.isContinuable()); - } - } - - /* - * make sure a job that has never been executed before, but does have - * saveExecutionAttributes = true, doesn't have restoreFrom called on it. - */ - // I can't get this test to pass, I think there is something wrong with the code I - // pulled from SimpleStepExecutor -// public void testNonRestartedJob() throws Exception { -// MockRestartableItemReader reader = new MockRestartableItemReader(); -// chunkedStep.setItemReader(reader); -// chunkedStep.setSaveExecutionAttributes(true); -// -// chunkedStep.execute(stepExecution); -// -// assertFalse(reader.isRestoreFromCalled()); -// assertTrue(reader.isGetExecutionAttributesCalled()); -// } -// -// /* -// * make sure a job that has been executed before, and is therefore being -// * restarted, is restored. -// */ -// public void testRestartedJob() throws Exception { -// StepInstance step = new StepInstance(new Long(1)); -// step.setStepExecutionCount(1); -// MockRestartableTasklet tasklet = new MockRestartableTasklet(); -// chunkedStep.setItemReader(tasklet); -// stepConfiguration.setSaveExecutionAttributes(true); -// JobExecution jobExecutionContext = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); -// -// stepExecution -// .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); -// step.setLastExecution(stepExecution); -// chunkedStep.execute(stepExecution); -// -// assertTrue(tasklet.isRestoreFromCalled()); -// assertTrue(tasklet.isRestoreFromCalledWithSomeContext()); -// assertTrue(tasklet.isGetExecutionAttributesCalled()); -// } -// -// /* -// * Test that a job that is being restarted, but has saveExecutionAttributes -// * set to false, doesn't have restore or getExecutionAttributes called on -// * it. -// */ -// public void testNoSaveExecutionAttributesRestartableJob() { -// StepInstance step = new StepInstance(new Long(1)); -// step.setStepExecutionCount(1); -// MockRestartableTasklet tasklet = new MockRestartableTasklet(); -// stepConfiguration.setItemReader(tasklet); -// stepConfiguration.setSaveExecutionAttributes(false); -// JobExecution jobExecutionContext = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); -// -// try { -// chunkedStep.execute(stepExecution); -// } -// catch (Throwable t) { -// fail(); -// } -// -// assertFalse(tasklet.isRestoreFromCalled()); -// assertFalse(tasklet.isGetExecutionAttributesCalled()); -// } -// -// /* -// * Even though the job is restarted, and saveExecutionAttributes is true, -// * nothing will be restored because the Tasklet does not implement -// * Restartable. -// */ -// public void testRestartJobOnNonRestartableTasklet() throws Exception { -// StepInstance step = new StepInstance(new Long(1)); -// step.setStepExecutionCount(1); -// stepConfiguration.setItemReader(new ItemReader() { -// public Object read() throws Exception { -// return ExitStatus.FINISHED; -// } -// }); -// stepConfiguration.setSaveExecutionAttributes(true); -// JobExecution jobExecution = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecution); -// -// chunkedStep.execute(stepExecution); -// } -// -//// public void testApplyConfigurationWithExceptionHandler() throws Exception { -//// AbstractStep stepConfiguration = new SimpleStep("foo"); -//// final List list = new ArrayList(); -//// chunkedStep.setStepOperations(new RepeatTemplate() { -//// public void setExceptionHandler(ExceptionHandler exceptionHandler) { -//// list.add(exceptionHandler); -//// } -//// }); -//// stepConfiguration.setExceptionHandler(new DefaultExceptionHandler()); -//// chunkedStep.applyConfiguration(stepConfiguration); -//// assertEquals(1, list.size()); -//// } -//// -//// public void testApplyConfigurationWithZeroSkipLimit() throws Exception { -//// AbstractStep stepConfiguration = new SimpleStep("foo"); -//// stepConfiguration.setSkipLimit(0); -//// final List list = new ArrayList(); -//// chunkedStep.setStepOperations(new RepeatTemplate() { -//// public void setExceptionHandler(ExceptionHandler exceptionHandler) { -//// list.add(exceptionHandler); -//// } -//// }); -//// chunkedStep.applyConfiguration(stepConfiguration); -//// assertEquals(0, list.size()); -//// } -//// -//// public void testApplyConfigurationWithNonZeroSkipLimit() throws Exception { -//// AbstractStep stepConfiguration = new SimpleStep("foo"); -//// stepConfiguration.setSkipLimit(1); -//// final List list = new ArrayList(); -//// chunkedStep.setStepOperations(new RepeatTemplate() { -//// public void setExceptionHandler(ExceptionHandler exceptionHandler) { -//// list.add(exceptionHandler); -//// } -//// }); -//// chunkedStep.applyConfiguration(stepConfiguration); -//// assertEquals(1, list.size()); -//// } -// -// public void testStreamManager() throws Exception { -// StepInstance step = new StepInstance(new Long(1)); -// step.setStepExecutionCount(1); -// stepConfiguration.setItemReader(new ItemReader() { -// public Object read() throws Exception { -// return ExitStatus.FINISHED; -// } -// }); -// stepConfiguration.setSaveExecutionAttributes(true); -// JobExecution jobExecution = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecution); -// -// assertEquals(false, stepExecution.getExecutionAttributes().containsKey("foo")); -// -// final Map map = new HashMap(); -// chunkedStep.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) { -// public ExecutionAttributes getExecutionAttributes(Object key) { -// // TODO Auto-generated method stub -// return new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar")); -// } -// }); -// -// chunkedStep.execute(stepExecution); -// -// // At least once in that process the statistics service was asked for -// // statistics... -// assertEquals("bar", stepExecution.getExecutionAttributes().getString("foo")); -// // ...but nothing was registered because nothing with step scoped. -// assertEquals(0, map.size()); -// } -// -// public void testStatusForInterruptedException() { -// -// StepInterruptionPolicy interruptionPolicy = new StepInterruptionPolicy() { -// -// public void checkInterrupted(RepeatContext context) throws JobInterruptedException { -// throw new JobInterruptedException(""); -// } -// }; -// -// chunkedStep.setInterruptionPolicy(interruptionPolicy); -// -// ItemReader itemReader = new ItemReader() { -// -// public Object read() throws Exception { -// int counter = 0; -// counter++; -// -// if (counter == 1) { -// throw new RuntimeException(); -// } -// -// return ExitStatus.CONTINUABLE; -// } -// -// }; -// -// chunkedStep.setItemReader(itemReader); -// -// StepInstance step = new StepInstance(new Long(1)); -// JobExecution jobExecutionContext = new JobExecution(jobInstance); -// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); -// -// stepExecution -// .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); -// step.setLastExecution(stepExecution); -// -// try { -// chunkedStep.execute(stepExecution); -// fail("Expected StepInterruptedException"); -// } -// catch (JobInterruptedException ex) { -// assertEquals(BatchStatus.STOPPED, stepExecution.getStatus()); -// String msg = stepExecution.getExitStatus().getExitDescription(); -// assertTrue("Message does not contain JobInterruptedException: " + msg, msg -// .contains("JobInterruptedException")); -// } -// } -// -// public void testStatusForResetFailedException() throws Exception { -// -// ItemReader itemReader = new ItemReader() { -// public Object read() throws Exception { -// // Trigger a rollback -// throw new RuntimeException("Foo"); -// } -// }; -// chunkedStep.setItemReader(itemReader); -// chunkedStep.setStreamManager(new SimpleStreamManager(transactionManager) { -// public void rollback(TransactionStatus status) { -// super.rollback(status); -// // Simulate failure on rollback when stream resets -// throw new ResetFailedException("Bar"); -// } -// }); -// -// StepInstance step = new StepInstance(new Long(1)); -// JobExecution jobExecutionContext = jobInstance.createJobExecution(); -// StepExecution stepExecution = new StepExecution(step, jobExecutionContext); -// -// stepExecution -// .setExecutionAttributes(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); -// step.setLastExecution(stepExecution); -// -// try { -// chunkedStep.execute(stepExecution); -// fail("Expected ResetFailedException"); -// } -// catch (ResetFailedException ex) { -// assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus()); -// String msg = stepExecution.getExitStatus().getExitDescription(); -// assertTrue("Message does not contain ResetFailedException: " + msg, msg.contains("ResetFailedException")); -// // The original rollback was caused by this one: -// assertEquals("Foo", ex.getCause().getMessage()); -// } -// } -// -// private class MockRestartableItemReader extends ItemStreamSupport implements ItemReader { -// -// private boolean getExecutionAttributesCalled = false; -// -// private boolean restoreFromCalled = false; -// -// private boolean restoreFromCalledWithSomeContext = false; -// -// private int counter = 0; -// -// public Object read() throws Exception { -// StepSynchronizationManager.getContext().setAttribute("TASKLET_TEST", this); -// counter++; -// if(counter > 4){ -// return "item"; -// } -// else{ -// return null; -// } -// } -// -// public boolean isRestoreFromCalledWithSomeContext() { -// return restoreFromCalledWithSomeContext; -// } -// -// public ExecutionContext getExecutionContext() { -// getExecutionAttributesCalled = true; -// return new ExecutionContext(PropertiesConverter.stringToProperties("spam=bucket")); -// } -// -// public void restoreFrom(ExecutionContext data) { -// restoreFromCalled = true; -// restoreFromCalledWithSomeContext = data.getProperties().size() > 0; -// } -// -// public boolean isGetExecutionAttributesCalled() { -// return getExecutionAttributesCalled; -// } -// -// public boolean isRestoreFromCalled() { -// return restoreFromCalled; -// } -// -// public void open() throws StreamException { -// } -// -// public void close() throws StreamException { -// } -// -// } - -} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemChunkerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemChunkerTests.java deleted file mode 100644 index a8f7e9a37..000000000 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemChunkerTests.java +++ /dev/null @@ -1,112 +0,0 @@ -/* - * 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.support; - -import junit.framework.TestCase; - -import org.springframework.batch.core.domain.ChunkingResult; -import org.springframework.batch.core.domain.ItemSkipPolicy; -import org.springframework.batch.core.domain.JobExecution; -import org.springframework.batch.core.domain.JobInstance; -import org.springframework.batch.core.domain.JobParameters; -import org.springframework.batch.core.domain.JobSupport; -import org.springframework.batch.core.domain.StepContribution; -import org.springframework.batch.core.domain.StepExecution; -import org.springframework.batch.core.domain.StepSupport; -import org.springframework.batch.execution.step.support.ItemChunker; - -public class ItemChunkerTests extends TestCase { - - StepContribution stepContribution; - - protected void setUp() throws Exception { - super.setUp(); - JobExecution jobExecution = new JobExecution(new JobInstance(new Long(1), new JobParameters(), new JobSupport("jobName"))); - StepExecution execution = new StepExecution(new StepSupport("stepName"), jobExecution); - stepContribution = execution.createStepContribution(); - } - - public void testSizeNegative() { - try { - MockItemReader itemReader = new MockItemReader(10); - ItemChunker chunkReader = new ItemChunker(itemReader); - chunkReader.chunk(-1, stepContribution); - fail(); - } - catch (IllegalArgumentException e) { - } - } - - public void testSizeZero() { - try { - MockItemReader itemReader = new MockItemReader(10); - ItemChunker chunkReader = new ItemChunker(itemReader); - chunkReader.chunk(0, stepContribution); - fail(); - } - catch (IllegalArgumentException e) { - } - } - - public void testSizePositive() { - MockItemReader itemReader = new MockItemReader(10); - ItemChunker chunkReader = new ItemChunker(itemReader); - ChunkingResult chunkingResult = chunkReader.chunk(10, stepContribution); - assertEquals(10, chunkingResult.getChunk().getItems().size()); - } - - public void testIncompleteChunk() { - MockItemReader itemReader = new MockItemReader(5); - ItemChunker chunkReader = new ItemChunker(itemReader); - ChunkingResult chunkingResult = chunkReader.chunk(10, stepContribution); - assertEquals(5, chunkingResult.getChunk().getItems().size()); - } - - public void testPolicyNoContinue() { - MockItemReader itemReader = new MockItemReader(1); - itemReader.setFail(true); - ItemChunker chunkReader = new ItemChunker(itemReader); - chunkReader.setItemSkipPolicy(new StubReadFailurePolicy(true)); - try { - chunkReader.chunk(10, stepContribution); - fail(); - } - catch (RuntimeException e) { - } - } - - public void testPolicyContinueWithFailure() { - MockItemReader itemReader = new MockItemReader(1); - itemReader.setFail(true); - ItemChunker chunkReader = new ItemChunker(itemReader); - chunkReader.setItemSkipPolicy(new StubReadFailurePolicy(false)); - ChunkingResult chunkingResult = chunkReader.chunk(1, stepContribution); - assertEquals(1, chunkingResult.getChunk().getItems().size()); - } - - private class StubReadFailurePolicy implements ItemSkipPolicy { - - private final boolean fail; - - public StubReadFailurePolicy(boolean fail) { - this.fail = fail; - } - - public boolean shouldSkip(Exception ex, StepContribution stepContribution) { - return !fail; - } - } -} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemDechunkerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemDechunkerTests.java deleted file mode 100644 index c0aeb50ee..000000000 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemDechunkerTests.java +++ /dev/null @@ -1,105 +0,0 @@ -/* - * 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.support; - -import java.util.ArrayList; -import java.util.List; - -import org.easymock.MockControl; -import org.springframework.batch.core.domain.Chunk; -import org.springframework.batch.core.domain.DechunkingResult; -import org.springframework.batch.core.domain.StepContribution; -import org.springframework.batch.core.domain.StepExecution; -import org.springframework.batch.core.domain.StepSupport; -import org.springframework.batch.execution.step.support.AlwaysSkipItemSkipPolicy; -import org.springframework.batch.execution.step.support.ItemDechunker; -import org.springframework.batch.io.exception.WriteFailureException; -import org.springframework.batch.item.ItemWriter; - -import junit.framework.TestCase; - -/** - * @author Lucas Ward - * - */ -public class ItemDechunkerTests extends TestCase { - - private ItemDechunker dechunker; - private StepContribution stepContribution; - private Chunk chunk; - private ItemWriter itemWriter; - private MockControl writerControl = MockControl.createControl(ItemWriter.class); - - - /* (non-Javadoc) - * @see junit.framework.TestCase#setUp() - */ - protected void setUp() throws Exception { - super.setUp(); - - itemWriter = (ItemWriter)writerControl.getMock(); - StepExecution execution = new StepExecution(new StepSupport("stepName"),null); - stepContribution = execution.createStepContribution(); - dechunker = new ItemDechunker(itemWriter); - List items = new ArrayList(); - items.add("1"); - items.add("2"); - chunk = new Chunk(new Long(1),items); - } - - - public void testNormalProcessing() throws Exception{ - - itemWriter.write("1"); - itemWriter.write("2"); - itemWriter.flush(); - writerControl.replay(); - dechunker.dechunk(chunk, stepContribution); - writerControl.verify(); - } - - public void testSkip() throws Exception{ - - dechunker.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); - itemWriter.write("1"); - itemWriter.write("2"); - writerControl.setThrowable(new Exception()); - itemWriter.flush(); - writerControl.replay(); - DechunkingResult result = dechunker.dechunk(chunk, stepContribution); - writerControl.verify(); - List exceptions = result.getExceptions(); - assertEquals(1, exceptions.size()); - WriteFailureException exception = (WriteFailureException)exceptions.get(0); - assertEquals("2",exception.getItem()); - - } - - public void testFailure() throws Exception{ - itemWriter.write("1"); - itemWriter.write("2"); - writerControl.setThrowable(new NullPointerException()); - itemWriter.clear(); - writerControl.replay(); - try{ - dechunker.dechunk(chunk, stepContribution); - fail(); - } - catch(NullPointerException ex){ - //expected - } - } -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java index 5898d5422..f2c5b6bc3 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java @@ -20,6 +20,7 @@ import org.springframework.batch.io.Skippable; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.KeyedItemReader; import org.springframework.batch.item.exception.StreamException; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; @@ -31,7 +32,7 @@ import org.springframework.util.Assert; * * @author Dave Syer */ -public class DelegatingItemReader extends AbstractItemReader implements Skippable, InitializingBean, ItemStream { +public class DelegatingItemReader extends AbstractItemReader implements Skippable, InitializingBean, ItemStream, KeyedItemReader { private ItemReader itemReader; @@ -78,6 +79,13 @@ public class DelegatingItemReader extends AbstractItemReader implements Skippabl public void setItemReader(ItemReader source) { this.itemReader = source; } + + /* (non-Javadoc) + * @see org.springframework.batch.item.KeyedItemReader#getKey(java.lang.Object) + */ + public Object getKey(Object item) { + return item; + } public ItemReader getItemReader() { return itemReader; diff --git a/spring-batch-samples/.springBeans b/spring-batch-samples/.springBeans index 2b728a193..f7c954084 100644 --- a/spring-batch-samples/.springBeans +++ b/spring-batch-samples/.springBeans @@ -32,6 +32,7 @@ src/main/resources/jobs/parallelJob.xml src/main/resources/jobs/rollbackJob.xml src/main/resources/jobs/retrySample.xml + src/main/resources/jobs/simpleJob.xml diff --git a/spring-batch-samples/src/main/resources/jobs/compositeProcessorSampleJob.xml b/spring-batch-samples/src/main/resources/jobs/compositeProcessorSampleJob.xml index 71d752174..6742e3ec8 100644 --- a/spring-batch-samples/src/main/resources/jobs/compositeProcessorSampleJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/compositeProcessorSampleJob.xml @@ -12,7 +12,7 @@ - + diff --git a/spring-batch-samples/src/main/resources/jobs/delegatingJob.xml b/spring-batch-samples/src/main/resources/jobs/delegatingJob.xml index e41e54d52..0d3e1f13b 100644 --- a/spring-batch-samples/src/main/resources/jobs/delegatingJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/delegatingJob.xml @@ -23,7 +23,7 @@ - + diff --git a/spring-batch-samples/src/main/resources/jobs/fixedLengthImportJob.xml b/spring-batch-samples/src/main/resources/jobs/fixedLengthImportJob.xml index a48cbb777..b8ffeea32 100644 --- a/spring-batch-samples/src/main/resources/jobs/fixedLengthImportJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/fixedLengthImportJob.xml @@ -14,7 +14,7 @@ - + diff --git a/spring-batch-samples/src/main/resources/jobs/footballJob.xml b/spring-batch-samples/src/main/resources/jobs/footballJob.xml index d99adfa31..3d1aaca3a 100644 --- a/spring-batch-samples/src/main/resources/jobs/footballJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/footballJob.xml @@ -16,8 +16,8 @@ - - + + - - + + @@ -48,8 +48,8 @@ - - + + diff --git a/spring-batch-samples/src/main/resources/jobs/ibatisJob.xml b/spring-batch-samples/src/main/resources/jobs/ibatisJob.xml index a5ac282ee..05798f3ef 100644 --- a/spring-batch-samples/src/main/resources/jobs/ibatisJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/ibatisJob.xml @@ -16,7 +16,7 @@ - + - + diff --git a/spring-batch-samples/src/main/resources/jobs/multilineOrderJob.xml b/spring-batch-samples/src/main/resources/jobs/multilineOrderJob.xml index d62101512..c81930b21 100644 --- a/spring-batch-samples/src/main/resources/jobs/multilineOrderJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/multilineOrderJob.xml @@ -15,7 +15,7 @@ - + diff --git a/spring-batch-samples/src/main/resources/jobs/restartSample.xml b/spring-batch-samples/src/main/resources/jobs/restartSample.xml index afbcaf315..b86f5b166 100644 --- a/spring-batch-samples/src/main/resources/jobs/restartSample.xml +++ b/spring-batch-samples/src/main/resources/jobs/restartSample.xml @@ -15,7 +15,7 @@ - + @@ -30,7 +30,7 @@ - + diff --git a/spring-batch-samples/src/main/resources/jobs/rollbackJob.xml b/spring-batch-samples/src/main/resources/jobs/rollbackJob.xml index a75b064a6..84b888c2a 100644 --- a/spring-batch-samples/src/main/resources/jobs/rollbackJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/rollbackJob.xml @@ -17,46 +17,35 @@ - + - - - - - - - - - - - - - - + class="org.springframework.batch.retry.policy.SimpleRetryPolicy" /> + + + + + + + + - + - - - - - - - - + class="org.springframework.batch.execution.step.support.LimitCheckingItemSkipPolicy"> + + + + - + - + diff --git a/spring-batch-samples/src/main/resources/jobs/tradeJob.xml b/spring-batch-samples/src/main/resources/jobs/tradeJob.xml index 4c0ca35ef..53cedaad6 100644 --- a/spring-batch-samples/src/main/resources/jobs/tradeJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/tradeJob.xml @@ -16,7 +16,7 @@ - + @@ -32,7 +32,7 @@ p:dao-ref="tradeDao" /> - + - + - + - - - - - - - - - - - -