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

Refactored the JsrJobOperator to start and restart jobs asynchronously

Browse Source 
by default.
* Refactored the JsrJobOperator to use SimpleAsyncTaskExecutor by
 default
* Introduced a JobContextFactory to provide a JobContext per thread.
* Updated stop logic to correctly handle stopping jobs running on
 another thread.
This commit is contained in:
Michael Minella
2013-12-02 09:54:13 -06:00
parent 8bee7ac310
commit 78cf001e9e
13 changed files with 612 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
dictionary.txt
View File

@@ -99,3 +99,5 @@ accessors
subclassing
ajax
javascript
chris
schaefer

2
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/JobContext.java
View File

@@ -16,8 +16,8 @@
package org.springframework.batch.core.jsr;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.batch.runtime.BatchStatus;
import org.springframework.batch.core.ExitStatus;

133
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/JobContextFactoryBean.java Normal file
View File

@@ -0,0 +1,133 @@
/*
* Copyright 2013 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.jsr;
import java.util.Properties;
import javax.batch.runtime.context.JobContext;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.FactoryBeanNotInitializedException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.Assert;
/**
* Provides a single {@link JobContext} for each thread in a running job.
* Subsequent calls to {@link FactoryBean#getObject()} on the same thread will
* return the same instance. The {@link JobContext} wraps a {@link JobExecution}
* which is obtained in one of two ways:
* <ul>
* <li>The current step scope (getting it from the current {@link StepExecution}</li>
* <li>The provided {@link JobExecution} via the {@link #setJobExecution(JobExecution)}
* </ul>
*
* @author Michael Minella
* @since 3.0
*/
public class JobContextFactoryBean implements FactoryBean<JobContext> {
private JobExecution jobExecution;
@Autowired
private BatchPropertyContext propertyContext;
private static final ThreadLocal<JobContext> contextHolder = new ThreadLocal<JobContext>();
/* (non-Javadoc)
* @see org.springframework.beans.factory.FactoryBean#getObject()
*/
@Override
public JobContext getObject() throws Exception {
return getCurrent();
}
/* (non-Javadoc)
* @see org.springframework.beans.factory.FactoryBean#getObjectType()
*/
@Override
public Class<?> getObjectType() {
return JobContext.class;
}
/* (non-Javadoc)
* @see org.springframework.beans.factory.FactoryBean#isSingleton()
*/
@Override
public boolean isSingleton() {
return false;
}
/**
* Used to provide {@link JobContext} instances to batch artifacts that
* are not within the scope of a given step.
*
* @param jobExecution set the current {@link JobExecution}
*/
public void setJobExecution(JobExecution jobExecution) {
Assert.notNull(jobExecution, "A JobExecution is required");
this.jobExecution = jobExecution;
}
/**
* @param propertyContext the {@link BatchPropertyContext} to obtain job properties from
*/
public void setBatchPropertyContext(BatchPropertyContext propertyContext) {
this.propertyContext = propertyContext;
}
/**
* Used to remove the {@link JobContext} for the current thread. Not used via
* normal processing but useful for testing.
*/
public void close() {
if(contextHolder.get() != null) {
contextHolder.remove();
}
}
private JobContext getCurrent() {
if(contextHolder.get() == null) {
JobExecution curJobExecution = null;
if(StepSynchronizationManager.getContext() != null) {
curJobExecution = StepSynchronizationManager.getContext().getStepExecution().getJobExecution();
}
if(curJobExecution != null) {
jobExecution = curJobExecution;
}
if(jobExecution == null) {
throw new FactoryBeanNotInitializedException("A JobExecution is required");
}
org.springframework.batch.core.jsr.JobContext jobContext = new org.springframework.batch.core.jsr.JobContext();
jobContext.setJobExecution(jobExecution);
if(propertyContext != null) {
jobContext.setProperties(propertyContext.getJobProperties());
} else {
jobContext.setProperties(new Properties());
}
contextHolder.set(jobContext);
}
return contextHolder.get();
}
}

2
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobParser.java
View File

@@ -50,6 +50,8 @@ public class JobParser extends AbstractSingleBeanDefinitionParser {
String jobName = element.getAttribute(ID_ATTRIBUTE);
builder.setLazyInit(true);
builder.addConstructorArgValue(jobName);
String restartableAttribute = element.getAttribute(RESTARTABLE_ATTRIBUTE);

1
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java
View File

@@ -107,6 +107,7 @@ public class ListenerParser {
if (beanDefinitionRegistry.containsBeanDefinition(beanName)) {
BeanDefinition beanDefinition = beanDefinitionRegistry.getBeanDefinition(beanName);
beanDefinition.setScope(BeanDefinition.SCOPE_SINGLETON);
beanDefinition.setLazyInit(true);
}
}
}

209
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java
View File

