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

IN PROGRESS - issue BATCH-7: Remove transaction synchronization and state management from input/output sources (formerly buffering)

Browse Source 
http://jira.springframework.org/browse/BATCH-7

Remove some references to Statistics
This commit is contained in:
dsyer
2008-01-31 14:11:52 +00:00
parent acc1460fb8
commit 4267462395
10 changed files with 192 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-execution/src/main/java/org/springframework/batch/execution/bootstrap/support/SimpleExportedJobLauncher.java
View File

@@ -103,7 +103,7 @@ public class SimpleExportedJobLauncher implements ExportedJobLauncher, Initializ
int i = 0;
for (Iterator iterator = execution.getStepExecutions().iterator(); iterator.hasNext();) {
StepExecution stepExecution = (StepExecution) iterator.next();
Properties statistics = stepExecution.getStatistics();
Properties statistics = stepExecution.getStreamContext().getProperties();
for (Iterator iter = statistics.keySet().iterator(); iter.hasNext();) {
String key = (String) iter.next();
result.setProperty(prefix + "step" + i + "." + key, statistics.getProperty(key));

16
spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java
View File

@@ -161,7 +161,9 @@ public class JdbcStepDao implements StepDao, InitializingBean {
StepInstance step = new StepInstance(new Long(rs.getLong(1)));
step.setStatus(BatchStatus.getStatus(rs.getString(2)));
step.setStreamContext(new GenericStreamContext(PropertiesConverter.stringToProperties(rs.getString(3))));
step
.setStreamContext(new GenericStreamContext(PropertiesConverter.stringToProperties(rs
.getString(3))));
return step;
}
@@ -211,7 +213,8 @@ public class JdbcStepDao implements StepDao, InitializingBean {
stepExecution.setStatus(BatchStatus.getStatus(rs.getString(5)));
stepExecution.setCommitCount(rs.getInt(6));
stepExecution.setTaskCount(rs.getInt(7));
stepExecution.setStatistics(PropertiesConverter.stringToProperties(rs.getString(8)));
stepExecution.setStreamContext(new GenericStreamContext(PropertiesConverter.stringToProperties(rs
.getString(8))));
stepExecution.setExitStatus(new ExitStatus("Y".equals(rs.getString(9)), rs.getString(10), rs
.getString(11)));
return stepExecution;
@@ -243,7 +246,9 @@ public class JdbcStepDao implements StepDao, InitializingBean {
StepInstance step = new StepInstance(job, rs.getString(2), new Long(rs.getLong(1)));
String status = rs.getString(3);
step.setStatus(BatchStatus.getStatus(status));
step.setStreamContext(new GenericStreamContext(PropertiesConverter.stringToProperties(rs.getString(3))));
step
.setStreamContext(new GenericStreamContext(PropertiesConverter.stringToProperties(rs
.getString(3))));
return step;
}
};
@@ -312,7 +317,8 @@ public class JdbcStepDao implements StepDao, InitializingBean {
Object[] parameters = new Object[] { stepExecution.getId(), stepExecution.getVersion(),
stepExecution.getStepId(), stepExecution.getJobExecutionId(), stepExecution.getStartTime(),
stepExecution.getEndTime(), stepExecution.getStatus().toString(), stepExecution.getCommitCount(),
stepExecution.getTaskCount(), PropertiesConverter.propertiesToString(stepExecution.getStatistics()),
stepExecution.getTaskCount(),
PropertiesConverter.propertiesToString(stepExecution.getStreamContext().getProperties()),
stepExecution.getExitStatus().isContinuable() ? "Y" : "N", stepExecution.getExitStatus().getExitCode(),
stepExecution.getExitStatus().getExitDescription() };
jdbcTemplate.update(getSaveStepExecutionQuery(), parameters, new int[] { Types.INTEGER, Types.INTEGER,
@@ -392,7 +398,7 @@ public class JdbcStepDao implements StepDao, InitializingBean {
Integer version = new Integer(stepExecution.getVersion().intValue() + 1);
Object[] parameters = new Object[] { stepExecution.getStartTime(), stepExecution.getEndTime(),
stepExecution.getStatus().toString(), stepExecution.getCommitCount(), stepExecution.getTaskCount(),
PropertiesConverter.propertiesToString(stepExecution.getStatistics()),
PropertiesConverter.propertiesToString(stepExecution.getStreamContext().getProperties()),
stepExecution.getExitStatus().isContinuable() ? "Y" : "N",
stepExecution.getExitStatus().getExitCode(), exitDescription, version, stepExecution.getId(),
stepExecution.getVersion() };

4
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java
View File

@@ -34,6 +34,7 @@ import org.springframework.batch.execution.scope.StepSynchronizationManager;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.item.StreamContext;
import org.springframework.batch.item.stream.GenericStreamContext;
import org.springframework.batch.item.stream.SimpleStreamManager;
import org.springframework.batch.item.stream.StreamManager;
import org.springframework.batch.repeat.ExitStatus;
@@ -227,7 +228,8 @@ public class SimpleStepExecutor {
// aggregate these contributions if they
// come in asynchronously.
StreamContext statistics = stepContext.getStreamContext();
contribution.setStatistics(statistics.getProperties());
contribution.setStreamContext(new GenericStreamContext(statistics
.getProperties()));
contribution.incrementCommitCount();
// Apply the contribution to the step
// only if chunk was successful

136
spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java
View File

@@ -22,10 +22,10 @@ import java.util.Properties;
import org.springframework.batch.core.domain.BatchStatus;
import org.springframework.batch.core.domain.Job;
import org.springframework.batch.core.domain.JobSupport;
import org.springframework.batch.core.domain.JobExecution;
import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.JobParameters;
import org.springframework.batch.core.domain.JobSupport;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.core.runtime.ExitCodeExceptionClassifier;
@@ -51,15 +51,15 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
protected StepDao stepDao;
protected JobInstance jobInstance;
protected StepInstance step1;
protected StepInstance step2;
protected StepExecution stepExecution;
protected JobExecution jobExecution;
protected JobParameters jobParameters = new JobParameters();
public void setJobDao(JobDao jobDao) {
@@ -88,67 +88,68 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
step1 = stepDao.createStep(jobInstance, "TestStep1");
step2 = stepDao.createStep(jobInstance, "TestStep2");
jobExecution = new JobExecution(step2.getJobInstance());
stepExecution = new StepExecution(step1, jobExecution, null);
stepExecution.setStatus(BatchStatus.STARTED);
stepExecution.setStartTime(new Date(System.currentTimeMillis()));
stepDao.save(stepExecution);
}
public void testVersionIsNotNullForStep() throws Exception {
int version = jdbcTemplate.queryForInt("select version from BATCH_STEP_INSTANCE where ID="+step1.getId());
int version = jdbcTemplate.queryForInt("select version from BATCH_STEP_INSTANCE where ID=" + step1.getId());
assertEquals(0, version);
}
public void testVersionIsNotNullForStepExecution() throws Exception {
int version = jdbcTemplate.queryForInt("select version from BATCH_STEP_EXECUTION where ID="+stepExecution.getId());
int version = jdbcTemplate.queryForInt("select version from BATCH_STEP_EXECUTION where ID="
+ stepExecution.getId());
assertEquals(0, version);
}
public void testFindStepNull(){
public void testFindStepNull() {
StepInstance step = stepDao.findStep(jobInstance, "UnSavedStep");
assertNull(step);
}
public void testFindStep(){
public void testFindStep() {
StepInstance tempStep = stepDao.findStep(jobInstance, "TestStep1");
assertEquals(tempStep, step1);
}
public void testFindSteps(){
public void testFindSteps() {
List steps = stepDao.findSteps(jobInstance);
assertEquals(steps.size(), 2);
assertTrue(steps.contains(step1));
assertTrue(steps.contains(step2));
}
public void testFindStepsNotSaved(){
//no steps are saved for given id, empty list should be returned
public void testFindStepsNotSaved() {
// no steps are saved for given id, empty list should be returned
List steps = stepDao.findSteps(new JobInstance(new Long(38922), jobParameters));
assertEquals(steps.size(), 0);
}
public void testCreateStep(){
public void testCreateStep() {
StepInstance step3 = stepDao.createStep(jobInstance, "TestStep3");
StepInstance tempStep = stepDao.findStep(jobInstance, "TestStep3");
assertEquals(step3, tempStep);
}
public void testUpdateStepWithoutStreamContext(){
public void testUpdateStepWithoutStreamContext() {
step1.setStatus(BatchStatus.COMPLETED);
stepDao.update(step1);
StepInstance tempStep = stepDao.findStep(jobInstance, step1.getName());
assertEquals(tempStep, step1);
}
public void testUpdateStepWithStreamContext(){
public void testUpdateStepWithStreamContext() {
step1.setStatus(BatchStatus.COMPLETED);
Properties data = new Properties();
data.setProperty("restart.key1", "restartData");
@@ -157,87 +158,94 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
stepDao.update(step1);
StepInstance tempStep = stepDao.findStep(jobInstance, step1.getName());
assertEquals(tempStep, step1);
assertEquals(tempStep.getStreamContext().getProperties().toString(),
streamContext.getProperties().toString());
assertEquals(tempStep.getStreamContext().getProperties().toString(), streamContext.getProperties().toString());
}
public void testSaveStepExecution(){
public void testSaveStepExecution() {
StepExecution execution = new StepExecution(step2, jobExecution, null);
execution.setStatus(BatchStatus.STARTED);
execution.setStartTime(new Date(System.currentTimeMillis()));
Properties statistics = new Properties();
statistics.setProperty("statistic.key1", "0");
statistics.setProperty("statistic.key2", "5");
execution.setStatistics(statistics);
execution.setExitStatus(new ExitStatus(false, ExitCodeExceptionClassifier.FATAL_EXCEPTION, "java.lang.Exception"));
execution.setStreamContext(new GenericStreamContext(statistics));
execution.setExitStatus(new ExitStatus(false, ExitCodeExceptionClassifier.FATAL_EXCEPTION,
"java.lang.Exception"));
stepDao.save(execution);
List executions = stepDao.findStepExecutions(step2);
assertEquals(1, executions.size());
StepExecution tempExecution = (StepExecution)executions.get(0);
StepExecution tempExecution = (StepExecution) executions.get(0);
assertEquals(execution, tempExecution);
assertEquals(execution.getStatistics(), tempExecution.getStatistics());
assertEquals(execution.getStreamContext().getString("statistic.key1"), tempExecution.getStreamContext()
.getString("statistic.key1"));
assertEquals(execution.getExitStatus(), tempExecution.getExitStatus());
}
public void testUpdateStepExecution(){
public void testUpdateStepExecution() {
stepExecution.setStatus(BatchStatus.COMPLETED);
stepExecution.setEndTime(new Date(System.currentTimeMillis()));
stepExecution.setCommitCount(5);
stepExecution.setTaskCount(5);
stepExecution.setStatistics(new Properties());
stepExecution.setExitStatus(new ExitStatus(false, ExitCodeExceptionClassifier.FATAL_EXCEPTION, "java.lang.Exception"));
stepExecution.setStreamContext(new StreamContext());
stepExecution.setExitStatus(new ExitStatus(false, ExitCodeExceptionClassifier.FATAL_EXCEPTION,
"java.lang.Exception"));
stepDao.update(stepExecution);
List executions = stepDao.findStepExecutions(step1);
assertEquals(1, executions.size());
StepExecution tempExecution = (StepExecution)executions.get(0);
StepExecution tempExecution = (StepExecution) executions.get(0);
assertEquals(stepExecution, tempExecution);
assertEquals(stepExecution.getExitStatus(), tempExecution.getExitStatus());
}
public void testUpdateStepExecutionWithNullId(){
public void testUpdateStepExecutionWithNullId() {
StepExecution stepExecution = new StepExecution(null, null, null);
try{
try {
stepDao.update(stepExecution);
fail("Expected IllegalArgumentException");
}catch(IllegalArgumentException ex){
//expected
}
catch (IllegalArgumentException ex) {
// expected
}
}
public void testGetStepExecutionCountForNoExecutions(){
public void testGetStepExecutionCountForNoExecutions() {
int executionCount = stepDao.getStepExecutionCount(step2);
assertEquals(executionCount, 0);
}
public void testIncrementStepExecutionCount(){
public void testIncrementStepExecutionCount() {
assertEquals(1, stepDao.getStepExecutionCount(step1));
StepExecution execution = new StepExecution(step1, new JobExecution(step1.getJobInstance(), new Long(123)), null);
StepExecution execution = new StepExecution(step1, new JobExecution(step1.getJobInstance(), new Long(123)),
null);
stepDao.save(execution);
assertEquals(2, stepDao.getStepExecutionCount(step1));
}
public void testUpdateStepExecutionVersion() throws Exception {
int before = stepExecution.getVersion().intValue();
stepDao.update(stepExecution);
int after = stepExecution.getVersion().intValue();
assertEquals("StepExecution version not updated", before+1, after);
assertEquals("StepExecution version not updated", before + 1, after);
}
public void testUpdateStepExecutionOptimisticLocking() throws Exception {
stepExecution.incrementVersion(); // not really allowed outside dao code
stepExecution.incrementVersion(); // not really allowed outside dao
// code
try {
stepDao.update(stepExecution);
fail("Expected OptimisticLockingFailureException");
}
catch (OptimisticLockingFailureException e) {
// expected
assertTrue("Exception message should contain step execution id: "+e.getMessage(), e.getMessage().indexOf(""+stepExecution.getId())>=0);
assertTrue("Exception message should contain step execution version: "+e.getMessage(), e.getMessage().indexOf(""+stepExecution.getVersion())>=0);
assertTrue("Exception message should contain step execution id: " + e.getMessage(), e.getMessage().indexOf(
"" + stepExecution.getId()) >= 0);
assertTrue("Exception message should contain step execution version: " + e.getMessage(), e.getMessage()
.indexOf("" + stepExecution.getVersion()) >= 0);
}
}
}

4
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java
View File

@@ -409,7 +409,7 @@ public class SimpleStepExecutorTests extends TestCase {
JobExecution jobExecution = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(step, jobExecution);
assertEquals(null, stepExecution.getStatistics().getProperty("foo"));
assertEquals(null, stepExecution.getStreamContext().getString("foo"));
final Map map = new HashMap();
stepExecutor.setStreamManager(new SimpleStreamManager() {
@@ -423,7 +423,7 @@ public class SimpleStepExecutorTests extends TestCase {
// At least once in that process the statistics service was asked for
// statistics...
assertEquals("bar", stepExecution.getStatistics().getProperty("foo"));
assertEquals("bar", stepExecution.getStreamContext().getString("foo"));
// ...but nothing was registered because nothing with step scoped.
assertEquals(0, map.size());
}