diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/DefaultStepExecutionAggregator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/DefaultStepExecutionAggregator.java new file mode 100644 index 000000000..b2bf2ad52 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/DefaultStepExecutionAggregator.java @@ -0,0 +1,60 @@ +/* + * 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.Collection; + +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.StepExecution; +import org.springframework.util.Assert; + +/** + * Convenience class for aggregating a set of {@link StepExecution} instances + * into a single result. + * + * @author Dave Syer + * @since 2.1 + */ +public class DefaultStepExecutionAggregator implements StepExecutionAggregator { + + /** + * Aggregates the status and exit status using their built in combination + * rules (i.e. {@link BatchStatus#max(BatchStatus, BatchStatus) max} and + * {@link ExitStatus#and(ExitStatus) and} respectively), and the counters by + * simple arithmetic. + * + * @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 || executions.isEmpty()) { + throw new IllegalArgumentException("Cannot aggregate empty or null executions: " + executions); + } + for (StepExecution stepExecution : executions) { + BatchStatus status = stepExecution.getStatus(); + result.setStatus(BatchStatus.max(result.getStatus(), status)); + result.setExitStatus(result.getExitStatus().and(stepExecution.getExitStatus())); + result.setCommitCount(result.getCommitCount() + stepExecution.getCommitCount()); + result.setRollbackCount(result.getRollbackCount() + stepExecution.getRollbackCount()); + result.setReadCount(result.getReadCount() + stepExecution.getReadCount()); + result.setReadSkipCount(result.getReadSkipCount() + stepExecution.getReadSkipCount()); + result.setWriteCount(result.getWriteCount() + stepExecution.getWriteCount()); + result.setWriteSkipCount(result.getWriteSkipCount() + stepExecution.getWriteSkipCount()); + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java index 1462cac99..0e5105b3c 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java @@ -41,16 +41,29 @@ public class PartitionStep extends AbstractStep { private PartitionHandler partitionHandler; - private StepExecutionAggregator aggregator = new StepExecutionAggregator(); + private StepExecutionAggregator stepExecutionAggregator = new DefaultStepExecutionAggregator(); /** - * Public setter for mandatory property {@link PartitionHandler}. + * A {@link PartitionHandler} which can send out step executions for remote + * processing and bring back the results. + * * @param partitionHandler the {@link PartitionHandler} to set */ public void setPartitionHandler(PartitionHandler partitionHandler) { this.partitionHandler = partitionHandler; } + /** + * A {@link StepExecutionAggregator} that can aggregate step executions when + * they come back from the handler. Defaults to a + * {@link DefaultStepExecutionAggregator}. + * + * @param stepExecutionAggregator the {@link StepExecutionAggregator} to set + */ + public void setStepExecutionAggregator(StepExecutionAggregator stepExecutionAggregator) { + this.stepExecutionAggregator = stepExecutionAggregator; + } + /** * Public setter for mandatory property {@link StepExecutionSplitter}. * @param stepExecutionSplitter the {@link StepExecutionSplitter} to set @@ -90,7 +103,7 @@ public class PartitionStep extends AbstractStep { // Wait for task completion and then aggregate the results Collection executions = partitionHandler.handle(stepExecutionSplitter, stepExecution); stepExecution.upgradeStatus(BatchStatus.COMPLETED); - aggregator.aggregate(stepExecution, executions); + stepExecutionAggregator.aggregate(stepExecution, executions); // If anything failed or had a problem we need to crap out if (stepExecution.getStatus().isUnsuccessful()) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/StepExecutionAggregator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/StepExecutionAggregator.java index a3e9ec2d9..4b5c1692a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/StepExecutionAggregator.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/StepExecutionAggregator.java @@ -1,19 +1,3 @@ -/* - * 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.Collection; @@ -21,16 +5,8 @@ import java.util.Collection; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.StepExecution; -import org.springframework.util.Assert; -/** - * Convenience class for aggregating a set of {@link StepExecution} instances - * into a single result. - * - * @author Dave Syer - * @since 2.0 - */ -public class StepExecutionAggregator { +public interface StepExecutionAggregator { /** * Take the inputs and aggregate certain fields, putting the aggregates into @@ -44,22 +20,6 @@ public class StepExecutionAggregator { * @param result the result to overwrite * @param executions the inputs */ - public void aggregate(StepExecution result, Collection executions) { - Assert.notNull(result, "To aggregate into a result it must be non-null."); - if (executions == null || executions.isEmpty()) { - throw new IllegalArgumentException("Cannot aggregate empty or null executions: " + executions); - } - for (StepExecution stepExecution : executions) { - BatchStatus status = stepExecution.getStatus(); - result.setStatus(BatchStatus.max(result.getStatus(), status)); - result.setExitStatus(result.getExitStatus().and(stepExecution.getExitStatus())); - result.setCommitCount(result.getCommitCount() + stepExecution.getCommitCount()); - result.setRollbackCount(result.getRollbackCount() + stepExecution.getRollbackCount()); - result.setReadCount(result.getReadCount() + stepExecution.getReadCount()); - result.setReadSkipCount(result.getReadSkipCount() + stepExecution.getReadSkipCount()); - result.setWriteCount(result.getWriteCount() + stepExecution.getWriteCount()); - result.setWriteSkipCount(result.getWriteSkipCount() + stepExecution.getWriteSkipCount()); - } - } + void aggregate(StepExecution result, Collection executions); -} +} \ No newline at end of file diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/StepExecutionAggregatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/DefaultStepExecutionAggregatorTests.java similarity index 92% rename from spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/StepExecutionAggregatorTests.java rename to spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/DefaultStepExecutionAggregatorTests.java index c9fff9c88..f8f76e6c7 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/StepExecutionAggregatorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/DefaultStepExecutionAggregatorTests.java @@ -12,9 +12,9 @@ import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.StepExecution; -public class StepExecutionAggregatorTests { +public class DefaultStepExecutionAggregatorTests { - private StepExecutionAggregator aggregator = new StepExecutionAggregator(); + private StepExecutionAggregator aggregator = new DefaultStepExecutionAggregator(); private JobExecution jobExecution = new JobExecution(11L); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java index c09f9f546..afeeb5c7b 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java @@ -17,6 +17,7 @@ package org.springframework.batch.core.partition.support; import static org.junit.Assert.assertEquals; +import java.util.Arrays; import java.util.Collection; import java.util.Set; @@ -128,4 +129,28 @@ public class PartitionStepTests { assertEquals(BatchStatus.STOPPED, stepExecution.getStatus()); } + @Test + public void testStepAggregator() throws Exception { + step.setStepExecutionAggregator(new DefaultStepExecutionAggregator() { + @Override + public void aggregate(StepExecution result, Collection executions) { + super.aggregate(result, executions); + result.getExecutionContext().put("aggregated", true); + } + }); + step.setStepExecutionSplitter(new SimpleStepExecutionSplitter(jobRepository, remote)); + step.setPartitionHandler(new PartitionHandler() { + public Collection handle(StepExecutionSplitter stepSplitter, StepExecution stepExecution) + throws Exception { + return Arrays.asList(stepExecution); + } + }); + step.afterPropertiesSet(); + JobExecution jobExecution = jobRepository.createJobExecution("vanillaJob", new JobParameters()); + StepExecution stepExecution = jobExecution.createStepExecution("foo"); + jobRepository.add(stepExecution); + step.execute(stepExecution); + assertEquals(true, stepExecution.getExecutionContext().get("aggregated")); + } + } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandlerTests.java index 851a164eb..390799084 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandlerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandlerTests.java @@ -95,7 +95,7 @@ public class TaskExecutorPartitionHandlerTests { } }); Collection executions = handler.handle(stepExecutionSplitter, stepExecution); - new StepExecutionAggregator().aggregate(stepExecution, executions); + new DefaultStepExecutionAggregator().aggregate(stepExecution, executions); assertEquals(1, count); assertEquals(ExitStatus.FAILED.getExitCode(), stepExecution.getExitStatus().getExitCode()); }