From b0c9af5e8ef73b4565f323449c034c49236a431e Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Mon, 6 Jan 2014 16:19:31 -0600 Subject: [PATCH] Fixes for stopping a partitioned step * 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 --- .../xml/SimpleFlowFactoryBean.java | 4 +- .../batch/core/job/SimpleJob.java | 13 +++++- .../batch/core/job/flow/FlowJob.java | 15 +++---- .../job/flow/support/state/StepState.java | 44 +++++++++++++++++-- .../jsr/job/flow/support/state/EndState.java | 21 ++------- .../jsr/partition/JsrPartitionHandler.java | 44 ++++++++++++++++++- .../batch/core/jsr/step/PartitionStep.java | 30 ++++++++++++- .../batch/core/job/flow/FlowJobTests.java | 38 +++++++++++++++- 8 files changed, 174 insertions(+), 35 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java index 38cb5b1f0..99ad573e2 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java @@ -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); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java index 3ea13e82a..ee75114b4 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java @@ -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 names = new ArrayList(); 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; 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 a3df9939a..5af9a71e0 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 @@ -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 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)); - } - } } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/StepState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/StepState.java index b319908b4..15e213d36 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/StepState.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/support/state/StepState.java @@ -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 getStepNames() { + List names = new ArrayList(); + + 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; + } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/support/state/EndState.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/support/state/EndState.java index 7a7833055..a863fee2f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/support/state/EndState.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/job/flow/support/state/EndState.java @@ -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()); - } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java index 00bc75c81..b7267ac37 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/partition/JsrPartitionHandler.java @@ -74,7 +74,26 @@ public class JsrPartitionHandler implements PartitionHandler, InitializingBean { private BatchPropertyContext propertyContext; private JobRepository jobRepository; private boolean allowStartIfComplete = false; + private Set partitionStepNames = new HashSet(); + /** + * @return the step that will be executed by each partition + */ + public Step getStep() { + return step; + } + + /** + * @return the names of each partitioned step + */ + public Collection 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 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> tasks, final Set 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 splitStepExecution(StepExecution stepExecution, boolean isRestart) throws Exception, JobExecutionException { Set partitionStepExecutions = new HashSet(); @@ -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) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java index ffde7aba7..201696d39 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/jsr/step/PartitionStep.java @@ -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 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 names = partitionHandler.getPartitionStepNames(); + + if(names.contains(stepName)) { + return partitionHandler.getStep(); + } else { + throw new NoSuchStepException(stepName + " was not found"); + } + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java index 41b58de0f..0ade6b988 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java @@ -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 transitions = new ArrayList(); + 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");