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

Fixes for stopping a partitioned step

Browse Source 
* StepState now implements StepLocator so that the step it wraps can return slave steps in the case of a partitioned step
* FlowJob and SimpleJob now look for a StepLocator as they look for steps
* The JSR-352 version of PartitionStep implements StepLocator
This commit is contained in:
Michael Minella
2014-01-06 16:19:31 -06:00
parent 1bb02e4f55
commit b0c9af5e8e
8 changed files with 174 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2013 the original author or authors.
* Copyright 2012-2014 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.
@@ -151,7 +151,7 @@ public class SimpleFlowFactoryBean implements FactoryBean, InitializingBean {
}
String stateName = prefix + oldName;
if (state instanceof StepState) {
return new StepState(stateName, ((StepState) state).getStep());
return new StepState(stateName, ((StepState) state).getStep(oldName));
}
return new DelegateState(stateName, state);
}

13
spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2013 the original author or authors.
* Copyright 2006-2014 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.
@@ -28,6 +28,7 @@ import org.springframework.batch.core.StartLimitExceededException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.core.step.StepLocator;
/**
* Simple implementation of {@link Job} interface providing the ability to run a
@@ -37,6 +38,7 @@ import org.springframework.batch.core.repository.JobRestartException;
*
* @author Lucas Ward
* @author Dave Syer
* @author Michael Minella
*/
public class SimpleJob extends AbstractJob {
@@ -77,6 +79,10 @@ public class SimpleJob extends AbstractJob {
List<String> names = new ArrayList<String>();
for (Step step : steps) {
names.add(step.getName());
if(step instanceof StepLocator) {
names.addAll(((StepLocator)step).getStepNames());
}
}
return names;
}
@@ -101,6 +107,11 @@ public class SimpleJob extends AbstractJob {
for (Step step : this.steps) {
if (step.getName().equals(stepName)) {
return step;
} else if(step instanceof StepLocator) {
Step result = ((StepLocator)step).getStep(stepName);
if(result != null) {
return result;
}
}
}
return null;

15
spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2007 the original author or authors.
* Copyright 2006-2014 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.
@@ -93,7 +93,12 @@ public class FlowJob extends AbstractJob {
private void findSteps(Flow flow, Map<String, Step> map) {
for (State state : flow.getStates()) {
if (state instanceof StepHolder) {
if (state instanceof StepLocator) {
StepLocator locator = (StepLocator) state;
for (String name : locator.getStepNames()) {
map.put(name, locator.getStep(name));
}
} else if (state instanceof StepHolder) {
Step step = ((StepHolder) state).getStep();
String name = step.getName();
stepMap.put(name, step);
@@ -103,12 +108,6 @@ public class FlowJob extends AbstractJob {
findSteps(subflow, map);
}
}
else if (state instanceof StepLocator) {
StepLocator locator = (StepLocator) state;
for (String name : locator.getStepNames()) {
map.put(name, locator.getStep(name));
}
}
}
}

44
spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/StepState.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2013 the original author or authors.
* Copyright 2006-2014 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.
@@ -16,20 +16,27 @@
package org.springframework.batch.core.job.flow.support.state;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.FlowExecutor;
import org.springframework.batch.core.job.flow.State;
import org.springframework.batch.core.step.NoSuchStepException;
import org.springframework.batch.core.step.StepHolder;
import org.springframework.batch.core.step.StepLocator;
/**
* {@link State} implementation that delegates to a {@link FlowExecutor} to
* execute the specified {@link Step}.
*
* @author Dave Syer
* @author Michael Minella
* @since 2.0
*/
public class StepState extends AbstractState implements StepHolder {
public class StepState extends AbstractState implements StepLocator, StepHolder {
private final Step step;
@@ -61,7 +68,7 @@ public class StepState extends AbstractState implements StepHolder {
}
/**
* @return the step
* @deprecated in favor of using {@link StepLocator#getStep(String)}.
*/
@Override
public Step getStep() {
@@ -76,4 +83,35 @@ public class StepState extends AbstractState implements StepHolder {
return false;
}
/* (non-Javadoc)
* @see org.springframework.batch.core.step.StepLocator#getStepNames()
*/
@Override
public Collection<String> getStepNames() {
List<String> names = new ArrayList<String>();
names.add(step.getName());
if(step instanceof StepLocator) {
names.addAll(((StepLocator)step).getStepNames());
}
return names;
}
/* (non-Javadoc)
* @see org.springframework.batch.core.step.StepLocator#getStep(java.lang.String)
*/
@Override
public Step getStep(String stepName) throws NoSuchStepException {
Step result = null;
if(step.getName().equals(stepName)) {
result = step;
} else if(step instanceof StepLocator) {
result = ((StepLocator) step).getStep(stepName);
}
return result;
}
}

21
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/support/state/EndState.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013 the original author or authors.
* Copyright 2013-2014 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.
@@ -64,22 +64,9 @@ public class EndState extends org.springframework.batch.core.job.flow.support.st
protected void setExitStatus(FlowExecutor executor, String code) {
StepExecution stepExecution = executor.getStepExecution();
if(!isNonDefaultExitStatus(code)) {
stepExecution.getJobExecution().setExitStatus(new ExitStatus(code));
ExitStatus status = new ExitStatus(code);
if(!ExitStatus.isNonDefaultExitStatus(status)) {
stepExecution.getJobExecution().setExitStatus(status);
}
}
/**
* @param curStatus the exit code to be evaluated
* @return true if the value matches a known exit code
*/
protected boolean isNonDefaultExitStatus(String curStatus) {
return curStatus == null ||
curStatus.equals(ExitStatus.COMPLETED.getExitCode()) ||
curStatus.equals(ExitStatus.EXECUTING.getExitCode()) ||
curStatus.equals(ExitStatus.FAILED.getExitCode()) ||
curStatus.equals(ExitStatus.NOOP.getExitCode()) ||
curStatus.equals(ExitStatus.STOPPED.getExitCode()) ||
curStatus.equals(ExitStatus.UNKNOWN.getExitCode());
}
}

44
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java
View File

@@ -74,7 +74,26 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean {
private BatchPropertyContext propertyContext;
private JobRepository jobRepository;
private boolean allowStartIfComplete = false;
private Set<String> partitionStepNames = new HashSet<String>();
/**
* @return the step that will be executed by each partition
*/
public Step getStep() {
return step;
}
/**
* @return the names of each partitioned step
*/
public Collection<String> getPartitionStepNames() {
return partitionStepNames;
}
/**
* @param allowStartIfComplete flag stating if the step should restart if it
* was complete in a previous run
*/
public void setAllowStartIfComplete(boolean allowStartIfComplete) {
this.allowStartIfComplete = allowStartIfComplete;
}
@@ -156,6 +175,10 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean {
Set<StepExecution> partitionStepExecutions = splitStepExecution(stepExecution, isRestart);
for (StepExecution curStepExecution : partitionStepExecutions) {
partitionStepNames.add(curStepExecution.getStepName());
}
taskExecutor.setCorePoolSize(threads);
taskExecutor.setMaxPoolSize(threads);
@@ -186,6 +209,15 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean {
return result;
}
/**
* Blocks until all partitioned steps have completed. As each step completes
* the PartitionAnalyzer analyzes the collector data received from each
* partition (if there is any).
*
* @param tasks The {@link Future} that contains the reference to the executing step
* @param result Set of completed {@link StepExecution}s
* @throws Exception
*/
private void processPartitionResults(
final List<Future<StepExecution>> tasks,
final Set<StepExecution> result) throws Exception {
@@ -209,6 +241,16 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean {
}
}
/**
* Uses either the {@link PartitionMapper} or the hard coded configuration to split
* the supplied master StepExecution into the slave StepExecutions.
*
* @param stepExecution master {@link StepExecution}
* @param isRestart true if this step is being restarted
* @return a {@link Set} of {@link StepExecution}s to be executed
* @throws Exception
* @throws JobExecutionException
*/
private Set<StepExecution> splitStepExecution(StepExecution stepExecution,
boolean isRestart) throws Exception, JobExecutionException {
Set<StepExecution> partitionStepExecutions = new HashSet<StepExecution>();
@@ -340,7 +382,7 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean {
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(propertyContext, "A BatchPropertyContext is required");
Assert.isTrue(mapper != null || threads > 0, "Either a mapper implementation or the number of partitions/threads is required");
Assert.isTrue(mapper != null || (threads > 0 || partitions > 0), "Either a mapper implementation or the number of partitions/threads is required");
Assert.notNull(jobRepository, "A JobRepository is required");
if(partitionDataQueue == null) {

30
spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013 the original author or authors.
* Copyright 2013-2014 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.
@@ -24,10 +24,13 @@ import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.jsr.partition.JsrPartitionHandler;
import org.springframework.batch.core.jsr.partition.support.JsrStepExecutionAggregator;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.StepExecutionSplitter;
import org.springframework.batch.core.partition.support.StepExecutionAggregator;
import org.springframework.batch.core.step.NoSuchStepException;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.batch.item.ExecutionContext;
/**
@@ -38,7 +41,7 @@ import org.springframework.batch.item.ExecutionContext;
* @author Michael Minella
* @since 3.0
*/
public class PartitionStep extends org.springframework.batch.core.partition.support.PartitionStep {
public class PartitionStep extends org.springframework.batch.core.partition.support.PartitionStep implements StepLocator {
private PartitionReducer reducer;
private boolean hasReducer = false;
@@ -87,4 +90,27 @@ public class PartitionStep extends org.springframework.batch.core.partition.supp
reducer.afterPartitionedStepCompletion(PartitionStatus.COMMIT);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.core.step.StepLocator#getStepNames()
*/
@Override
public Collection<String> getStepNames() {
return ((JsrPartitionHandler) getPartitionHandler()).getPartitionStepNames();
}
/* (non-Javadoc)
* @see org.springframework.batch.core.step.StepLocator#getStep(java.lang.String)
*/
@Override
public Step getStep(String stepName) throws NoSuchStepException {
JsrPartitionHandler partitionHandler = (JsrPartitionHandler) getPartitionHandler();
Collection<String> names = partitionHandler.getPartitionStepNames();
if(names.contains(stepName)) {
return partitionHandler.getStep();
} else {
throw new NoSuchStepException(stepName + " was not found");
}
}
}

38
spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2013 the original author or authors.
* Copyright 2006-2014 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.
@@ -43,6 +43,10 @@ import org.springframework.batch.core.job.flow.support.state.EndState;
import org.springframework.batch.core.job.flow.support.state.FlowState;
import org.springframework.batch.core.job.flow.support.state.SplitState;
import org.springframework.batch.core.job.flow.support.state.StepState;
import org.springframework.batch.core.jsr.configuration.support.BatchPropertyContext;
import org.springframework.batch.core.jsr.partition.JsrPartitionHandler;
import org.springframework.batch.core.jsr.step.PartitionStep;
import org.springframework.batch.core.partition.JsrStepExecutionSplitter;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.dao.JobExecutionDao;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
@@ -551,6 +555,38 @@ public class FlowJobTests {
assertEquals("step2", step.getName());
}
@Test
public void testGetPartitionedStep() throws Exception {
SimpleFlow flow = new SimpleFlow("job");
List<StateTransition> transitions = new ArrayList<StateTransition>();
PartitionStep step = new PartitionStep();
step.setName("step1");
JsrPartitionHandler partitionHandler = new JsrPartitionHandler();
partitionHandler.setPropertyContext(new BatchPropertyContext());
partitionHandler.setPartitions(3);
partitionHandler.setJobRepository(jobRepository);
partitionHandler.setStep(new StubStep("subStep"));
partitionHandler.afterPropertiesSet();
step.setPartitionHandler(partitionHandler);
step.setStepExecutionSplitter(new JsrStepExecutionSplitter(jobRepository, false, "step1", true));
step.setJobRepository(jobRepository);
step.afterPropertiesSet();
transitions.add(StateTransition.createStateTransition(new StepState("job.step", step), "end0"));
transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
flow.setStateTransitions(transitions);
flow.afterPropertiesSet();
job.setFlow(flow);
job.afterPropertiesSet();
job.execute(jobRepository.createJobExecution("partitionJob", new JobParameters()));
assertEquals(3, step.getStepNames().size());
Step subStep = job.getStep("step1:partition0");
assertNotNull(subStep);
assertEquals("subStep", subStep.getName());
assertNull(job.getStep("step that does not exist"));
}
@Test
public void testGetStepExistsWithPrefix() throws Exception {
SimpleFlow flow = new SimpleFlow("job");