@@ -17,11 +17,16 @@ package org.springframework.batch.core.jsr.launch;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import javax.batch.operations.BatchRuntimeException;
import javax.batch.operations.JobExecutionAlreadyCompleteException;
import javax.batch.operations.JobExecutionIsRunningException;
import javax.batch.operations.JobExecutionNotMostRecentException;
@@ -38,23 +43,31 @@ import javax.batch.runtime.JobExecution;
import javax.batch.runtime.JobInstance;
import javax.batch.runtime.StepExecution;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.DuplicateJobException;
import org.springframework.batch.core.converter.JobParametersConverter;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.jsr.JobContext;
import org.springframework.batch.core.jsr.JobContextFactoryBean;
import org.springframework.batch.core.jsr.JsrJobParametersConverter;
import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext;
import org.springframework.batch.core.jsr.configuration.support.JobParameterResolvingBeanFactoryPostProcessor;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.batch.core.step.NoSuchStepException;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.batch.core.step.tasklet.StoppableTasklet;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.access.BeanFactoryLocator;
import org.springframework.beans.factory.access.BeanFactoryReference;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.context.ApplicationContext;
@@ -63,9 +76,8 @@ import org.springframework.context.support.GenericXmlApplicationContext;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.core.task.TaskRejectedException;
import org.springframework.util.Assert;
/**
@@ -109,8 +121,9 @@ import org.springframework.util.Assert;
* Calls to {@link JobOperator#start(String, Properties)} will provide a child context to the above context
* using the job definition and batch.xml if provided.
*
* By default, calls to start/restart will result in synchronous execution of the batch job (via a synchronous {@link TaskExecutor}.
* For asynchronous behavior, a different {@link TaskExecutor} implementation is required to be provided.
* By default, calls to start/restart will result in asynchronous execution of the batch job (via an asynchronous {@link TaskExecutor}.
* For synchronous behavior or customization of thread behavior, a different {@link TaskExecutor} implementation is required to
* be provided.
*
* <em>Note</em>: This class is intended to only be used for JSR-352 configured jobs. Use of
* this {@link JobOperator} to start/stop/restart Spring Batch jobs may result in unexpected behaviors due to
@@ -121,15 +134,15 @@ import org.springframework.util.Assert;
* @since 3.0
*/
public class JsrJobOperator implements JobOperator, InitializingBean {
private static final String BATCH_PROPERTY_CONTEXT_BEAN_NAME = "batchPropertyContext";
private static final String JSR_JOB_CONTEXT_BEAN_NAME = "jsr_jobContext";
private final Log logger = LogFactory.getLog(getClass());
private org.springframework.batch.core.launch.JobOperator batchJobOperator;
private JobExplorer jobExplorer;
private JobRepository jobRepository;
private TaskExecutor taskExecutor;
private JobParametersConverter jobParametersConverter;
private static ApplicationContext baseContext;
private static ExecutingJobRegistry jobRegistry = new ExecutingJobRegistry();
/**
* Public constructor used by {@link BatchRuntime#getJobOperator()}. This will bootstrap a
@@ -145,7 +158,7 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
AutowireCapableBeanFactory.AUTOWIRE_BY_TYPE, false);
if(taskExecutor == null) {
taskExecutor = new SyncTaskExecutor();
taskExecutor = new SimpleAsyncTaskExecutor();
}
}
@@ -156,17 +169,15 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
*
* @param jobExplorer an instance of Spring Batch's {@link JobExplorer}
* @param jobRepository an instance of Spring Batch's {@link JobOperator}
* @param jobOperator an instance of Spring Batch's {@link org.springframework.batch.core.launch.JobOperator}
* @param jobParametersConverter an instance of Spring Batch's {@link JobParametersConverter}
*/
public JsrJobOperator(JobExplorer jobExplorer, JobRepository jobRepository, org.springframework.batch.core.launch.JobOperator jobOperator, JobParametersConverter jobParametersConverter) {
public JsrJobOperator(JobExplorer jobExplorer, JobRepository jobRepository, JobParametersConverter jobParametersConverter) {
Assert.notNull(jobExplorer, "A JobExplorer is required");
Assert.notNull(jobRepository, "A JobRepository is required");
Assert.notNull(jobOperator, "A JobOperator is required");
Assert.notNull(jobParametersConverter, "A ParametersConverter is required");
this.jobExplorer = jobExplorer;
this.jobRepository = jobRepository;
this.batchJobOperator = jobOperator;
this.jobParametersConverter = jobParametersConverter;
}
@@ -182,12 +193,6 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
this.jobRepository = jobRepository;
}
public void setJobOperator(org.springframework.batch.core.launch.JobOperator jobOperator) {
Assert.notNull(jobOperator, "A JobOperator is required");
this.batchJobOperator = jobOperator;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
@@ -199,7 +204,7 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
@Override
public void afterPropertiesSet() throws Exception {
if (this.taskExecutor == null) {
this.taskExecutor = new SyncTaskExecutor();
this.taskExecutor = new SimpleAsyncTaskExecutor();
}
}
@@ -444,7 +449,7 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
String jobName = previousJobExecution.getJobInstance().getJobName();
GenericXmlApplicationContext batchContext = new GenericXmlApplicationContext();
final GenericXmlApplicationContext batchContext = new GenericXmlApplicationContext();
batchContext.setValidating(false);
Resource batchXml = new ClassPathResource("/META-INF/batch.xml");
@@ -460,7 +465,8 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
batchContext.addBeanFactoryPostProcessor(new JobParameterResolvingBeanFactoryPostProcessor(params));
AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.jsr.JobContext").getBeanDefinition();
AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.jsr.JobContextFactoryBean").getBeanDefinition();
beanDefinition.setScope("singleton");
batchContext.registerBeanDefinition(JSR_JOB_CONTEXT_BEAN_NAME, beanDefinition);
batchContext.setParent(baseContext);
@@ -488,38 +494,34 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
}
try {
ConfigurableListableBeanFactory factory = batchContext.getBeanFactory();
BatchPropertyContext batchPropertyContext = factory.getBean(BATCH_PROPERTY_CONTEXT_BEAN_NAME, BatchPropertyContext.class);
Properties properties = batchPropertyContext.getJobProperties();
JobContext jobContext = factory.getBean(JSR_JOB_CONTEXT_BEAN_NAME, JobContext.class);
jobContext.setJobExecution(jobExecution);
jobContext.setProperties(properties);
JobContextFactoryBean factoryBean = (JobContextFactoryBean) batchContext.getBean("&" + JSR_JOB_CONTEXT_BEAN_NAME);
factoryBean.setJobExecution(jobExecution);
taskExecutor.execute(new Runnable() {
@Override
public void run() {
try {
jobRegistry.register(job, jobExecution);
job.execute(jobExecution);
jobRegistry.remove(jobExecution);
}
catch (Throwable t) {
throw new JobRestartException(t);
throw new JobStartException(t);
}
}
});
}
catch (TaskRejectedException e) {
catch (Exception e) {
jobExecution.upgradeStatus(BatchStatus.FAILED);
if (jobExecution.getExitStatus().equals(ExitStatus.UNKNOWN)) {
jobExecution.setExitStatus(ExitStatus.FAILED.addExitDescription(e));
}
jobRepository.update(jobExecution);
} finally {
batchContext.close();
}
batchContext.close();
return jobExecution.getId();
}
@@ -555,7 +557,7 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
@SuppressWarnings("resource")
public long start(String jobName, Properties params) throws JobStartException,
JobSecurityException {
GenericXmlApplicationContext batchContext = new GenericXmlApplicationContext();
final GenericXmlApplicationContext batchContext = new GenericXmlApplicationContext();
batchContext.setValidating(false);
Resource batchXml = new ClassPathResource("/META-INF/batch.xml");
@@ -571,8 +573,8 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
}
batchContext.addBeanFactoryPostProcessor(new JobParameterResolvingBeanFactoryPostProcessor(params));
AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.jsr.JobContext").getBeanDefinition();
AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.jsr.JobContextFactoryBean").getBeanDefinition();
beanDefinition.setScope("singleton");
batchContext.registerBeanDefinition(JSR_JOB_CONTEXT_BEAN_NAME, beanDefinition);
batchContext.setParent(baseContext);
@@ -583,58 +585,77 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
throw new JobStartException(e);
}
final Job job = batchContext.getBean(Job.class);
Assert.notNull(jobName, "The job name must not be null.");
final org.springframework.batch.core.JobExecution jobExecution;
try {
JobParameters jobParameters = jobParametersConverter.getJobParameters(params);
org.springframework.batch.core.JobInstance jobInstance = jobRepository.createJobInstance(job.getName(), jobParameters);
String [] jobNames = batchContext.getBeanNamesForType(Job.class);
if(jobNames == null || jobNames.length <= 0) {
throw new BatchRuntimeException("No Job defined in current context");
}
org.springframework.batch.core.JobInstance jobInstance = jobRepository.createJobInstance(jobNames[0], jobParameters);
jobExecution = jobRepository.createJobExecution(jobInstance, jobParameters, jobConfigurationLocation);
} catch (Exception e) {
throw new JobStartException(e);
}
try {
ConfigurableListableBeanFactory factory = batchContext.getBeanFactory();
BatchPropertyContext batchPropertyContext = factory.getBean(BATCH_PROPERTY_CONTEXT_BEAN_NAME, BatchPropertyContext.class);
Properties properties = batchPropertyContext.getJobProperties();
JobContext jobContext = factory.getBean(JSR_JOB_CONTEXT_BEAN_NAME, JobContext.class);
jobContext.setJobExecution(jobExecution);
jobContext.setProperties(properties);
final Semaphore semaphore = new Semaphore(1);
final List<Exception> exceptionHolder = Collections.synchronizedList(new ArrayList<Exception>());
semaphore.acquire();
taskExecutor.execute(new Runnable() {
@Override
public void run() {
try {
JobContextFactoryBean factoryBean = (JobContextFactoryBean) batchContext.getBean("&" + JSR_JOB_CONTEXT_BEAN_NAME);
factoryBean.setJobExecution(jobExecution);
final Job job = batchContext.getBean(Job.class);
semaphore.release();
// Initialization of the JobExecution for job level dependencies
jobRegistry.register(job, jobExecution);
job.execute(jobExecution);
jobRegistry.remove(jobExecution);
}
catch (Throwable t) {
throw new JobStartException(t);
catch (Exception e) {
exceptionHolder.add(e);
} finally {
if(semaphore.availablePermits() == 0) {
semaphore.release();
}
}
}
});
semaphore.acquire();
if(exceptionHolder.size() > 0) {
semaphore.release();
throw exceptionHolder.get(0);
}
}
catch (TaskRejectedException e) {
catch (Exception e) {
e.printStackTrace(System.err);
if(jobRegistry.exists(jobExecution.getId())) {
jobRegistry.remove(jobExecution);
}
jobExecution.upgradeStatus(BatchStatus.FAILED);
if (jobExecution.getExitStatus().equals(ExitStatus.UNKNOWN)) {
jobExecution.setExitStatus(ExitStatus.FAILED.addExitDescription(e));
}
jobRepository.update(jobExecution);
throw new JobStartException(e);
}
batchContext.close();
return jobExecution.getId();
}
/**
* Delegates to {@link org.springframework.batch.core.launch.JobOperator#stop(long)}
* Stops the running job execution if it is currently running.
*
* @param executionId the database id for the {@link JobExecution} to be stopped.
* @throws NoSuchJobExecutionException
@@ -643,12 +664,78 @@ public class JsrJobOperator implements JobOperator, InitializingBean {
@Override
public void stop(long executionId) throws NoSuchJobExecutionException,
JobExecutionNotRunningException, JobSecurityException {
org.springframework.batch.core.JobExecution jobExecution = jobExplorer.getJobExecution(executionId);
// Indicate the execution should be stopped by setting it's status to
// 'STOPPING'. It is assumed that
// the step implementation will check this status at chunk boundaries.
BatchStatus status = jobExecution.getStatus();
if (!(status == BatchStatus.STARTED || status == BatchStatus.STARTING)) {
throw new JobExecutionNotRunningException("JobExecution must be running so that it can be stopped: "+jobExecution);
}
jobExecution.setStatus(BatchStatus.STOPPING);
jobRepository.update(jobExecution);
try {
batchJobOperator.stop(executionId);
} catch (org.springframework.batch.core.launch.NoSuchJobExecutionException e) {
throw new NoSuchJobException(e);
} catch (org.springframework.batch.core.launch.JobExecutionNotRunningException e) {
throw new JobExecutionNotRunningException(e);
Job job = jobRegistry.getJob(jobExecution.getId());
if (job instanceof StepLocator) {//can only process as StepLocator is the only way to get the step object
//get the current stepExecution
for (org.springframework.batch.core.StepExecution stepExecution : jobExecution.getStepExecutions()) {
if (stepExecution.getStatus().isRunning()) {
try {
//have the step execution that's running -> need to 'stop' it
Step step = ((StepLocator)job).getStep(stepExecution.getStepName());
if (step instanceof TaskletStep) {
Tasklet tasklet = ((TaskletStep)step).getTasklet();
if (tasklet instanceof StoppableTasklet) {
StepSynchronizationManager.register(stepExecution);
((StoppableTasklet)tasklet).stop();
StepSynchronizationManager.release();
}
}
}
catch (NoSuchStepException e) {
logger.warn("Step not found",e);
}
}
}
}
}
catch (NoSuchJobException e) {
logger.warn("Cannot find Job object",e);
}
}
private static class ExecutingJobRegistry {
private Map<Long, Job> registry = new ConcurrentHashMap<Long, Job>();
public void register(Job job, org.springframework.batch.core.JobExecution jobExecution) throws DuplicateJobException {
if(registry.containsKey(jobExecution.getId())) {
throw new DuplicateJobException("This job execution has already been registered");
} else {
registry.put(jobExecution.getId(), job);
}
}
public void remove(org.springframework.batch.core.JobExecution jobExecution) {
if(!registry.containsKey(jobExecution.getId())) {
throw new NoSuchJobExecutionException("The job execution " + jobExecution.getId() + " was not found");
} else {
registry.remove(jobExecution.getId());
}
}
public boolean exists(long jobExecutionId) {
return registry.containsKey(jobExecutionId);
}
public Job getJob(long jobExecutionId) {
if(!registry.containsKey(jobExecutionId)) {
throw new NoSuchJobExecutionException("The job execution " + jobExecutionId + " was not found");
} else {
return registry.get(jobExecutionId);
}
}
}
}

3
spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java
View File

@@ -53,6 +53,7 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.batch.core.step.NoSuchStepException;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.batch.core.step.tasklet.StoppableTasklet;
@@ -413,7 +414,9 @@ public class SimpleJobOperator implements JobOperator, InitializingBean {
if (step instanceof TaskletStep) {
Tasklet tasklet = ((TaskletStep)step).getTasklet();
if (tasklet instanceof StoppableTasklet) {
StepSynchronizationManager.register(stepExecution);
((StoppableTasklet)tasklet).stop();
StepSynchronizationManager.release();
}
}
}

124
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/JobContextFactoryBeanTests.java Normal file
View File

@@ -0,0 +1,124 @@
/*
* Copyright 2013 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.jsr;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import javax.batch.runtime.context.JobContext;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.beans.factory.FactoryBeanNotInitializedException;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
public class JobContextFactoryBeanTests {
private JobContextFactoryBean factoryBean;
private BatchPropertyContext propertyContext;
@Before
public void setUp() throws Exception {
propertyContext = new BatchPropertyContext();
factoryBean = new JobContextFactoryBean();
}
@After
public void tearDown() throws Exception {
factoryBean.close();
}
@Test
public void testIntialCreationSingleThread() throws Exception {
factoryBean.setJobExecution(new JobExecution(5l));
factoryBean.setBatchPropertyContext(propertyContext);
assertTrue(factoryBean.getObjectType().isAssignableFrom(JobContext.class));
assertFalse(factoryBean.isSingleton());
JobContext jobContext1 = factoryBean.getObject();
JobContext jobContext2 = factoryBean.getObject();
assertEquals(5l, jobContext1.getExecutionId());
assertEquals(5l, jobContext2.getExecutionId());
assertTrue(jobContext1 == jobContext2);
}
@Test
public void testInitialCreationSingleThreadUsingStepScope() throws Exception {
factoryBean.setBatchPropertyContext(propertyContext);
StepSynchronizationManager.register(new StepExecution("step1", new JobExecution(5l)));
JobContext jobContext = factoryBean.getObject();
assertEquals(5l, jobContext.getExecutionId());
StepSynchronizationManager.close();
}
@Test(expected=FactoryBeanNotInitializedException.class)
public void testNoJobExecutionProvided() throws Exception {
factoryBean.getObject();
}
@Test
public void testOneJobContextPerThread() throws Exception {
List<Future<JobContext>> jobContexts = new ArrayList<Future<JobContext>>();
AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
for(int i = 0; i < 4; i++) {
final long count = i;
jobContexts.add(executor.submit(new Callable<JobContext>() {
@Override
public JobContext call() throws Exception {
try {
StepSynchronizationManager.register(new StepExecution("step" + count, new JobExecution(count)));
JobContext context = factoryBean.getObject();
Thread.sleep(1000l);
return context;
} catch (Throwable ignore) {
return null;
}finally {
StepSynchronizationManager.release();
}
}
}));
}
Set<JobContext> contexts = new HashSet<JobContext>();
for (Future<JobContext> future : jobContexts) {
contexts.add(future.get());
}
assertEquals(4, contexts.size());
}
}

105
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/JsrTestUtils.java Normal file
View File

@@ -0,0 +1,105 @@
/*
* Copyright 2013 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.jsr;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.TimeoutException;
import javax.batch.operations.JobOperator;
import javax.batch.runtime.BatchRuntime;
import javax.batch.runtime.BatchStatus;
import javax.batch.runtime.JobExecution;
public class JsrTestUtils {
private static JobOperator operator;
static {
operator = BatchRuntime.getJobOperator();
}
private JsrTestUtils() {}
/**
* Executes a job and waits for it's status to be any of {@link BatchStatus#STOPPED},
* {@link BatchStatus#COMPLETED}, or {@link BatchStatus#FAILED}. If the job does not
* reach one of those statuses within the given timeout, a {@link TimeoutException} is
* thrown.
*
* @param jobName
* @param properties
* @param timeout
* @return the {@link JobExecution} for the final state of the job
* @throws TimeoutException if the timeout occurs
*/
public static JobExecution runJob(String jobName, Properties properties, long timeout) throws TimeoutException{
long executionId = operator.start(jobName, properties);
JobExecution execution = operator.getJobExecution(executionId);
Date curDate = new Date();
BatchStatus curBatchStatus = execution.getBatchStatus();
while(true) {
if(curBatchStatus == BatchStatus.STOPPED || curBatchStatus == BatchStatus.COMPLETED || curBatchStatus == BatchStatus.FAILED) {
break;
}
if(new Date().getTime() - curDate.getTime() > timeout) {
throw new TimeoutException("Job processing did not complete in time");
}
execution = operator.getJobExecution(executionId);
curBatchStatus = execution.getBatchStatus();
}
return execution;
}
/**
* Restarts a job and waits for it's status to be any of {@link BatchStatus#STOPPED},
* {@link BatchStatus#COMPLETED}, or {@link BatchStatus#FAILED}. If the job does not
* reach one of those statuses within the given timeout, a {@link TimeoutException} is
* thrown.
*
* @param executionId
* @param properties
* @param timeout
* @return the {@link JobExecution} for the final state of the job
* @throws TimeoutException if the timeout occurs
*/
public static JobExecution restartJob(long executionId, Properties properties, long timeout) throws TimeoutException {
long restartId = operator.restart(executionId, properties);
JobExecution execution = operator.getJobExecution(restartId);
Date curDate = new Date();
BatchStatus curBatchStatus = execution.getBatchStatus();
while(true) {
if(curBatchStatus == BatchStatus.STOPPED || curBatchStatus == BatchStatus.COMPLETED || curBatchStatus == BatchStatus.FAILED) {
break;
}
if(new Date().getTime() - curDate.getTime() > timeout) {
throw new TimeoutException("Job processing did not complete in time");
}
execution = operator.getJobExecution(restartId);
curBatchStatus = execution.getBatchStatus();
}
return execution;
}
}

