From dc21da497f6f19ac4df38cf2c1eaf4514531e9b9 Mon Sep 17 00:00:00 2001 From: dsyer Date: Tue, 8 Dec 2009 18:03:30 +0000 Subject: [PATCH] RESOLVED - issue BATCH-1458: Add FlowStep: a Step implementation that executes a flow Javadocs + re-org update() methods --- .../batch/core/job/AbstractJob.java | 11 ----- .../batch/core/job/SimpleStepHandler.java | 26 +++++------ .../batch/core/job/StepHandler.java | 29 ++++++++++-- .../batch/core/job/flow/FlowJob.java | 2 +- .../batch/core/job/flow/FlowStep.java | 44 ++++++++++++++++--- .../batch/core/job/flow/JobFlowExecutor.java | 8 +++- .../core/job/SimpleStepHandlerTests.java | 20 +-------- 7 files changed, 86 insertions(+), 54 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java index 2551f683e..27155fbcc 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java @@ -352,17 +352,6 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In } - /** - * Convenience method for subclasses so they can change the state of a - * {@link StepExecution} if necessary. Use with care (and not at all - * preferably) and only before or after a step is executed. - * - * @param stepExecution - */ - protected void updateStepExecution(StepExecution stepExecution) { - stepHandler.updateStepExecution(stepExecution); - } - /** * Default mapping from throwable to {@link ExitStatus}. * diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java index 34c7a039b..1132425a5 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java @@ -32,22 +32,25 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; /** + * Implementation of {@link StepHandler} that manages repository and restart + * concerns. + * * @author Dave Syer - * + * */ public class SimpleStepHandler implements StepHandler, InitializingBean { private static final Log logger = LogFactory.getLog(SimpleStepHandler.class); private JobRepository jobRepository; - + /** * Convenient default constructor for configuration usage. */ public SimpleStepHandler() { this(null); } - + /** * @param jobRepository */ @@ -55,14 +58,14 @@ public class SimpleStepHandler implements StepHandler, InitializingBean { super(); this.jobRepository = jobRepository; } - + /** * Check mandatory properties (jobRepository). * * @see InitializingBean#afterPropertiesSet() */ public void afterPropertiesSet() throws Exception { - Assert.state(jobRepository!=null, "A JobRepository must be provided"); + Assert.state(jobRepository != null, "A JobRepository must be provided"); } /** @@ -82,10 +85,11 @@ public class SimpleStepHandler implements StepHandler, InitializingBean { StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step.getName()); if (stepExecutionPartOfExistingJobExecution(execution, lastStepExecution)) { - // If the last execution of this step was in the same job, it's probably - // intentional so we want to run it again... - logger.info(String.format("Duplicate step [%s] detected in execution of job=[%s]. " + - "If either step fails, both will be executed again on restart.", step.getName(), jobInstance.getJobName())); + // If the last execution of this step was in the same job, it's + // probably intentional so we want to run it again... + logger.info(String.format("Duplicate step [%s] detected in execution of job=[%s]. " + + "If either step fails, both will be executed again on restart.", step.getName(), jobInstance + .getJobName())); lastStepExecution = null; } StepExecution currentStepExecution = lastStepExecution; @@ -135,10 +139,6 @@ public class SimpleStepHandler implements StepHandler, InitializingBean { return currentStepExecution; } - public void updateStepExecution(StepExecution stepExecution) { - jobRepository.update(stepExecution); - } - /** * Detect whether a step execution belongs to this job execution. * @param jobExecution the current job execution diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/StepHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/StepHandler.java index c52c3199f..10df10892 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/StepHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/StepHandler.java @@ -16,6 +16,7 @@ package org.springframework.batch.core.job; +import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.StartLimitExceededException; @@ -23,11 +24,33 @@ import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.repository.JobRestartException; +/** + * Strategy interface for handling a {@link Step} on behalf of a {@link Job}. + * + * @author Dave Syer + * + */ public interface StepHandler { - StepExecution handleStep(Step step, JobExecution execution) throws JobInterruptedException, JobRestartException, + /** + * Handle a step and return the execution for it. Does not save the + * {@link JobExecution}, but should manage the persistence of the + * {@link StepExecution} if required (e.g. at least it needs to be added to + * a repository before the step can eb executed). + * + * @param step a {@link Step} + * @param jobExecution a {@link JobExecution} + * @return an execution of the step + * + * @throws JobInterruptedException if there is an interruption + * @throws JobRestartException if there is a problem restarting a failed + * step + * @throws StartLimitExceededException if the step exceeds its start limit + * + * @see Job#execute(JobExecution) + * @see Step#execute(StepExecution) + */ + StepExecution handleStep(Step step, JobExecution jobExecution) throws JobInterruptedException, JobRestartException, StartLimitExceededException; - - void updateStepExecution(StepExecution stepExecution); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java index 9ffa83702..76080a15a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java @@ -94,7 +94,7 @@ public class FlowJob extends AbstractJob { @Override protected void doExecute(final JobExecution execution) throws JobExecutionException { try { - JobFlowExecutor executor = new JobFlowExecutor(new SimpleStepHandler(getJobRepository()), + JobFlowExecutor executor = new JobFlowExecutor(getJobRepository(), new SimpleStepHandler(getJobRepository()), execution); executor.updateJobExecutionStatus(flow.start(executor).getStatus()); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java index 873b74253..9a9a64293 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java @@ -4,13 +4,42 @@ import org.springframework.batch.core.JobExecutionException; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.job.SimpleStepHandler; import org.springframework.batch.core.job.StepHandler; +import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.step.AbstractStep; import org.springframework.util.Assert; +import com.sun.org.apache.xerces.internal.impl.xpath.XPath.Step; + +/** + * A {@link Step} implementation that delegates to a {@link Flow}. Useful for + * logical grouping of steps, and especially for partitioning with multiple + * steps per execution. If the flow has steps then when the {@link FlowStep} + * executes, all steps including the parent {@link FlowStep} will have + * executions in the {@link JobRepository} (one for the parent and one each for + * the flow steps). + * + * @author Dave Syer + * + */ public class FlowStep extends AbstractStep { - + private Flow flow; - + + /** + * Default constructor convenient for configuration purposes. + */ + public FlowStep() { + super(null); + } + + /** + * Constructor for a {@link FlowStep} that sets the flow and of the step + * explicitly. + */ + public FlowStep(Flow flow) { + super(flow.getName()); + } + /** * Public setter for the flow. * @@ -19,7 +48,7 @@ public class FlowStep extends AbstractStep { public void setFlow(Flow flow) { this.flow = flow; } - + /** * Ensure that the flow is set. * @see AbstractStep#afterPropertiesSet() @@ -27,14 +56,19 @@ public class FlowStep extends AbstractStep { @Override public void afterPropertiesSet() throws Exception { super.afterPropertiesSet(); - Assert.state(flow!=null, "A Flow must be provided"); + Assert.state(flow != null, "A Flow must be provided"); } + /** + * Delegate to the flow provided for the execution of the step. + * + * @see AbstractStep#doExecute(StepExecution) + */ @Override protected void doExecute(StepExecution stepExecution) throws Exception { try { StepHandler stepHandler = new SimpleStepHandler(getJobRepository()); - FlowExecutor executor = new JobFlowExecutor(stepHandler, stepExecution.getJobExecution()); + FlowExecutor executor = new JobFlowExecutor(getJobRepository(), stepHandler, stepExecution.getJobExecution()); executor.updateJobExecutionStatus(flow.start(executor).getStatus()); } catch (FlowExecutionException e) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java index 6dfe94605..0f2a1f1b8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java @@ -24,6 +24,7 @@ import org.springframework.batch.core.StartLimitExceededException; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.job.StepHandler; +import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.JobRestartException; /** @@ -43,10 +44,13 @@ public class JobFlowExecutor implements FlowExecutor { private final StepHandler stepHandler; + private final JobRepository jobRepository; + /** * @param execution */ - public JobFlowExecutor(StepHandler stepHandler, JobExecution execution) { + public JobFlowExecutor(JobRepository jobRepository, StepHandler stepHandler, JobExecution execution) { + this.jobRepository = jobRepository; this.stepHandler = stepHandler; this.execution = execution; stepExecutionHolder.set(null); @@ -63,7 +67,7 @@ public class JobFlowExecutor implements FlowExecutor { StepExecution lastStepExecution = stepExecutionHolder.get(); if (lastStepExecution != null && lastStepExecution.getStatus().isGreaterThan(BatchStatus.STOPPING)) { lastStepExecution.upgradeStatus(BatchStatus.ABANDONED); - stepHandler.updateStepExecution(lastStepExecution); + jobRepository.update(lastStepExecution); } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleStepHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleStepHandlerTests.java index dbbfde9f7..f051544fe 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleStepHandlerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleStepHandlerTests.java @@ -16,7 +16,7 @@ package org.springframework.batch.core.job; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; import org.junit.Before; import org.junit.Test; @@ -27,7 +27,6 @@ import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.core.repository.dao.StepExecutionDao; import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; import org.springframework.batch.core.step.StepSupport; @@ -43,13 +42,10 @@ public class SimpleStepHandlerTests { private SimpleStepHandler stepHandler; - private StepExecutionDao stepExecutionDao; - @Before public void setUp() throws Exception { MapJobRepositoryFactoryBean jobRepositoryFactoryBean = new MapJobRepositoryFactoryBean(); jobRepository = jobRepositoryFactoryBean.getJobRepository(); - stepExecutionDao = jobRepositoryFactoryBean.getStepExecutionDao(); jobExecution = jobRepository.createJobExecution("job", new JobParameters()); stepHandler = new SimpleStepHandler(jobRepository); stepHandler.afterPropertiesSet(); @@ -75,20 +71,6 @@ public class SimpleStepHandlerTests { assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); } - /** - * Test method for - * {@link SimpleStepHandler#updateStepExecution(org.springframework.batch.core.StepExecution)} - * - */ - @Test - public void testUpdateStepExecution() { - StepExecution stepExecution = jobExecution.createStepExecution("step"); - jobRepository.add(stepExecution); - stepExecution.setStatus(BatchStatus.FAILED); - stepHandler.updateStepExecution(stepExecution); - assertEquals(stepExecution, stepExecutionDao.getStepExecution(jobExecution, stepExecution.getId())); - } - private class StubStep extends StepSupport { private StubStep(String name) {