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

RESOLVED - issue BATCH-1399: Expose StepExecutionAggregator as a strategy interface in PartitionStep

Browse Source
This commit is contained in:
dsyer
2009-09-11 09:34:30 +00:00
parent 046e1f8c12
commit 0cc68d71b8
6 changed files with 107 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/DefaultStepExecutionAggregator.java Normal file
View File

@@ -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<StepExecution> 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());
}
}
}

19
spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/PartitionStep.java
View File

@@ -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<StepExecution> 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()) {

46
spring-batch-core/src/main/java/org/springframework/batch/core/partition/support/StepExecutionAggregator.java
View File

@@ -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<StepExecution> 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<StepExecution> executions);
}
}

4
spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/StepExecutionAggregatorTests.java → spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/DefaultStepExecutionAggregatorTests.java
View File

@@ -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);

25
spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/PartitionStepTests.java
View File

@@ -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<StepExecution> executions) {
super.aggregate(result, executions);
result.getExecutionContext().put("aggregated", true);
}
});
step.setStepExecutionSplitter(new SimpleStepExecutionSplitter(jobRepository, remote));
step.setPartitionHandler(new PartitionHandler() {
public Collection<StepExecution> 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"));
}
}

2
spring-batch-core/src/test/java/org/springframework/batch/core/partition/support/TaskExecutorPartitionHandlerTests.java
View File

@@ -95,7 +95,7 @@ public class TaskExecutorPartitionHandlerTests {
}
});
Collection<StepExecution> 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());
}