diff --git a/spring-batch-core/.settings/com.springsource.sts.config.flow.prefs b/spring-batch-core/.settings/com.springsource.sts.config.flow.prefs index 7d8c1c6cd..5789486ba 100644 --- a/spring-batch-core/.settings/com.springsource.sts.config.flow.prefs +++ b/spring-batch-core/.settings/com.springsource.sts.config.flow.prefs @@ -1,7 +1,7 @@ -#Tue Mar 15 12:05:10 GMT 2011 +#Tue Mar 15 13:48:56 GMT 2011 //com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/JobExecutionListenerMethodAttributeParserTests-context.xml=\n\n\n\n\n\n //com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/JobRepositoryDefaultParserTests-context.xml=\n -//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepParserTests-context.xml=\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepParserTests-context.xml=\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n //com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepListenerInStepParserTests-context.xml=\n\n\n\n\n\n //com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepListenerMethodAttributeParserTests-context.xml=\n\n\n\n\n\n //com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepListenerParserTests-context.xml=\n\n\n\n\n\n diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java index 8cfc4123a..60c3be298 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java @@ -65,6 +65,8 @@ public abstract class AbstractStepParser { private static final String PARTITIONER_ATTR = "partitioner"; + private static final String AGGREGATOR_ATTR = "aggregator"; + private static final String HANDLER_ATTR = "handler"; private static final String HANDLER_ELE = "handler"; @@ -150,6 +152,7 @@ public abstract class AbstractStepParser { bd.setAttribute("isNamespaceStep", true); String stepRef = partitionElement.getAttribute(STEP_ATTR); String partitionerRef = partitionElement.getAttribute(PARTITIONER_ATTR); + String aggregatorRef = partitionElement.getAttribute(AGGREGATOR_ATTR); String handlerRef = partitionElement.getAttribute(HANDLER_ATTR); if (!StringUtils.hasText(partitionerRef)) { @@ -160,6 +163,9 @@ public abstract class AbstractStepParser { MutablePropertyValues propertyValues = bd.getPropertyValues(); propertyValues.addPropertyValue("partitioner", new RuntimeBeanReference(partitionerRef)); + if (StringUtils.hasText(aggregatorRef)) { + propertyValues.addPropertyValue("stepExecutionAggregator", new RuntimeBeanReference(aggregatorRef)); + } boolean customHandler = false; if (!StringUtils.hasText(handlerRef)) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java index 9544aac34..c5ec72977 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java @@ -38,6 +38,7 @@ import org.springframework.batch.core.partition.PartitionHandler; import org.springframework.batch.core.partition.support.PartitionStep; import org.springframework.batch.core.partition.support.Partitioner; import org.springframework.batch.core.partition.support.SimpleStepExecutionSplitter; +import org.springframework.batch.core.partition.support.StepExecutionAggregator; import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.step.AbstractStep; @@ -201,6 +202,8 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { // private boolean hasChunkElement = false; + private StepExecutionAggregator stepExecutionAggregator; + /** * Create a {@link Step} from the configuration provided. * @@ -319,7 +322,7 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { + "It should be configured with its own step reference."); } - boolean allowStartIfComplete = this.allowStartIfComplete !=null ? this.allowStartIfComplete : false; + boolean allowStartIfComplete = this.allowStartIfComplete != null ? this.allowStartIfComplete : false; String name = this.name; if (step != null) { try { @@ -328,12 +331,15 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { } catch (Exception e) { logger.info("Ignored exception from step asking for name and allowStartIfComplete flag. " - + "Using default from enclosing PartitionStep ("+name+","+allowStartIfComplete+")."); + + "Using default from enclosing PartitionStep (" + name + "," + allowStartIfComplete + ")."); } } SimpleStepExecutionSplitter splitter = new SimpleStepExecutionSplitter(jobRepository, allowStartIfComplete, name, partitioner); ts.setStepExecutionSplitter(splitter); + if (stepExecutionAggregator != null) { + ts.setStepExecutionAggregator(stepExecutionAggregator); + } } private void configureSimple(SimpleStepFactoryBean fb) { @@ -644,6 +650,13 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { public void setPartitioner(Partitioner partitioner) { this.partitioner = partitioner; } + + /** + * @param stepExecutionAggregator the stepExecutionAggregator to set + */ + public void setStepExecutionAggregator(StepExecutionAggregator stepExecutionAggregator) { + this.stepExecutionAggregator = stepExecutionAggregator; + } /** * @param partitionHandler the partitionHandler to set diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/RemoteStepExecutionAggregator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/RemoteStepExecutionAggregator.java new file mode 100644 index 000000000..109a93acf --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/RemoteStepExecutionAggregator.java @@ -0,0 +1,102 @@ +/* + * Copyright 2006-2007 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.partition.support; + +import java.util.ArrayList; +import java.util.Collection; + +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.Assert; + +/** + * Convenience class for aggregating a set of {@link StepExecution} instances + * when the input comes from remote steps, so the data need to be refreshed from + * the repository. + * + * @author Dave Syer + * @since 2.1 + */ +public class RemoteStepExecutionAggregator implements StepExecutionAggregator, InitializingBean { + + private StepExecutionAggregator delegate = new DefaultStepExecutionAggregator(); + + private JobExplorer jobExplorer; + + /** + * Create a new instance (useful for configuration purposes). + */ + public RemoteStepExecutionAggregator() { + } + + /** + * Create a new instance with a job explorer that can be used to refresh the + * data when aggregating. + * + * @param jobExplorer the {@link JobExplorer} to use + */ + public RemoteStepExecutionAggregator(JobExplorer jobExplorer) { + super(); + this.jobExplorer = jobExplorer; + } + + /** + * @param jobExplorer the jobExplorer to set + */ + public void setJobExplorer(JobExplorer jobExplorer) { + this.jobExplorer = jobExplorer; + } + + /** + * @param delegate the delegate to set + */ + public void setDelegate(StepExecutionAggregator delegate) { + this.delegate = delegate; + } + + /** + * @throws Exception if the job explorer is not provided + */ + public void afterPropertiesSet() throws Exception { + Assert.state(jobExplorer != null, "A JobExplorer must be provided"); + } + + /** + * Aggregates the input executions into the result {@link StepExecution} + * delegating to the delegate aggregator once the input has been refreshed + * from the {@link JobExplorer}. + * + * @see StepExecutionAggregator #aggregate(StepExecution, Collection) + */ + public void aggregate(StepExecution result, Collection executions) { + Assert.notNull(result, "To aggregate into a result it must be non-null."); + if (executions == null) { + return; + } + Collection updates = new ArrayList(); + for (StepExecution stepExecution : executions) { + Long id = stepExecution.getId(); + Assert.state(id != null, "StepExecution has null id. It must be saved first: " + stepExecution); + StepExecution update = jobExplorer.getStepExecution(stepExecution.getJobExecutionId(), id); + Assert.state(update != null, "Could not reload StepExecution from JobRepository: " + stepExecution); + updates.add(update); + } + delegate.aggregate(result, updates); + } + +} diff --git a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd index e885541f8..5593eba61 100644 --- a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd +++ b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd @@ -498,6 +498,18 @@ ref" is not required, and only needs to be specified explicitly + + + + + + + + + + + stepNames = getStepNames(jobExecution); - assertEquals(1, stepNames.size()); - assertEquals("[j5s1]", stepNames.toString()); + assertEquals(3, stepNames.size()); + assertEquals("[s1, step1:partition0, step1:partition1]", stepNames.toString()); + assertEquals("bar", jobExecution.getExecutionContext().get("foo")); } - /** - * BATCH-1509 we now support the ability define steps inline for partitioned - * steps. this demonstates that the execution proceeds as expected and that - * the partitionhandler has a reference to the inline step definition - */ @Test - public void testNestedPartitionStep() throws Throwable { - assertNotNull("the reference to the job4 configured in the XML file must not be null", job4); - JobExecution jobExecution = jobRepository.createJobExecution(job4.getName(), new JobParameters()); - - job4.execute(jobExecution); - - for (StepExecution se : jobExecution.getStepExecutions()) { - String stepExecutionName = se.getStepName(); - if (stepExecutionName.equalsIgnoreCase("j4s1")) { // the partitioned - // step - PartitionStep partitionStep = (PartitionStep) this.applicationContext.getBean(stepExecutionName); - - // prove that the reference in the {@link - // TaskExecutorPartitionHandler} is the step configured inline - TaskExecutorPartitionHandler taskExecutorPartitionHandler = accessPrivateField(partitionStep, - "partitionHandler"); - TaskletStep taskletStep = accessPrivateField(taskExecutorPartitionHandler, "step"); - - assertNotNull("the taskletStep wasn't configured with a step. " - + "We're trusting that the factory ensured " + "a reference was given.", taskletStep); - } - } + public void testHandlerRefStep() throws Exception { + assertNotNull(job2); + JobExecution jobExecution = jobRepository.createJobExecution(job2.getName(), new JobParameters()); + job2.execute(jobExecution); assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); - // Step names not saved by this one (it geosn't have that tasklet) - assertEquals("[]", savedStepNames.toString()); + Collections.sort(savedStepNames); + assertEquals("[s3, step1:partition0, step1:partition1, step1:partition2]", savedStepNames.toString()); List stepNames = getStepNames(jobExecution); - assertEquals(4, stepNames.size()); - assertEquals("[j4s1, j4s1:partition0, j4s1:partition1, j4s1:partition2]", stepNames.toString()); + assertEquals(5, stepNames.size()); + assertEquals("[s2, s3, step1:partition0, step1:partition1, step1:partition2]", stepNames.toString()); } /** @@ -190,30 +172,51 @@ public class PartitionStepParserTests implements ApplicationContextAware { assertEquals("[j3s1, j3s1:partition0, j3s1:partition1, j3s1:partition2]", stepNames.toString()); } + /** + * BATCH-1509 we now support the ability define steps inline for partitioned + * steps. this demonstates that the execution proceeds as expected and that + * the partitionhandler has a reference to the inline step definition + */ @Test - public void testDefaultHandlerStep() throws Exception { - assertNotNull(job1); - JobExecution jobExecution = jobRepository.createJobExecution(job1.getName(), new JobParameters()); - job1.execute(jobExecution); + public void testNestedPartitionStep() throws Throwable { + assertNotNull("the reference to the job4 configured in the XML file must not be null", job4); + JobExecution jobExecution = jobRepository.createJobExecution(job4.getName(), new JobParameters()); + + job4.execute(jobExecution); + + for (StepExecution se : jobExecution.getStepExecutions()) { + String stepExecutionName = se.getStepName(); + if (stepExecutionName.equalsIgnoreCase("j4s1")) { // the partitioned + // step + PartitionStep partitionStep = (PartitionStep) this.applicationContext.getBean(stepExecutionName); + + // prove that the reference in the {@link + // TaskExecutorPartitionHandler} is the step configured inline + TaskExecutorPartitionHandler taskExecutorPartitionHandler = accessPrivateField(partitionStep, + "partitionHandler"); + TaskletStep taskletStep = accessPrivateField(taskExecutorPartitionHandler, "step"); + + assertNotNull("the taskletStep wasn't configured with a step. " + + "We're trusting that the factory ensured " + "a reference was given.", taskletStep); + } + } assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); - Collections.sort(savedStepNames); - assertEquals("[step1:partition0, step1:partition1]", savedStepNames.toString()); + // Step names not saved by this one (it geosn't have that tasklet) + assertEquals("[]", savedStepNames.toString()); List stepNames = getStepNames(jobExecution); - assertEquals(3, stepNames.size()); - assertEquals("[s1, step1:partition0, step1:partition1]", stepNames.toString()); + assertEquals(4, stepNames.size()); + assertEquals("[j4s1, j4s1:partition0, j4s1:partition1, j4s1:partition2]", stepNames.toString()); } @Test - public void testHandlerRefStep() throws Exception { - assertNotNull(job2); - JobExecution jobExecution = jobRepository.createJobExecution(job2.getName(), new JobParameters()); - job2.execute(jobExecution); + public void testCustomHandlerRefStep() throws Exception { + assertNotNull(job5); + JobExecution jobExecution = jobRepository.createJobExecution(job5.getName(), new JobParameters()); + job5.execute(jobExecution); assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); - Collections.sort(savedStepNames); - assertEquals("[s3, step1:partition0, step1:partition1, step1:partition2]", savedStepNames.toString()); List stepNames = getStepNames(jobExecution); - assertEquals(5, stepNames.size()); - assertEquals("[s2, s3, step1:partition0, step1:partition1, step1:partition2]", stepNames.toString()); + assertEquals(1, stepNames.size()); + assertEquals("[j5s1]", stepNames.toString()); } private List getStepNames(JobExecution jobExecution) { @@ -234,4 +237,11 @@ public class PartitionStepParserTests implements ApplicationContextAware { } + public static class CustomStepExecutionAggregator implements StepExecutionAggregator { + + public void aggregate(StepExecution result, Collection executions) { + result.getJobExecution().getExecutionContext().put("foo", "bar"); + } + + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/RemoteStepExecutionAggregatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/RemoteStepExecutionAggregatorTests.java new file mode 100644 index 000000000..321b73394 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/RemoteStepExecutionAggregatorTests.java @@ -0,0 +1,74 @@ +package org.springframework.batch.core.partition.support; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.Arrays; +import java.util.Collections; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.explore.JobExplorer; +import org.springframework.batch.core.explore.support.MapJobExplorerFactoryBean; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; + +public class RemoteStepExecutionAggregatorTests { + + private RemoteStepExecutionAggregator aggregator = new RemoteStepExecutionAggregator(); + + private JobExecution jobExecution; + + private StepExecution result; + + private StepExecution stepExecution1; + + private StepExecution stepExecution2; + + @Before + public void init() throws Exception { + MapJobRepositoryFactoryBean factory = new MapJobRepositoryFactoryBean(); + JobRepository jobRepository = (JobRepository) factory.getObject(); + aggregator.setJobExplorer((JobExplorer) new MapJobExplorerFactoryBean(factory).getObject()); + jobExecution = jobRepository.createJobExecution("job", new JobParameters()); + result = jobExecution.createStepExecution("aggregate"); + stepExecution1 = jobExecution.createStepExecution("foo:1"); + stepExecution2 = jobExecution.createStepExecution("foo:2"); + jobRepository.add(stepExecution1); + jobRepository.add(stepExecution2); + } + + @Test + public void testAggregateEmpty() { + aggregator.aggregate(result, Collections. emptySet()); + } + + @Test + public void testAggregateNull() { + aggregator.aggregate(result, null); + } + + @Test + public void testAggregateStatusSunnyDay() { + stepExecution1.setStatus(BatchStatus.COMPLETED); + stepExecution2.setStatus(BatchStatus.COMPLETED); + aggregator.aggregate(result, Arrays. asList(stepExecution1, stepExecution2)); + assertNotNull(result); + assertEquals(BatchStatus.STARTING, result.getStatus()); + } + + @Test(expected=IllegalStateException.class) + public void testAggregateStatusMissingExecution() { + stepExecution2 = jobExecution.createStepExecution("foo:3"); + stepExecution1.setStatus(BatchStatus.COMPLETED); + stepExecution2.setStatus(BatchStatus.COMPLETED); + aggregator.aggregate(result, Arrays. asList(stepExecution1, stepExecution2)); + assertNotNull(result); + assertEquals(BatchStatus.STARTING, result.getStatus()); + } + +} diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepParserTests-context.xml index ca3be0e34..613040b89 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepParserTests-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/PartitionStepParserTests-context.xml @@ -8,7 +8,7 @@ - + @@ -53,6 +53,9 @@ + +