38
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java
View File

@@ -17,6 +17,7 @@ package org.springframework.batch.core.jsr.configuration.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.springframework.batch.core.jsr.JsrTestUtils.runJob;
import java.io.Serializable;
import java.util.ArrayList;
@@ -35,28 +36,19 @@ import javax.batch.api.chunk.AbstractItemReader;
import javax.batch.api.chunk.AbstractItemWriter;
import javax.batch.api.partition.PartitionPlan;
import javax.batch.api.partition.PartitionPlanImpl;
import javax.batch.operations.JobOperator;
import javax.batch.runtime.BatchRuntime;
import javax.batch.runtime.BatchStatus;
import javax.batch.runtime.JobExecution;
import javax.batch.runtime.context.JobContext;
import javax.inject.Inject;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.util.Assert;
public class PartitionParserTests {
private static JobOperator operator;
private Pattern caPattern = Pattern.compile("ca");
private Pattern asPattern = Pattern.compile("AS");
@BeforeClass
public static void beforeClass() {
operator = BatchRuntime.getJobOperator();
}
private static final long TIMEOUT = 10000l;
@Before
public void before() {
@@ -67,17 +59,17 @@ public class PartitionParserTests {
}
@Test
public void testBatchletNoProperties() {
JobExecution execution = operator.getJobExecution(operator.start("partitionParserTestsBatchlet", new Properties()));
public void testBatchletNoProperties() throws Exception {
BatchStatus curBatchStatus = runJob("partitionParserTestsBatchlet", new Properties(), TIMEOUT).getBatchStatus();
assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus());
assertEquals(BatchStatus.COMPLETED, curBatchStatus);
assertEquals(10, MyBatchlet.processed);
assertEquals(10, MyBatchlet.threadNames.size());
}
@Test
public void testChunkNoProperties() {
JobExecution execution = operator.getJobExecution(operator.start("partitionParserTestsChunk", new Properties()));
public void testChunkNoProperties() throws Exception {
JobExecution execution = runJob("partitionParserTestsChunk", new Properties(), TIMEOUT);
assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus());
assertEquals(30, ItemReader.processedItems.size());
@@ -87,8 +79,8 @@ public class PartitionParserTests {
}
@Test
public void testFullPartitionConfiguration() {
JobExecution execution = operator.getJobExecution(operator.start("fullPartitionParserTests", new Properties()));
public void testFullPartitionConfiguration() throws Exception {
JobExecution execution = runJob("fullPartitionParserTests", new Properties(), TIMEOUT);
assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus());
assertTrue(execution.getExitStatus().startsWith("BPS_"));
@@ -100,8 +92,8 @@ public class PartitionParserTests {
}
@Test
public void testFullPartitionConfigurationWithProperties() {
JobExecution execution = operator.getJobExecution(operator.start("fullPartitionParserWithPropertiesTests", new Properties()));
public void testFullPartitionConfigurationWithProperties() throws Exception {
JobExecution execution = runJob("fullPartitionParserWithPropertiesTests", new Properties(), TIMEOUT);
assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus());
assertTrue(execution.getExitStatus().startsWith("BPS_"));
@@ -119,8 +111,8 @@ public class PartitionParserTests {
}
@Test
public void testFullPartitionConfigurationWithMapperSuppliedProperties() {
JobExecution execution = operator.getJobExecution(operator.start("fullPartitionParserWithMapperPropertiesTests", new Properties()));
public void testFullPartitionConfigurationWithMapperSuppliedProperties() throws Exception {
JobExecution execution = runJob("fullPartitionParserWithMapperPropertiesTests", new Properties(), TIMEOUT);
assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus());
assertTrue(execution.getExitStatus().startsWith("BPS_"));
@@ -145,8 +137,8 @@ public class PartitionParserTests {
}
@Test
public void testFullPartitionConfigurationWithHardcodedProperties() {
JobExecution execution = operator.getJobExecution(operator.start("fullPartitionParserWithHardcodedPropertiesTests", new Properties()));
public void testFullPartitionConfigurationWithHardcodedProperties() throws Exception {
JobExecution execution = runJob("fullPartitionParserWithHardcodedPropertiesTests", new Properties(), TIMEOUT);
assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus());
assertTrue(execution.getExitStatus().startsWith("BPS_"));

18
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/ThreadLocalClassloaderBeanPostProcessorTests.java
View File

@@ -16,29 +16,19 @@
package org.springframework.batch.core.jsr.configuration.xml;
import static org.junit.Assert.assertEquals;
import static org.springframework.batch.core.jsr.JsrTestUtils.runJob;
import java.util.Properties;
import javax.batch.operations.JobOperator;
import javax.batch.runtime.BatchRuntime;
import javax.batch.runtime.BatchStatus;
import javax.batch.runtime.JobExecution;
import org.junit.Before;
import org.junit.Test;
public class ThreadLocalClassloaderBeanPostProcessorTests {
private JobOperator jobOperator;
@Before
public void setUp() throws Exception {
jobOperator = BatchRuntime.getJobOperator();
}
@Test
public void test() throws Exception {
long executionId = jobOperator.start("threadLocalClassloaderBeanPostProcessorTestsJob", new Properties());
JobExecution execution = runJob("threadLocalClassloaderBeanPostProcessorTestsJob", null, 10000);
assertEquals(BatchStatus.COMPLETED, jobOperator.getJobExecution(executionId).getBatchStatus());
assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus());
}
}

120
spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java
View File

@@ -20,9 +20,10 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.springframework.batch.core.jsr.JsrTestUtils.restartJob;
import static org.springframework.batch.core.jsr.JsrTestUtils.runJob;
import java.util.ArrayList;
import java.util.Date;
@@ -31,6 +32,7 @@ import java.util.List;
import java.util.Properties;
import java.util.Set;
import javax.batch.api.Batchlet;
import javax.batch.operations.JobExecutionIsRunningException;
import javax.batch.operations.JobOperator;
import javax.batch.operations.JobRestartException;
@@ -55,29 +57,27 @@ import org.springframework.batch.core.converter.JobParametersConverterSupport;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.explore.support.SimpleJobExplorer;
import org.springframework.batch.core.jsr.JsrJobParametersConverter;
import org.springframework.batch.core.launch.support.SimpleJobOperator;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.JobRepositorySupport;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SyncTaskExecutor;
public class JsrJobOperatorTests {
private JobOperator jsrJobOperator;
@Mock
private org.springframework.batch.core.launch.JobOperator jobOperator;
@Mock
private JobExplorer jobExplorer;
@Mock
private JobRepository jobRepository;
private JobParametersConverter parameterConverter;
private static final long TIMEOUT = 10000l;
@Before
public void setup() {
MockitoAnnotations.initMocks(this);
parameterConverter = new JobParametersConverterSupport();
jsrJobOperator = new JsrJobOperator(jobExplorer, jobRepository, jobOperator, parameterConverter);
jsrJobOperator = new JsrJobOperator(jobExplorer, jobRepository, parameterConverter);
}
@Test
@@ -89,30 +89,24 @@ public class JsrJobOperatorTests {
@Test
public void testNullsInConstructor() {
try {
new JsrJobOperator(null, new JobRepositorySupport(), new SimpleJobOperator(), parameterConverter);
new JsrJobOperator(null, new JobRepositorySupport(), parameterConverter);
fail("JobExplorer should be required");
} catch (IllegalArgumentException correct) {
}
try {
new JsrJobOperator(new SimpleJobExplorer(null, null, null, null), null, new SimpleJobOperator(), parameterConverter);
new JsrJobOperator(new SimpleJobExplorer(null, null, null, null), null, parameterConverter);
fail("JobRepository should be required");
} catch (IllegalArgumentException correct) {
}
try {
new JsrJobOperator(new SimpleJobExplorer(null, null, null, null), new JobRepositorySupport(), null, parameterConverter);
fail("JobOperator should be required");
} catch (IllegalArgumentException correct) {
}
try {
new JsrJobOperator(new SimpleJobExplorer(null, null, null, null), new JobRepositorySupport(), new SimpleJobOperator(), null);
new JsrJobOperator(new SimpleJobExplorer(null, null, null, null), new JobRepositorySupport(), null);
fail("ParameterConverter should be required");
} catch (IllegalArgumentException correct) {
}
new JsrJobOperator(new SimpleJobExplorer(null, null, null, null), new JobRepositorySupport(), new SimpleJobOperator(), parameterConverter);
new JsrJobOperator(new SimpleJobExplorer(null, null, null, null), new JobRepositorySupport(), parameterConverter);
}
@Test
@@ -120,16 +114,16 @@ public class JsrJobOperatorTests {
JsrJobOperator jsrJobOperatorImpl = (JsrJobOperator) jsrJobOperator;
jsrJobOperatorImpl.afterPropertiesSet();
assertNotNull(jsrJobOperatorImpl.getTaskExecutor());
assertTrue((jsrJobOperatorImpl.getTaskExecutor() instanceof SyncTaskExecutor));
assertTrue((jsrJobOperatorImpl.getTaskExecutor() instanceof AsyncTaskExecutor));
}
@Test
public void testCustomTaskExecutor() throws Exception {
JsrJobOperator jsrJobOperatorImpl = (JsrJobOperator) jsrJobOperator;
jsrJobOperatorImpl.setTaskExecutor(new SimpleAsyncTaskExecutor());
jsrJobOperatorImpl.setTaskExecutor(new SyncTaskExecutor());
jsrJobOperatorImpl.afterPropertiesSet();
assertNotNull(jsrJobOperatorImpl.getTaskExecutor());
assertTrue((jsrJobOperatorImpl.getTaskExecutor() instanceof SimpleAsyncTaskExecutor));
assertTrue((jsrJobOperatorImpl.getTaskExecutor() instanceof SyncTaskExecutor));
}
@Test
@@ -389,12 +383,10 @@ public class JsrJobOperatorTests {
}
@Test
public void testStartRoseyScenario() {
jsrJobOperator = BatchRuntime.getJobOperator();
public void testStartRoseyScenario() throws Exception {
javax.batch.runtime.JobExecution execution = runJob("jsrJobOperatorTestJob", new Properties(), TIMEOUT);
long executionId = jsrJobOperator.start("jsrJobOperatorTestJob", null);
assertEquals(BatchStatus.COMPLETED, jsrJobOperator.getJobExecution(executionId).getBatchStatus());
assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus());
}
@Test
@@ -408,13 +400,13 @@ public class JsrJobOperatorTests {
} catch (NoSuchJobException ignore) {
}
long run1 = jsrJobOperator.start("jsrJobOperatorTestJob", null);
long run2 = jsrJobOperator.start("jsrJobOperatorTestJob", null);
long run3 = jsrJobOperator.start("jsrJobOperatorTestJob", null);
javax.batch.runtime.JobExecution execution1 = runJob("jsrJobOperatorTestJob", new Properties(), TIMEOUT);
javax.batch.runtime.JobExecution execution2 = runJob("jsrJobOperatorTestJob", new Properties(), TIMEOUT);
javax.batch.runtime.JobExecution execution3 = runJob("jsrJobOperatorTestJob", new Properties(), TIMEOUT);
assertEquals(BatchStatus.COMPLETED, jsrJobOperator.getJobExecution(run1).getBatchStatus());
assertEquals(BatchStatus.COMPLETED, jsrJobOperator.getJobExecution(run2).getBatchStatus());
assertEquals(BatchStatus.COMPLETED, jsrJobOperator.getJobExecution(run3).getBatchStatus());
assertEquals(BatchStatus.COMPLETED, execution1.getBatchStatus());
assertEquals(BatchStatus.COMPLETED, execution2.getBatchStatus());
assertEquals(BatchStatus.COMPLETED, execution3.getBatchStatus());
int jobInstanceCountAfter = jsrJobOperator.getJobInstanceCount("myJob3");
@@ -422,28 +414,25 @@ public class JsrJobOperatorTests {
}
@Test
public void testRestartRoseyScenario() {
jsrJobOperator = BatchRuntime.getJobOperator();
public void testRestartRoseyScenario() throws Exception {
javax.batch.runtime.JobExecution execution = runJob("jsrJobOperatorTestRestartJob", new Properties(), TIMEOUT);
long executionId = jsrJobOperator.start("jsrJobOperatorTestRestartJob", null);
assertEquals(BatchStatus.FAILED, execution.getBatchStatus());
assertEquals(BatchStatus.FAILED, jsrJobOperator.getJobExecution(executionId).getBatchStatus());
execution = restartJob(execution.getExecutionId(), null, TIMEOUT);
long finalExecutionId = jsrJobOperator.restart(executionId, null);
assertEquals(BatchStatus.COMPLETED, jsrJobOperator.getJobExecution(finalExecutionId).getBatchStatus());
assertEquals(BatchStatus.COMPLETED, execution.getBatchStatus());
}
@Test(expected = JobRestartException.class)
public void testRestartAbandoned() {
public void testRestartAbandoned() throws Exception {
jsrJobOperator = BatchRuntime.getJobOperator();
javax.batch.runtime.JobExecution execution = runJob("jsrJobOperatorTestRestartAbandonJob", null, TIMEOUT);
long executionId = jsrJobOperator.start("jsrJobOperatorTestRestartAbandonJob", null);
assertEquals(BatchStatus.FAILED, execution.getBatchStatus());
assertEquals(BatchStatus.FAILED, jsrJobOperator.getJobExecution(executionId).getBatchStatus());
jsrJobOperator.abandon(executionId);
jsrJobOperator.restart(executionId, null);
jsrJobOperator.abandon(execution.getExecutionId());
jsrJobOperator.restart(execution.getExecutionId(), null);
}
@Test
@@ -506,18 +495,43 @@ public class JsrJobOperatorTests {
fail("Should have failed");
}
@Test(expected = JobStartException.class)
public void testBeanCreationExceptionOnRestart() throws Exception {
JsrJobOperator jsrJobOperator1 = mock(JsrJobOperator.class);
when(jsrJobOperator1.restart(0l, null)).thenThrow(new JobStartException(new BeanCreationException("Bean creation exception")));
@SuppressWarnings("unchecked")
@Test(expected=JobStartException.class)
public void testStartUnableToCreateJobExecution() throws Exception {
when(jobRepository.createJobExecution("myJob", null)).thenThrow(RuntimeException.class);
try {
jsrJobOperator1.restart(0l, null);
} catch (JobStartException e) {
assertTrue(e.getCause() instanceof BeanCreationException);
throw e;
jsrJobOperator.start("myJob", null);
}
@Test
public void testJobStopRoseyScenario() throws Exception {
jsrJobOperator = BatchRuntime.getJobOperator();
long executionId = jsrJobOperator.start("longRunningJob", null);
// Give the job a chance to get started
Thread.sleep(1000l);
jsrJobOperator.stop(executionId);
// Give the job the chance to finish stopping
Thread.sleep(1000l);
assertEquals(BatchStatus.STOPPED, jsrJobOperator.getJobExecution(executionId).getBatchStatus());
}
public static class LongRunningBatchlet implements Batchlet {
private boolean stopped = false;
@Override
public String process() throws Exception {
while(!stopped) {
Thread.sleep(250);
}
return null;
}
fail("Should have failed");
@Override
public void stop() throws Exception {
stopped = true;
}
}
}

7
spring-batch-core/src/test/resources/META-INF/batch-jobs/longRunningJob.xml Normal file
View File

@@ -0,0 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<!-- This job should parse fine but the objects won't build -->
<job id="job1" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
<step id="step1">
<batchlet ref="org.springframework.batch.core.jsr.launch.JsrJobOperatorTests$LongRunningBatchlet"/>
</step>
</job>