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

RESOLVED - issue BATCH-404: FactoryBeans for step configuration

Browse Source 
http://jira.springframework.org/browse/BATCH-404

Tidied up listener registration and integration a bit
This commit is contained in:
dsyer
2008-03-04 13:16:18 +00:00
parent 9065b6ba25
commit f73ce3553c
16 changed files with 285 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AbstractStepFactoryBean.java
View File

@@ -21,7 +21,7 @@ import org.springframework.batch.execution.step.ItemOrientedStep;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
* @author Dave Syer
*
*/
public abstract class AbstractStepFactoryBean extends AbstractFactoryBean implements BeanNameAware {
public abstract class AbstractStepFactoryBean implements FactoryBean, BeanNameAware {
private String name;
@@ -49,6 +49,8 @@ public abstract class AbstractStepFactoryBean extends AbstractFactoryBean implem
private JobRepository jobRepository;
private boolean singleton = true;
/**
*
*/
@@ -140,7 +142,12 @@ public abstract class AbstractStepFactoryBean extends AbstractFactoryBean implem
this.transactionManager = transactionManager;
}
protected Object createInstance() throws Exception {
/**
* Create a {@link Step} from the configuration provided.
*
* @see org.springframework.beans.factory.FactoryBean#getObject()
*/
public final Object getObject() throws Exception {
ItemOrientedStep step = new ItemOrientedStep(getName());
applyConfiguration(step);
return step;
@@ -156,7 +163,7 @@ public abstract class AbstractStepFactoryBean extends AbstractFactoryBean implem
Assert.notNull(getItemWriter(), "ItemWriter must be provided");
Assert.notNull(jobRepository, "JobRepository must be provided");
Assert.notNull(transactionManager, "TransactionManager must be provided");
step.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
step.setTransactionManager(transactionManager);
step.setJobRepository(jobRepository);
@@ -169,4 +176,23 @@ public abstract class AbstractStepFactoryBean extends AbstractFactoryBean implem
return Step.class;
}
/**
* Returns true by default, but in most cases a {@link Step} should not be
* treated as thread safe. Clients are recommended to create a new step for
* each job execution.
*
* @see org.springframework.beans.factory.FactoryBean#isSingleton()
*/
public boolean isSingleton() {
return this.singleton;
}
/**
* Public setter for the singleton flag.
* @param singleton the value to set. Defaults to true.
*/
public void setSingleton(boolean singleton) {
this.singleton = singleton;
}
}

50
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java
View File

@@ -39,9 +39,6 @@ import org.springframework.batch.retry.support.RetryTemplate;
* limit given by the {@link RetryPolicy}. When the retry is exhausted instead
* of the item being skipped it is handled by an {@link ItemRecoverer}.<br/>
*
* TODO: checking for null retry callback is a sucky way of determining if a
* stateful retry has been requested.
*
* @author Dave Syer
*
*/
@@ -51,6 +48,8 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
private ItemKeyGenerator itemKeyGenerator;
private ItemRecoverer itemRecoverer;
/**
* Public setter for the {@link RetryPolicy}.
* @param retryPolicy the {@link RetryPolicy} to set
@@ -73,6 +72,18 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
this.itemKeyGenerator = itemKeyGenerator;
}
/**
* Public setter for the {@link ItemRecoverer}. If this is set the
* {@link ItemRecoverer#recover(Object, Throwable)} will be called when
* retry is exhausted, and within the business transaction (which will not
* roll back because of any other item-related errors).
*
* @param itemRecoverer the {@link ItemRecoverer} to set
*/
public void setItemRecoverer(ItemRecoverer itemRecoverer) {
this.itemRecoverer = itemRecoverer;
}
/**
* @param step
*
@@ -87,36 +98,20 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
// exception handler limit, so this is a hack for now.
getStepOperations().setExceptionHandler(new SimpleLimitExceptionHandler(Integer.MAX_VALUE));
ItemReaderRetryCallback retryCallback = new ItemReaderRetryCallback(getItemReader(), getKeyGenerator(),
ItemReaderRetryCallback retryCallback = new ItemReaderRetryCallback(getItemReader(), itemKeyGenerator,
getItemWriter());
retryCallback.setRecoverer(itemRecoverer);
ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy);
RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy(itemProviderRetryPolicy);
StatefulRetryItemHandler itemProcessor = new StatefulRetryItemHandler(getItemReader(), getItemWriter(), retryTemplate, retryCallback);
StatefulRetryItemHandler itemProcessor = new StatefulRetryItemHandler(getItemReader(), getItemWriter(),
retryTemplate, retryCallback);
step.setItemProcessor(itemProcessor);
}
}
/**
* @return an {@link ItemKeyGenerator} or null if none is found.
*/
private ItemKeyGenerator getKeyGenerator() {
if (itemKeyGenerator != null) {
return itemKeyGenerator;
}
if (getItemReader() instanceof ItemKeyGenerator) {
return (ItemKeyGenerator) getItemReader();
}
if (getItemWriter() instanceof ItemKeyGenerator) {
return (ItemKeyGenerator) getItemWriter();
}
return null;
}
@@ -129,10 +124,11 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
/**
* @param itemReader
* @param itemWriter
* @param retryCallback
* @param retryTemplate
* @param retryCallback
* @param retryTemplate
*/
public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate, ItemReaderRetryCallback retryCallback) {
public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate,
ItemReaderRetryCallback retryCallback) {
super(itemReader, itemWriter);
this.retryOperations = retryTemplate;
this.retryCallback = retryCallback;

2
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java
View File

@@ -79,8 +79,6 @@ public class SimpleJobRepositoryTests extends TestCase {
ExecutionContext executionContext;
private JobExecution jobExecution;
public void setUp() throws Exception {
jobExecutionDao = (JobExecutionDao) jobExecutionDaoControl.getMock();

2
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java
View File

@@ -393,7 +393,7 @@ public class ItemOrientedStepTests extends TestCase {
public void testAfterStep() throws Exception{
final ExitStatus customStatus = new ExitStatus(true, "custom code");
final ExitStatus customStatus = new ExitStatus(false, "custom code");
itemOrientedStep.setStepListeners(new StepListener[] {new StepListenerSupport() {
public ExitStatus afterStep() {

21
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBeanTests.java
View File

@@ -26,7 +26,6 @@ import junit.framework.TestCase;
import org.springframework.batch.core.domain.BatchListener;
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.listener.ItemListenerSupport;
import org.springframework.batch.execution.job.SimpleJob;
@@ -36,7 +35,6 @@ import org.springframework.batch.execution.repository.dao.MapJobInstanceDao;
import org.springframework.batch.execution.repository.dao.MapStepExecutionDao;
import org.springframework.batch.execution.step.AbstractStep;
import org.springframework.batch.execution.step.ItemOrientedStep;
import org.springframework.batch.execution.step.support.DefaultStepFactoryBean;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.reader.ListItemReader;
@@ -84,7 +82,6 @@ public class DefaultStepFactoryBeanTests extends TestCase {
private DefaultStepFactoryBean getStep(String[] args) throws Exception {
DefaultStepFactoryBean factory = new DefaultStepFactoryBean();
factory.setSingleton(false);
List items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(args));
@@ -95,13 +92,6 @@ public class DefaultStepFactoryBeanTests extends TestCase {
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
factory.setBeanName("stepName");
// step.setItemRecoverer(new ItemRecoverer() {
// public boolean recover(Object item, Throwable cause) {
// recovered.add(item);
// assertTrue(TransactionSynchronizationManager.isActualTransactionActive());
// return true;
// }
// });
return factory;
}
@@ -115,12 +105,10 @@ public class DefaultStepFactoryBeanTests extends TestCase {
step.setName("step2");
job.addStep(step);
JobInstance jobInstance = repository.createJobExecution(job, new JobParameters()).getJobInstance();
JobExecution jobExecution = repository.createJobExecution(job, new JobParameters());
JobExecution jobExecutionContext = new JobExecution(jobInstance);
job.execute(jobExecutionContext);
assertEquals(BatchStatus.COMPLETED, jobExecutionContext.getStatus());
job.execute(jobExecution);
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
assertEquals(3, processed.size());
assertTrue(processed.contains("foo"));
}
@@ -154,6 +142,7 @@ public class DefaultStepFactoryBeanTests extends TestCase {
public void onReadError(Exception ex) {
recovered.add(ex);
}
public void onWriteError(Exception ex, Object item) {
recovered.add(ex);
}
@@ -173,8 +162,6 @@ public class DefaultStepFactoryBeanTests extends TestCase {
assertEquals(null, provider.read());
assertEquals(3, recovered.size());
}
// TODO: test recovery and stateful retry
public void testExceptionTerminates() throws Exception {
DefaultStepFactoryBean factory = getStep(new String[] { "foo", "bar", "spam" });

11
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/RepeatOperationsStepFactoryBeanTests.java
View File

@@ -44,7 +44,14 @@ public class RepeatOperationsStepFactoryBeanTests extends TestCase {
private List list;
private JobExecution jobExecution = new JobExecution(new JobInstance(new Long(0L), new JobParameters(),
new JobSupport("job")));;
new JobSupport("job")));
protected void setUp() throws Exception {
factory.setItemReader(new ListItemReader(new ArrayList()));
factory.setItemWriter(new EmptyItemWriter());
factory.setJobRepository(new JobRepositorySupport());
factory.setTransactionManager(new ResourcelessTransactionManager());
}
public void testType() throws Exception {
assertEquals(Step.class, factory.getObjectType());
@@ -70,8 +77,6 @@ public class RepeatOperationsStepFactoryBeanTests extends TestCase {
}
});
factory.setSingleton(false);
Step step = (Step) factory.getObject();
step.execute(new StepExecution(step, jobExecution));

99
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBeanTests.java
View File

@@ -15,24 +15,115 @@
*/
package org.springframework.batch.execution.step.support;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import junit.framework.TestCase;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobParameters;
import org.springframework.batch.core.domain.JobParametersBuilder;
import org.springframework.batch.core.domain.Step;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.execution.job.JobSupport;
import org.springframework.batch.execution.repository.SimpleJobRepository;
import org.springframework.batch.execution.repository.dao.MapJobExecutionDao;
import org.springframework.batch.execution.repository.dao.MapJobInstanceDao;
import org.springframework.batch.execution.repository.dao.MapStepExecutionDao;
import org.springframework.batch.execution.step.ItemOrientedStep;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.reader.ListItemReader;
import org.springframework.batch.item.writer.AbstractItemWriter;
import org.springframework.batch.retry.policy.AlwaysRetryPolicy;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
import org.springframework.transaction.support.TransactionSynchronizationManager;
/**
* @author Dave Syer
*
*
*/
public class StatefulRetryStepFactoryBeanTests extends TestCase {
private AbstractStepFactoryBean factory = new StatefulRetryStepFactoryBean();
private StatefulRetryStepFactoryBean factory = new StatefulRetryStepFactoryBean();
private List recovered = new ArrayList();
private List processed = new ArrayList();
private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(),
new MapStepExecutionDao());
JobExecution jobExecution;
private ItemWriter processor = new AbstractItemWriter() {
public void write(Object data) throws Exception {
processed.add((String) data);
}
};
/*
* (non-Javadoc)
* @see junit.framework.TestCase#setUp()
*/
protected void setUp() throws Exception {
MapJobInstanceDao.clear();
MapJobExecutionDao.clear();
MapStepExecutionDao.clear();
factory.setBeanName("step");
factory.setItemReader(new ListItemReader(new ArrayList()));
factory.setItemWriter(processor);
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
JobSupport job = new JobSupport("jobName");
job.setRestartable(true);
JobParameters jobParameters = new JobParametersBuilder().addString("statefulTest", "make_this_unique").toJobParameters();
jobExecution = repository.createJobExecution(job, jobParameters);
jobExecution.setEndTime(new Date());
}
public void testType() throws Exception {
assertEquals(Step.class, factory.getObjectType());
}
public void testDefaultValue() throws Exception {
assertTrue(factory.getObject() instanceof Step);
}
public void testRecovery() throws Exception {
factory.setItemRecoverer(new ItemRecoverer() {
public boolean recover(Object item, Throwable cause) {
recovered.add(item);
assertTrue(TransactionSynchronizationManager.isActualTransactionActive());
return true;
}
});
List items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(new String[] { "a", "b", "c" }));
ItemReader provider = new ListItemReader(items) {
int count = 0;
public Object read() {
count++;
if (count == 2) {
throw new RuntimeException("Temporary error - retry for success.");
}
return super.read();
}
};
factory.setItemReader(provider);
factory.setRetryPolicy(new AlwaysRetryPolicy());
ItemOrientedStep step = (ItemOrientedStep) factory.getObject();
step.execute(new StepExecution(step, jobExecution));
}
}

13
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java
View File

@@ -26,12 +26,9 @@ import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.execution.job.JobSupport;
import org.springframework.batch.execution.repository.SimpleJobRepository;
import org.springframework.batch.execution.repository.dao.JobExecutionDao;
import org.springframework.batch.execution.repository.dao.JobInstanceDao;
import org.springframework.batch.execution.repository.dao.MapJobExecutionDao;
import org.springframework.batch.execution.repository.dao.MapJobInstanceDao;
import org.springframework.batch.execution.repository.dao.MapStepExecutionDao;
import org.springframework.batch.execution.repository.dao.StepExecutionDao;
import org.springframework.batch.execution.step.ItemOrientedStep;
import org.springframework.batch.item.reader.AbstractItemReader;
import org.springframework.batch.item.reader.ItemReaderAdapter;
@@ -42,14 +39,6 @@ import org.springframework.batch.support.transaction.ResourcelessTransactionMana
public class StepExecutorInterruptionTests extends TestCase {
private JobRepository jobRepository;
private JobInstanceDao jobInstanceDao = new MapJobInstanceDao();
private JobExecutionDao jobExecutionDao = new MapJobExecutionDao();
private StepExecutionDao stepExecutionDao = new MapStepExecutionDao();
private ItemOrientedStep step;
private JobExecution jobExecution;
@@ -61,7 +50,7 @@ public class StepExecutorInterruptionTests extends TestCase {
MapJobExecutionDao.clear();
MapStepExecutionDao.clear();
jobRepository = new SimpleJobRepository(jobInstanceDao, jobExecutionDao, stepExecutionDao);
JobRepository jobRepository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), new MapStepExecutionDao());
JobSupport jobConfiguration = new JobSupport();
step = new ItemOrientedStep("interruptedStep");