diff --git a/dictionary.txt b/dictionary.txt index 63e4bda79..2b131be9f 100644 --- a/dictionary.txt +++ b/dictionary.txt @@ -99,3 +99,5 @@ accessors subclassing ajax javascript +chris +schaefer diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/JobContext.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/JobContext.java index 8574e06ac..fec0f38d1 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/JobContext.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/JobContext.java @@ -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; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/JobContextFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/JobContextFactoryBean.java new file mode 100644 index 000000000..7994c25e4 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/JobContextFactoryBean.java @@ -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: + * + * + * @author Michael Minella + * @since 3.0 + */ +public class JobContextFactoryBean implements FactoryBean { + + private JobExecution jobExecution; + @Autowired + private BatchPropertyContext propertyContext; + + private static final ThreadLocal contextHolder = new ThreadLocal(); + + /* (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(); + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobParser.java index 634d8c141..f11d378ac 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/JobParser.java @@ -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); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java index ec3af3c12..df387469a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/configuration/xml/ListenerParser.java @@ -107,6 +107,7 @@ public class ListenerParser { if (beanDefinitionRegistry.containsBeanDefinition(beanName)) { BeanDefinition beanDefinition = beanDefinitionRegistry.getBeanDefinition(beanName); beanDefinition.setScope(BeanDefinition.SCOPE_SINGLETON); + beanDefinition.setLazyInit(true); } } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java index 13efbff67..12c349989 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java @@ -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. * * Note: 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 exceptionHolder = Collections.synchronizedList(new ArrayList()); + 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 registry = new ConcurrentHashMap(); + + 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); + } } } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java index f0ea157c9..1ec5a5641 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/launch/support/SimpleJobOperator.java @@ -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(); } } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/JobContextFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/JobContextFactoryBeanTests.java new file mode 100644 index 000000000..5b1399b1a --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/JobContextFactoryBeanTests.java @@ -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> jobContexts = new ArrayList>(); + + AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(); + + for(int i = 0; i < 4; i++) { + final long count = i; + jobContexts.add(executor.submit(new Callable() { + + @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 contexts = new HashSet(); + for (Future future : jobContexts) { + contexts.add(future.get()); + } + + assertEquals(4, contexts.size()); + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/JsrTestUtils.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/JsrTestUtils.java new file mode 100644 index 000000000..af7961158 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/JsrTestUtils.java @@ -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; + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java index a8fc87ea3..455165300 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/PartitionParserTests.java @@ -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_")); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/ThreadLocalClassloaderBeanPostProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/ThreadLocalClassloaderBeanPostProcessorTests.java index 86cbee342..2e8aa292f 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/ThreadLocalClassloaderBeanPostProcessorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/configuration/xml/ThreadLocalClassloaderBeanPostProcessorTests.java @@ -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()); } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java index b98a9ebc0..fcb7838c1 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java @@ -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; + } } } diff --git a/spring-batch-core/src/test/resources/META-INF/batch-jobs/longRunningJob.xml b/spring-batch-core/src/test/resources/META-INF/batch-jobs/longRunningJob.xml new file mode 100644 index 000000000..9a85eb8d6 --- /dev/null +++ b/spring-batch-core/src/test/resources/META-INF/batch-jobs/longRunningJob.xml @@ -0,0 +1,7 @@ + + + + + + +