Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Moved chunks to a branch (called chunks)

Browse Source
This commit is contained in:
dsyer
2008-02-26 08:06:11 +00:00
parent 23ab7d3783
commit 3b48ece7e5
28 changed files with 52 additions and 1990 deletions
Download Patch File Download Diff File Expand all files Collapse all files

548
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ChunkedStep.java
View File

@@ -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;
/**
* <p>
* 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}
* </p>
*
* <p>
* 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
* </p>
*
* <p>
* 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.
* </p>
*
* <p>
* 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.</p.
*
* <p>
* 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.
* </p>
*
* @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}.<br/>
*
* @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);
}
}

131
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemChunker.java
View File

@@ -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}. <strong>It does not buffer chunks</strong>. 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();
}
}

121
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemDechunker.java
View File

@@ -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();
}
}
}

563
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ChunkedStepTests.java
View File

@@ -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 {
// }
//
// }
}

112
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemChunkerTests.java
View File

@@ -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;
}
}
}

105
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemDechunkerTests.java
View File

@@ -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
}
}
}