From 625480382c7bdf9626ae2803de4ecc2f3abfc708 Mon Sep 17 00:00:00 2001 From: lucasward Date: Tue, 5 Feb 2008 02:32:48 +0000 Subject: [PATCH] BATCH-219 and BATCH-330: Created ExecutionAttributes, which are stored in the table BATCH_STEP_EXECUTION_ATTRIBUTES. --- .../repository/SimpleJobRepository.java | 8 +- .../execution/repository/dao/JdbcStepDao.java | 269 +++++++++++++++++- .../execution/repository/dao/MapStepDao.java | 18 +- .../execution/repository/dao/StepDao.java | 31 ++ .../execution/scope/SimpleStepContext.java | 8 +- .../batch/execution/scope/StepContext.java | 4 +- .../step/simple/SimpleStepExecutor.java | 4 +- .../src/main/resources/schema-db2.sql | 15 +- .../src/main/resources/schema-derby.sql | 15 +- .../src/main/resources/schema-hsqldb.sql | 15 +- .../src/main/resources/schema-mysql.sql | 15 +- .../src/main/resources/schema-oracle10g.sql | 15 +- .../src/main/resources/schema-postgresql.sql | 15 +- .../src/main/sql/db2.properties | 2 + .../src/main/sql/derby.properties | 2 + .../src/main/sql/destroy.sql.vpp | 3 +- .../src/main/sql/hsqldb.properties | 2 + .../src/main/sql/init.sql.vpp | 12 +- .../src/main/sql/mysql.properties | 2 + .../src/main/sql/oracle10g.properties | 2 + .../src/main/sql/postgresql.properties | 2 + .../SimpleExportedJobLauncherTests.java | 3 +- .../execution/repository/MockStepDao.java | 13 + .../repository/SimpleJobRepositoryTests.java | 11 +- .../repository/dao/AbstractStepDaoTests.java | 34 ++- .../repository/dao/MapStepDaoTests.java | 10 +- .../scope/SimpleStepContextTests.java | 10 +- .../step/simple/SimpleStepExecutorTests.java | 14 +- .../execution/repository/dao/destroy.sql | 3 +- .../batch/execution/repository/dao/init.sql | 36 ++- 30 files changed, 518 insertions(+), 75 deletions(-) diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java index b2368786a..cd2411499 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java @@ -33,7 +33,7 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.execution.repository.dao.JobDao; import org.springframework.batch.execution.repository.dao.StepDao; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.transaction.annotation.Isolation; import org.springframework.util.Assert; @@ -255,10 +255,12 @@ public class SimpleJobRepository implements JobRepository { if (stepExecution.getId() == null) { // new execution, obtain id and insert stepDao.save(stepExecution); + stepDao.save(stepExecution.getId(), stepExecution.getStreamContext()); } else { // existing execution, update stepDao.update(stepExecution); + stepDao.update(stepExecution.getId(), stepExecution.getStreamContext()); } } @@ -303,7 +305,7 @@ public class SimpleJobRepository implements JobRepository { StepInstance stepInstance = stepDao.createStep(job, step.getName()); // Ensure valid restart data is being returned. if (stepInstance.getStreamContext() == null || stepInstance.getStreamContext() == null) { - stepInstance.setStreamContext(new StreamContext()); + stepInstance.setStreamContext(new ExecutionAttributes()); } stepInstances.add(stepInstance); } @@ -326,7 +328,7 @@ public class SimpleJobRepository implements JobRepository { step.setStepExecutionCount(stepDao.getStepExecutionCount(step)); // Ensure valid restart data is being returned. if (step.getStreamContext() == null || step.getStreamContext() == null) { - step.setStreamContext(new StreamContext()); + step.setStreamContext(new ExecutionAttributes()); } stepInstances.add(step); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java index b91a304ce..86bade7cb 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/JdbcStepDao.java @@ -16,12 +16,17 @@ package org.springframework.batch.execution.repository.dao; +import java.io.Serializable; +import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Types; +import java.util.Iterator; import java.util.List; import java.util.Properties; +import java.util.Map.Entry; +import org.apache.commons.lang.SerializationUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.domain.BatchStatus; @@ -30,22 +35,30 @@ import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; import org.springframework.batch.execution.repository.dao.JdbcJobDao.JobExecutionRowMapper; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.io.exception.BatchCriticalException; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.support.PropertiesConverter; import org.springframework.beans.factory.InitializingBean; +import org.springframework.dao.DataAccessException; import org.springframework.dao.IncorrectResultSizeDataAccessException; import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.jdbc.core.JdbcOperations; +import org.springframework.jdbc.core.PreparedStatementCallback; +import org.springframework.jdbc.core.RowCallbackHandler; import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.core.support.AbstractLobCreatingPreparedStatementCallback; import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer; +import org.springframework.jdbc.support.lob.DefaultLobHandler; +import org.springframework.jdbc.support.lob.LobCreator; +import org.springframework.jdbc.support.lob.LobHandler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** * Jdbc implementation of {@link StepDao}.
* - * Allows customisation of the tables names used by Spring Batch for step meta + * Allows customization of the tables names used by Spring Batch for step meta * data via a prefix property.
* * Uses sequences or tables (via Spring's {@link DataFieldMaxValueIncrementer} @@ -57,6 +70,7 @@ import org.springframework.util.StringUtils; * * @author Lucas Ward * @author Dave Syer + * * @see StepDao */ public class JdbcStepDao implements StepDao, InitializingBean { @@ -90,6 +104,15 @@ public class JdbcStepDao implements StepDao, InitializingBean { + "STATUS = ?, COMMIT_COUNT = ?, TASK_COUNT = ?, TASK_STATISTICS = ?, CONTINUABLE = ? , EXIT_CODE = ?, " + "EXIT_MESSAGE = ?, VERSION = ? where ID = ? and VERSION = ?"; + private static final String UPDATE_STEP_EXECUTION_ATTRS = "UPDATE %PREFIX%STEP_EXECUTION_ATTRS set " + + "TYPE_CD = ?, STRING_VAL = ?, DOUBLE_VAL = ?, LONG_VAL = ?, OBJECT_VAL = ? where EXECUTION_ID = ? and KEY_NAME = ?"; + + private static final String INSERT_STEP_EXECUTION_ATTRS = "INSERT into %PREFIX%STEP_EXECUTION_ATTRS(EXECUTION_ID, TYPE_CD," + + " KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL) values(?,?,?,?,?,?,?)"; + + private static final String FIND_STEP_EXECUTION_ATTRS = "SELECT TYPE_CD, KEY_NAME, STRING_VAL, DOUBLE_VAL, LONG_VAL, OBJECT_VAL " + + "from %PREFIX%STEP_EXECUTION_ATTRS where EXECUTION_ID = ?"; + private JdbcOperations jdbcTemplate; private JobDao jobDao; @@ -97,6 +120,8 @@ public class JdbcStepDao implements StepDao, InitializingBean { private DataFieldMaxValueIncrementer stepExecutionIncrementer; private DataFieldMaxValueIncrementer stepIncrementer; + + private LobHandler lobHandler = new DefaultLobHandler(); private String tablePrefix = JdbcJobDao.DEFAULT_TABLE_PREFIX; @@ -160,7 +185,7 @@ 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 StreamContext(PropertiesConverter.stringToProperties(rs.getString(3)))); + step.setStreamContext(new ExecutionAttributes(PropertiesConverter.stringToProperties(rs.getString(3)))); return step; } @@ -210,7 +235,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { stepExecution.setStatus(BatchStatus.getStatus(rs.getString(5))); stepExecution.setCommitCount(rs.getInt(6)); stepExecution.setTaskCount(rs.getInt(7)); - stepExecution.setStreamContext(new StreamContext(PropertiesConverter + stepExecution.setStreamContext(new ExecutionAttributes(PropertiesConverter .stringToProperties(rs.getString(8)))); stepExecution.setExitStatus(new ExitStatus("Y".equals(rs.getString(9)), rs.getString(10), rs .getString(11))); @@ -221,6 +246,156 @@ public class JdbcStepDao implements StepDao, InitializingBean { return jdbcTemplate.query(getFindStepExecutionsQuery(), new Object[] { step.getId() }, rowMapper); } + + /* + * Insert execution attributes. A lob creator must be used, since any attributes + * that don't match a provided type must be serialized into a blob. + */ + public void save(final Long executionId, final ExecutionAttributes executionAttributes){ + + Assert.notNull(executionId, "ExecutionId must not be null."); + Assert.notNull(executionAttributes, "The ExecutionAttributes must not be null."); + + for(Iterator it = executionAttributes.entrySet().iterator();it.hasNext();){ + Entry entry = (Entry)it.next(); + final String key = entry.getKey().toString(); + final Object value = entry.getValue(); + + if(value instanceof String){ + insertExecutionAttribute(executionId, key, value, AttributeType.STRING); + } + else if(value instanceof Double){ + insertExecutionAttribute(executionId, key, value, AttributeType.DOUBLE); + } + else if(value instanceof Long){ + insertExecutionAttribute(executionId, key, value, AttributeType.LONG); + } + else + { + insertExecutionAttribute(executionId, key, value, AttributeType.OBJECT); + } + } + } + + private void insertExecutionAttribute(final Long executionId, final String key, final Object value, final AttributeType type){ + PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler){ + + protected void setValues(PreparedStatement ps, LobCreator lobCreator) + throws SQLException, DataAccessException { + + ps.setLong(1, executionId.longValue()); + ps.setString(3, key); + if(type == AttributeType.STRING){ + ps.setString(2,AttributeType.STRING.toString()); + ps.setString(4,value.toString()); + ps.setDouble(5, 0.0); + ps.setLong(6, 0); + lobCreator.setBlobAsBytes(ps, 7, null); + } + else if(type == AttributeType.DOUBLE){ + ps.setString(2,AttributeType.DOUBLE.toString()); + ps.setString(4,null); + ps.setDouble(5, ((Double)value).doubleValue()); + ps.setLong(6, 0); + lobCreator.setBlobAsBytes(ps, 7, null); + } + else if(type == AttributeType.LONG){ + ps.setString(2,AttributeType.LONG.toString()); + ps.setString(4,null); + ps.setDouble(5, 0.0 ); + ps.setLong(6, ((Long)value).longValue()); + lobCreator.setBlobAsBytes(ps, 7, null); + } + else{ + ps.setString(2,AttributeType.OBJECT.toString()); + ps.setString(4,null); + ps.setDouble(5, 0.0 ); + ps.setLong(6, 0); + lobCreator.setBlobAsBytes(ps, 7, SerializationUtils.serialize((Serializable)value)); + } + }}; + + jdbcTemplate.execute(getQuery(INSERT_STEP_EXECUTION_ATTRS), callback); + } + + /** + * update execution attributes. A lob creator must be used, since any attributes + * that don't match a provided type must be serialized into a blob. + * + * @see {@link LobCreator} + */ + public void update(final Long executionId, ExecutionAttributes executionAttributes){ + + Assert.notNull(executionId, "ExecutionId must not be null."); + Assert.notNull(executionAttributes, "The ExecutionAttributes must not be null."); + + for(Iterator it = executionAttributes.entrySet().iterator();it.hasNext();){ + Entry entry = (Entry)it.next(); + final String key = entry.getKey().toString(); + final Object value = entry.getValue(); + + if(value instanceof String){ + updateExecutionAttribute(executionId, key, value, AttributeType.STRING); + } + else if(value instanceof Double){ + updateExecutionAttribute(executionId, key, value, AttributeType.DOUBLE); + } + else if(value instanceof Long){ + updateExecutionAttribute(executionId, key, value, AttributeType.LONG); + } + else + { + updateExecutionAttribute(executionId, key, value, AttributeType.OBJECT); + } + } + } + + private void updateExecutionAttribute(final Long executionId, final String key, final Object value, final AttributeType type){ + + PreparedStatementCallback callback = new AbstractLobCreatingPreparedStatementCallback(lobHandler){ + + protected void setValues(PreparedStatement ps, LobCreator lobCreator) + throws SQLException, DataAccessException { + + ps.setLong(6, executionId.longValue()); + ps.setString(7, key); + if(type == AttributeType.STRING){ + ps.setString(1,AttributeType.STRING.toString()); + ps.setString(2,value.toString()); + ps.setDouble(3, 0.0); + ps.setLong(4, 0); + lobCreator.setBlobAsBytes(ps, 5, null); + } + else if(type == AttributeType.DOUBLE){ + ps.setString(1,AttributeType.DOUBLE.toString()); + ps.setString(2,null); + ps.setDouble(3, ((Double)value).doubleValue()); + ps.setLong(4, 0); + lobCreator.setBlobAsBytes(ps, 5, null); + } + else if(type == AttributeType.LONG){ + ps.setString(1,AttributeType.LONG.toString()); + ps.setString(2,null); + ps.setDouble(3, 0.0 ); + ps.setLong(4, ((Long)value).longValue()); + lobCreator.setBlobAsBytes(ps, 5, null); + } + else{ + ps.setString(1,AttributeType.OBJECT.toString()); + ps.setString(2,null); + ps.setDouble(3, 0.0 ); + ps.setLong(4, 0); + lobCreator.setBlobAsBytes(ps, 5, SerializationUtils.serialize((Serializable)value)); + } + }}; + + //LobCreating callbacks always return the affect row count for SQL DML statements, if less than 1 row + //is affected, then this row is new and should be inserted. + Integer affectedRows = (Integer)jdbcTemplate.execute(getQuery(UPDATE_STEP_EXECUTION_ATTRS), callback); + if(affectedRows.intValue() < 1){ + insertExecutionAttribute(executionId, key, value, type); + } + } /** * @see StepDao#findSteps(JobInstance) @@ -244,7 +419,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { String status = rs.getString(3); step.setStatus(BatchStatus.getStatus(status)); step - .setStreamContext(new StreamContext(PropertiesConverter.stringToProperties(rs + .setStreamContext(new ExecutionAttributes(PropertiesConverter.stringToProperties(rs .getString(3)))); return step; } @@ -321,7 +496,6 @@ public class JdbcStepDao implements StepDao, InitializingBean { jdbcTemplate.update(getSaveStepExecutionQuery(), parameters, new int[] { Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.INTEGER, Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER, Types.INTEGER, Types.VARCHAR, Types.CHAR, Types.VARCHAR, Types.VARCHAR }); - } public void setJdbcTemplate(JdbcOperations jdbcTemplate) { @@ -408,11 +582,48 @@ public class JdbcStepDao implements StepDao, InitializingBean { throw new OptimisticLockingFailureException("Attempt to update step execution id=" + stepExecution.getId() + " with out of date version (" + stepExecution.getVersion() + ")"); } - + stepExecution.incrementVersion(); - + } } + + public ExecutionAttributes findExecutionAttributes(final Long executionId){ + + Assert.notNull(executionId, "ExecutionId must not be null."); + + final ExecutionAttributes executionAttributes = new ExecutionAttributes(); + + RowCallbackHandler callback = new RowCallbackHandler(){ + + public void processRow(ResultSet rs) throws SQLException { + + String typeCd = rs.getString("TYPE_CD"); + AttributeType type = AttributeType.getType(typeCd); + String key = rs.getString("KEY_NAME"); + if(type == AttributeType.STRING){ + executionAttributes.putString(key, rs.getString("STRING_VAL")); + } + else if(type == AttributeType.LONG){ + executionAttributes.putLong(key, rs.getLong("LONG_VAL")); + } + else if(type == AttributeType.DOUBLE){ + executionAttributes.putDouble(key, rs.getDouble("DOUBLE_VAL")); + } + else if(type == AttributeType.OBJECT){ + executionAttributes.putLong(key, rs.getLong("OBJECT_VAL")); + } + else{ + throw new BatchCriticalException("Invalid type found: [" + typeCd + "] for execution id: [" + + executionId + "]"); + } + } + }; + + jdbcTemplate.query(getQuery(FIND_STEP_EXECUTION_ATTRS), new Object[]{executionId}, callback); + + return executionAttributes; + } /** * @see StepDao#update(StepInstance) @@ -425,7 +636,7 @@ public class JdbcStepDao implements StepDao, InitializingBean { Assert.notNull(step.getId(), "Step Id cannot be null."); Properties restartProps = null; - StreamContext streamContext = step.getStreamContext(); + ExecutionAttributes streamContext = step.getStreamContext(); if (streamContext != null) { restartProps = streamContext.getProperties(); } @@ -443,11 +654,49 @@ public class JdbcStepDao implements StepDao, InitializingBean { * @param jobExecution @throws IllegalArgumentException */ private void validateStepExecution(StepExecution stepExecution) { - Assert.notNull(stepExecution); Assert.notNull(stepExecution.getStepId(), "StepExecution Step-Id cannot be null."); Assert.notNull(stepExecution.getStartTime(), "StepExecution start time cannot be null."); Assert.notNull(stepExecution.getStatus(), "StepExecution status cannot be null."); } + + public void setLobHandler(LobHandler lobHandler) { + this.lobHandler = lobHandler; + } + + public static class AttributeType { + + private final String type; + + private AttributeType(String type) { + this.type = type; + } + + public String toString() { + return type; + } + + public static final AttributeType STRING = new AttributeType("STRING"); + + public static final AttributeType LONG = new AttributeType("LONG"); + + public static final AttributeType OBJECT = new AttributeType("OBJECT"); + + public static final AttributeType DOUBLE = new AttributeType("DOUBLE"); + + private static final AttributeType[] VALUES = { STRING, OBJECT, LONG, + DOUBLE }; + + public static AttributeType getType(String typeAsString) { + + for (int i = 0; i < VALUES.length; i++) { + if (VALUES[i].toString().equals(typeAsString)) { + return (AttributeType) VALUES[i]; + } + } + + return null; + } + } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java index 323e4b4c2..268166d20 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapStepDao.java @@ -25,7 +25,7 @@ import java.util.Set; import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; public class MapStepDao implements StepDao { @@ -80,8 +80,8 @@ public class MapStepDao implements StepDao { return new ArrayList(steps); } - public StreamContext getStreamContext(Long stepId) { - return (StreamContext) restartsById.get(stepId); + public ExecutionAttributes getStreamContext(Long stepId) { + return (ExecutionAttributes) restartsById.get(stepId); } public int getStepExecutionCount(StepInstance stepInstance) { @@ -119,5 +119,17 @@ public class MapStepDao implements StepDao { // no-op } + public ExecutionAttributes findExecutionAttributes(Long executionId) { + return null; + } + + public void save(Long executionId, + ExecutionAttributes executionAttributes) { + } + + public void update(Long executionId, + ExecutionAttributes executionAttributes) { + } + } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepDao.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepDao.java index 1629d9faf..0bdeebe0b 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepDao.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/StepDao.java @@ -21,6 +21,7 @@ import java.util.List; import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; +import org.springframework.batch.item.ExecutionAttributes; /** * Data access object for steps. @@ -101,4 +102,34 @@ public interface StepDao { * @return list of stepExecutions */ public List findStepExecutions(StepInstance step); + + /** + * Find all {@link ExecutionAttributes} for the given execution id. + * + * @param executionId - Long id of the {@link StepExecution} + * that the attributes belongs to. + * @return attributes for the provided id. If + * none are found, an empty {@link ExecutionAttributes} will be returned. + * @throws IllegalArgumentException if the id is null. + */ + ExecutionAttributes findExecutionAttributes(final Long executionId); + + /** + * Save the provided {@link ExecutionAttributes} for the given + * execution Id. + * + * @param executionId to be saved + * @param executionAttributes to be saved. + * @throws IllegalArgumentException if the executionId or + * attributes are null. + */ + void save(final Long executionId, final ExecutionAttributes executionAttributes); + + /** + * Update the provided execution attributes. + * + * @param executionId + * @param executionAttributes + */ + void update(final Long executionId, ExecutionAttributes executionAttributes); } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java index a06f12a43..2b940f8cd 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java @@ -26,7 +26,7 @@ import java.util.Set; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.BatchCriticalException; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.context.SynchronizedAttributeAccessor; @@ -46,7 +46,7 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements private StreamManager streamManager; - private StreamContext streamContext; + private ExecutionAttributes streamContext; /** * Default constructor. @@ -196,14 +196,14 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements * (non-Javadoc) * @see org.springframework.batch.item.ItemStream#getStreamContext() */ - public StreamContext getStreamContext() { + public ExecutionAttributes getStreamContext() { return streamManager.getStreamContext(this); } /* (non-Javadoc) * @see org.springframework.batch.execution.scope.StepContext#restoreFrom(org.springframework.batch.item.StreamContext) */ - public void restoreFrom(StreamContext streamContext) { + public void restoreFrom(ExecutionAttributes streamContext) { this.streamContext = streamContext; } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java index 9ca89bfa3..2b2044d99 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java @@ -17,7 +17,7 @@ package org.springframework.batch.execution.scope; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.StreamContextProvider; import org.springframework.core.AttributeAccessor; @@ -63,5 +63,5 @@ public interface StepContext extends AttributeAccessor, StreamContextProvider { * * @param streamContext */ - void restoreFrom(StreamContext streamContext); + void restoreFrom(ExecutionAttributes streamContext); } \ No newline at end of file diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index d6cfff9da..b83b55abc 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -34,7 +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.ItemStream; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.ExitStatus; @@ -206,7 +206,7 @@ public class SimpleStepExecutor { // TODO: check that stepExecution can // aggregate these contributions if they // come in asynchronously. - StreamContext statistics = stepContext.getStreamContext(); + ExecutionAttributes statistics = stepContext.getStreamContext(); contribution.setStreamContext(statistics); contribution.incrementCommitCount(); diff --git a/spring-batch-execution/src/main/resources/schema-db2.sql b/spring-batch-execution/src/main/resources/schema-db2.sql index bf2e09371..7c324321a 100644 --- a/spring-batch-execution/src/main/resources/schema-db2.sql +++ b/spring-batch-execution/src/main/resources/schema-db2.sql @@ -3,7 +3,8 @@ DROP TABLE BATCH_STEP_EXECUTION ; DROP TABLE BATCH_JOB_EXECUTION ; DROP TABLE BATCH_STEP_INSTANCE ; DROP TABLE BATCH_JOB_INSTANCE ; -DROP TABLE BATCH_JOB_INSTANCE_PROPERTIES ; +DROP TABLE BATCH_JOB_INSTANCE_PARAMS ; +DROP TABLE BATCH_STEP_EXECUTION_ATTRS ; DROP SEQUENCE BATCH_STEP_EXECUTION_SEQ ; DROP SEQUENCE BATCH_STEP_SEQ ; @@ -35,7 +36,7 @@ CREATE TABLE BATCH_JOB_INSTANCE_PARAMS ( KEY_NAME VARCHAR(100) NOT NULL , STRING_VAL VARCHAR(250) , DATE_VAL TIMESTAMP , - LONG_VAL VARCHAR(10) ); + LONG_VAL BIGINT ); CREATE TABLE BATCH_STEP_INSTANCE ( ID BIGINT PRIMARY KEY , @@ -59,6 +60,16 @@ CREATE TABLE BATCH_STEP_EXECUTION ( CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(2500)); + +CREATE TABLE BATCH_STEP_EXECUTION_ATTRS ( + EXECUTION_ID BIGINT NOT NULL , + TYPE_CD VARCHAR(6) NOT NULL , + KEY_NAME VARCHAR(100) NOT NULL , + STRING_VAL VARCHAR(250) , + DATE_VAL TIMESTAMP , + LONG_VAL VARCHAR(10) , + DOUBLE_VAL DOUBLE PRECISION , + OBJECT_VAL BLOB); CREATE SEQUENCE BATCH_STEP_EXECUTION_SEQ; CREATE SEQUENCE BATCH_STEP_SEQ; diff --git a/spring-batch-execution/src/main/resources/schema-derby.sql b/spring-batch-execution/src/main/resources/schema-derby.sql index 97d10dee2..46aa7359d 100644 --- a/spring-batch-execution/src/main/resources/schema-derby.sql +++ b/spring-batch-execution/src/main/resources/schema-derby.sql @@ -3,7 +3,8 @@ DROP TABLE BATCH_STEP_EXECUTION ; DROP TABLE BATCH_JOB_EXECUTION ; DROP TABLE BATCH_STEP_INSTANCE ; DROP TABLE BATCH_JOB_INSTANCE ; -DROP TABLE BATCH_JOB_INSTANCE_PROPERTIES ; +DROP TABLE BATCH_JOB_INSTANCE_PARAMS ; +DROP TABLE BATCH_STEP_EXECUTION_ATTRS ; DROP TABLE BATCH_STEP_EXECUTION_SEQ ; DROP TABLE BATCH_STEP_SEQ ; @@ -35,7 +36,7 @@ CREATE TABLE BATCH_JOB_INSTANCE_PARAMS ( KEY_NAME VARCHAR(100) NOT NULL , STRING_VAL VARCHAR(250) , DATE_VAL TIMESTAMP , - LONG_VAL VARCHAR(10) ); + LONG_VAL BIGINT ); CREATE TABLE BATCH_STEP_INSTANCE ( ID BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY, @@ -59,6 +60,16 @@ CREATE TABLE BATCH_STEP_EXECUTION ( CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(2500)); + +CREATE TABLE BATCH_STEP_EXECUTION_ATTRS ( + EXECUTION_ID BIGINT NOT NULL , + TYPE_CD VARCHAR(6) NOT NULL , + KEY_NAME VARCHAR(100) NOT NULL , + STRING_VAL VARCHAR(250) , + DATE_VAL TIMESTAMP , + LONG_VAL VARCHAR(10) , + DOUBLE_VAL DOUBLE PRECISION , + OBJECT_VAL BLOB); CREATE TABLE BATCH_STEP_EXECUTION_SEQ (ID BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY, DUMMY VARCHAR(1)); CREATE TABLE BATCH_STEP_SEQ (ID BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY, DUMMY VARCHAR(1)); diff --git a/spring-batch-execution/src/main/resources/schema-hsqldb.sql b/spring-batch-execution/src/main/resources/schema-hsqldb.sql index ef9f17e07..f0a7379cf 100644 --- a/spring-batch-execution/src/main/resources/schema-hsqldb.sql +++ b/spring-batch-execution/src/main/resources/schema-hsqldb.sql @@ -3,7 +3,8 @@ DROP TABLE BATCH_STEP_EXECUTION IF EXISTS; DROP TABLE BATCH_JOB_EXECUTION IF EXISTS; DROP TABLE BATCH_STEP_INSTANCE IF EXISTS; DROP TABLE BATCH_JOB_INSTANCE IF EXISTS; -DROP TABLE BATCH_JOB_INSTANCE_PROPERTIES IF EXISTS; +DROP TABLE BATCH_JOB_INSTANCE_PARAMS IF EXISTS; +DROP TABLE BATCH_STEP_EXECUTION_ATTRS IF EXISTS; DROP TABLE BATCH_STEP_EXECUTION_SEQ IF EXISTS; DROP TABLE BATCH_STEP_SEQ IF EXISTS; @@ -35,7 +36,7 @@ CREATE TABLE BATCH_JOB_INSTANCE_PARAMS ( KEY_NAME VARCHAR(100) NOT NULL , STRING_VAL VARCHAR(250) , DATE_VAL TIMESTAMP , - LONG_VAL VARCHAR(10) ); + LONG_VAL BIGINT ); CREATE TABLE BATCH_STEP_INSTANCE ( ID BIGINT IDENTITY PRIMARY KEY , @@ -59,6 +60,16 @@ CREATE TABLE BATCH_STEP_EXECUTION ( CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(2500)); + +CREATE TABLE BATCH_STEP_EXECUTION_ATTRS ( + EXECUTION_ID BIGINT NOT NULL , + TYPE_CD VARCHAR(6) NOT NULL , + KEY_NAME VARCHAR(100) NOT NULL , + STRING_VAL VARCHAR(250) , + DATE_VAL TIMESTAMP , + LONG_VAL VARCHAR(10) , + DOUBLE_VAL DOUBLE PRECISION , + OBJECT_VAL LONGVARBINARY); CREATE TABLE BATCH_STEP_EXECUTION_SEQ ( ID BIGINT IDENTITY diff --git a/spring-batch-execution/src/main/resources/schema-mysql.sql b/spring-batch-execution/src/main/resources/schema-mysql.sql index 43842336a..239ccd5c5 100644 --- a/spring-batch-execution/src/main/resources/schema-mysql.sql +++ b/spring-batch-execution/src/main/resources/schema-mysql.sql @@ -3,7 +3,8 @@ DROP TABLE IF EXISTS BATCH_STEP_EXECUTION ; DROP TABLE IF EXISTS BATCH_JOB_EXECUTION ; DROP TABLE IF EXISTS BATCH_STEP_INSTANCE ; DROP TABLE IF EXISTS BATCH_JOB_INSTANCE ; -DROP TABLE IF EXISTS BATCH_JOB_INSTANCE_PROPERTIES ; +DROP TABLE IF EXISTS BATCH_JOB_INSTANCE_PARAMS ; +DROP TABLE IF EXISTS BATCH_STEP_EXECUTION_ATTRS ; DROP TABLE IF EXISTS BATCH_STEP_EXECUTION_SEQ ; DROP TABLE IF EXISTS BATCH_STEP_SEQ ; @@ -35,7 +36,7 @@ CREATE TABLE BATCH_JOB_INSTANCE_PARAMS ( KEY_NAME VARCHAR(100) NOT NULL , STRING_VAL VARCHAR(250) , DATE_VAL TIMESTAMP , - LONG_VAL VARCHAR(10) ); + LONG_VAL BIGINT ); CREATE TABLE BATCH_STEP_INSTANCE ( ID BIGINT unsigned PRIMARY KEY , @@ -59,6 +60,16 @@ CREATE TABLE BATCH_STEP_EXECUTION ( CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(2500)); + +CREATE TABLE BATCH_STEP_EXECUTION_ATTRS ( + EXECUTION_ID BIGINT NOT NULL , + TYPE_CD VARCHAR(6) NOT NULL , + KEY_NAME VARCHAR(100) NOT NULL , + STRING_VAL VARCHAR(250) , + DATE_VAL TIMESTAMP , + LONG_VAL VARCHAR(10) , + DOUBLE_VAL DOUBLE PRECISION , + OBJECT_VAL BLOB); CREATE TABLE BATCH_STEP_EXECUTION_SEQ (ID BIGINT NOT NULL) type=MYISAM; INSERT INTO BATCH_STEP_EXECUTION_SEQ values(0); diff --git a/spring-batch-execution/src/main/resources/schema-oracle10g.sql b/spring-batch-execution/src/main/resources/schema-oracle10g.sql index d1b6362c3..c38de7b37 100644 --- a/spring-batch-execution/src/main/resources/schema-oracle10g.sql +++ b/spring-batch-execution/src/main/resources/schema-oracle10g.sql @@ -3,7 +3,8 @@ DROP TABLE BATCH_STEP_EXECUTION ; DROP TABLE BATCH_JOB_EXECUTION ; DROP TABLE BATCH_STEP_INSTANCE ; DROP TABLE BATCH_JOB_INSTANCE ; -DROP TABLE BATCH_JOB_INSTANCE_PROPERTIES ; +DROP TABLE BATCH_JOB_INSTANCE_PARAMS ; +DROP TABLE BATCH_STEP_EXECUTION_ATTRS ; DROP SEQUENCE BATCH_STEP_EXECUTION_SEQ ; DROP SEQUENCE BATCH_STEP_SEQ ; @@ -35,7 +36,7 @@ CREATE TABLE BATCH_JOB_INSTANCE_PARAMS ( KEY_NAME VARCHAR(100) NOT NULL , STRING_VAL VARCHAR(250) , DATE_VAL TIMESTAMP , - LONG_VAL VARCHAR(10) ); + LONG_VAL NUMBER(38) ); CREATE TABLE BATCH_STEP_INSTANCE ( ID NUMBER(38) PRIMARY KEY , @@ -59,6 +60,16 @@ CREATE TABLE BATCH_STEP_EXECUTION ( CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(2500)); + +CREATE TABLE BATCH_STEP_EXECUTION_ATTRS ( + EXECUTION_ID NUMBER(38) NOT NULL , + TYPE_CD VARCHAR(6) NOT NULL , + KEY_NAME VARCHAR(100) NOT NULL , + STRING_VAL VARCHAR(250) , + DATE_VAL TIMESTAMP , + LONG_VAL VARCHAR(10) , + DOUBLE_VAL DOUBLE PRECISION , + OBJECT_VAL BLOB); CREATE SEQUENCE BATCH_STEP_EXECUTION_SEQ; CREATE SEQUENCE BATCH_STEP_SEQ; diff --git a/spring-batch-execution/src/main/resources/schema-postgresql.sql b/spring-batch-execution/src/main/resources/schema-postgresql.sql index bf2e09371..7c324321a 100644 --- a/spring-batch-execution/src/main/resources/schema-postgresql.sql +++ b/spring-batch-execution/src/main/resources/schema-postgresql.sql @@ -3,7 +3,8 @@ DROP TABLE BATCH_STEP_EXECUTION ; DROP TABLE BATCH_JOB_EXECUTION ; DROP TABLE BATCH_STEP_INSTANCE ; DROP TABLE BATCH_JOB_INSTANCE ; -DROP TABLE BATCH_JOB_INSTANCE_PROPERTIES ; +DROP TABLE BATCH_JOB_INSTANCE_PARAMS ; +DROP TABLE BATCH_STEP_EXECUTION_ATTRS ; DROP SEQUENCE BATCH_STEP_EXECUTION_SEQ ; DROP SEQUENCE BATCH_STEP_SEQ ; @@ -35,7 +36,7 @@ CREATE TABLE BATCH_JOB_INSTANCE_PARAMS ( KEY_NAME VARCHAR(100) NOT NULL , STRING_VAL VARCHAR(250) , DATE_VAL TIMESTAMP , - LONG_VAL VARCHAR(10) ); + LONG_VAL BIGINT ); CREATE TABLE BATCH_STEP_INSTANCE ( ID BIGINT PRIMARY KEY , @@ -59,6 +60,16 @@ CREATE TABLE BATCH_STEP_EXECUTION ( CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(2500)); + +CREATE TABLE BATCH_STEP_EXECUTION_ATTRS ( + EXECUTION_ID BIGINT NOT NULL , + TYPE_CD VARCHAR(6) NOT NULL , + KEY_NAME VARCHAR(100) NOT NULL , + STRING_VAL VARCHAR(250) , + DATE_VAL TIMESTAMP , + LONG_VAL VARCHAR(10) , + DOUBLE_VAL DOUBLE PRECISION , + OBJECT_VAL BLOB); CREATE SEQUENCE BATCH_STEP_EXECUTION_SEQ; CREATE SEQUENCE BATCH_STEP_SEQ; diff --git a/spring-batch-execution/src/main/sql/db2.properties b/spring-batch-execution/src/main/sql/db2.properties index 532bd942b..79b46a64c 100644 --- a/spring-batch-execution/src/main/sql/db2.properties +++ b/spring-batch-execution/src/main/sql/db2.properties @@ -2,5 +2,7 @@ platform=db2 # SQL language oddities BIGINT = BIGINT IDENTITY = +DOUBLE = DOUBLE PRECISION +BLOB = BLOB # for generating drop statements... SEQUENCE = SEQUENCE diff --git a/spring-batch-execution/src/main/sql/derby.properties b/spring-batch-execution/src/main/sql/derby.properties index b0620ad81..506d250c8 100644 --- a/spring-batch-execution/src/main/sql/derby.properties +++ b/spring-batch-execution/src/main/sql/derby.properties @@ -3,5 +3,7 @@ platform=db2 BIGINT = BIGINT IDENTITY = GENERATED = GENERATED BY DEFAULT AS IDENTITY +DOUBLE = DOUBLE PRECISION +BLOB = BLOB # for generating drop statements... SEQUENCE = TABLE diff --git a/spring-batch-execution/src/main/sql/destroy.sql.vpp b/spring-batch-execution/src/main/sql/destroy.sql.vpp index 6803b5a4d..2bb836b8e 100644 --- a/spring-batch-execution/src/main/sql/destroy.sql.vpp +++ b/spring-batch-execution/src/main/sql/destroy.sql.vpp @@ -3,7 +3,8 @@ DROP TABLE $!{IFEXISTSBEFORE} BATCH_STEP_EXECUTION $!{IFEXISTS}; DROP TABLE $!{IFEXISTSBEFORE} BATCH_JOB_EXECUTION $!{IFEXISTS}; DROP TABLE $!{IFEXISTSBEFORE} BATCH_STEP_INSTANCE $!{IFEXISTS}; DROP TABLE $!{IFEXISTSBEFORE} BATCH_JOB_INSTANCE $!{IFEXISTS}; -DROP TABLE $!{IFEXISTSBEFORE} BATCH_JOB_INSTANCE_PROPERTIES $!{IFEXISTS}; +DROP TABLE $!{IFEXISTSBEFORE} BATCH_JOB_INSTANCE_PARAMS $!{IFEXISTS}; +DROP TABLE $!{IFEXISTSBEFORE} BATCH_STEP_EXECUTION_ATTRS $!{IFEXISTS}; DROP ${SEQUENCE} $!{IFEXISTSBEFORE} BATCH_STEP_EXECUTION_SEQ $!{IFEXISTS}; DROP ${SEQUENCE} $!{IFEXISTSBEFORE} BATCH_STEP_SEQ $!{IFEXISTS}; diff --git a/spring-batch-execution/src/main/sql/hsqldb.properties b/spring-batch-execution/src/main/sql/hsqldb.properties index d72d28999..5304a4aa8 100644 --- a/spring-batch-execution/src/main/sql/hsqldb.properties +++ b/spring-batch-execution/src/main/sql/hsqldb.properties @@ -3,5 +3,7 @@ platform=hsqldb BIGINT = BIGINT IDENTITY = IDENTITY IFEXISTS = IF EXISTS +DOUBLE = DOUBLE PRECISION +BLOB = LONGVARBINARY # for generating drop statements... SEQUENCE = TABLE diff --git a/spring-batch-execution/src/main/sql/init.sql.vpp b/spring-batch-execution/src/main/sql/init.sql.vpp index ca3d519ec..7361c7618 100644 --- a/spring-batch-execution/src/main/sql/init.sql.vpp +++ b/spring-batch-execution/src/main/sql/init.sql.vpp @@ -23,7 +23,7 @@ CREATE TABLE BATCH_JOB_INSTANCE_PARAMS ( KEY_NAME VARCHAR(100) NOT NULL , STRING_VAL VARCHAR(250) , DATE_VAL TIMESTAMP , - LONG_VAL VARCHAR(10) ); + LONG_VAL ${BIGINT} ); CREATE TABLE BATCH_STEP_INSTANCE ( ID ${BIGINT} $!{IDENTITY} PRIMARY KEY $!{GENERATED}, @@ -47,6 +47,16 @@ CREATE TABLE BATCH_STEP_EXECUTION ( CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(2500)); + +CREATE TABLE BATCH_STEP_EXECUTION_ATTRS ( + EXECUTION_ID ${BIGINT} NOT NULL , + TYPE_CD VARCHAR(6) NOT NULL , + KEY_NAME VARCHAR(100) NOT NULL , + STRING_VAL VARCHAR(250) , + DATE_VAL TIMESTAMP , + LONG_VAL VARCHAR(10) , + DOUBLE_VAL ${DOUBLE} , + OBJECT_VAL ${BLOB}); #sequence( "BATCH_STEP_EXECUTION_SEQ" ) #sequence( "BATCH_STEP_SEQ" ) diff --git a/spring-batch-execution/src/main/sql/mysql.properties b/spring-batch-execution/src/main/sql/mysql.properties index ad5102915..efe4a06fe 100644 --- a/spring-batch-execution/src/main/sql/mysql.properties +++ b/spring-batch-execution/src/main/sql/mysql.properties @@ -4,5 +4,7 @@ BIGINT = BIGINT IDENTITY = unsigned GENERATED = IFEXISTSBEFORE = IF EXISTS +DOUBLE = DOUBLE PRECISION +BLOB = BLOB # for generating drop statements... SEQUENCE = TABLE diff --git a/spring-batch-execution/src/main/sql/oracle10g.properties b/spring-batch-execution/src/main/sql/oracle10g.properties index 7f8e634e6..b31cbc1da 100644 --- a/spring-batch-execution/src/main/sql/oracle10g.properties +++ b/spring-batch-execution/src/main/sql/oracle10g.properties @@ -3,5 +3,7 @@ platform=oracle10g BIGINT = NUMBER(38) IDENTITY = GENERATED = +DOUBLE = DOUBLE PRECISION +BLOB = BLOB # for generating drop statements... SEQUENCE = SEQUENCE diff --git a/spring-batch-execution/src/main/sql/postgresql.properties b/spring-batch-execution/src/main/sql/postgresql.properties index c716d74b2..b8b8435ec 100644 --- a/spring-batch-execution/src/main/sql/postgresql.properties +++ b/spring-batch-execution/src/main/sql/postgresql.properties @@ -3,5 +3,7 @@ platform=postgresql BIGINT = BIGINT IDENTITY = GENERATED = +DOUBLE = DOUBLE PRECISION +BLOB = BLOB # for generating drop statements... SEQUENCE = SEQUENCE diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/bootstrap/support/SimpleExportedJobLauncherTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/bootstrap/support/SimpleExportedJobLauncherTests.java index b613bbb80..b18e419ae 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/bootstrap/support/SimpleExportedJobLauncherTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/bootstrap/support/SimpleExportedJobLauncherTests.java @@ -32,6 +32,7 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep import org.springframework.batch.core.runtime.JobParametersFactory; import org.springframework.batch.execution.configuration.MapJobRegistry; import org.springframework.batch.execution.launch.JobLauncher; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.StreamContext; import org.springframework.batch.support.PropertiesConverter; @@ -53,7 +54,7 @@ public class SimpleExportedJobLauncherTests extends TestCase { public JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException { JobExecution result = new JobExecution(null); StepExecution stepExecution = result.createStepExecution(new StepInstance(null, "step")); - stepExecution.setStreamContext(new StreamContext(PropertiesConverter.stringToProperties("foo=bar"))); + stepExecution.setStreamContext(new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); list.add(jobParameters); return result; } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/MockStepDao.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/MockStepDao.java index fe534129f..94d148787 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/MockStepDao.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/MockStepDao.java @@ -22,6 +22,7 @@ import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; import org.springframework.batch.execution.repository.dao.StepDao; +import org.springframework.batch.item.ExecutionAttributes; public class MockStepDao implements StepDao { @@ -71,4 +72,16 @@ public class MockStepDao implements StepDao { return null; } + public ExecutionAttributes findExecutionAttributes(Long executionId) { + return null; + } + + public void save(Long executionId, + ExecutionAttributes executionAttributes) { + } + + public void update(Long executionId, + ExecutionAttributes executionAttributes) { + } + } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java index 0186d6dbb..b45e3a207 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java @@ -37,6 +37,7 @@ import org.springframework.batch.core.domain.StepSupport; import org.springframework.batch.core.repository.BatchRestartException; import org.springframework.batch.execution.repository.dao.JobDao; import org.springframework.batch.execution.repository.dao.StepDao; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.StreamContext; /* @@ -364,7 +365,10 @@ public class SimpleJobRepositoryTests extends TestCase { public void testUpdateStepExecution(){ StepExecution stepExecution = new StepExecution(new StepInstance(new Long(10L)), null, new Long(1)); stepExecution.setId(new Long(11)); + ExecutionAttributes executionAttributes = new ExecutionAttributes(); + stepExecution.setStreamContext(executionAttributes); stepDao.update(stepExecution); + stepDao.update(stepExecution.getId(), executionAttributes); stepDaoControl.replay(); jobRepository.saveOrUpdate(stepExecution); stepDaoControl.verify(); @@ -372,7 +376,10 @@ public class SimpleJobRepositoryTests extends TestCase { public void testSaveExistingStepExecution(){ StepExecution stepExecution = new StepExecution(new StepInstance(new Long(10L)), null, null); + ExecutionAttributes executionAttributes = new ExecutionAttributes(); + stepExecution.setStreamContext(executionAttributes); stepDao.save(stepExecution); + stepDao.save(stepExecution.getId(), executionAttributes); stepDaoControl.replay(); jobRepository.saveOrUpdate(stepExecution); stepDaoControl.verify(); @@ -408,7 +415,7 @@ public class SimpleJobRepositoryTests extends TestCase { databaseStep1.setStreamContext(null); stepDaoControl.setReturnValue(databaseStep1); stepDao.createStep(databaseJob, "TestStep2"); - databaseStep2.setStreamContext(new StreamContext()); + databaseStep2.setStreamContext(new ExecutionAttributes()); stepDaoControl.setReturnValue(databaseStep2); jobDao.save(new JobExecution(databaseJob)); jobDaoControl.setMatcher(new ArgumentsMatcher(){ @@ -443,7 +450,7 @@ public class SimpleJobRepositoryTests extends TestCase { stepDao.getStepExecutionCount(databaseStep1); stepDaoControl.setReturnValue(1); stepDao.findStep(databaseJob, "TestStep2"); - databaseStep2.setStreamContext(new StreamContext()); + databaseStep2.setStreamContext(new ExecutionAttributes()); stepDaoControl.setReturnValue(databaseStep2); stepDao.getStepExecutionCount(databaseStep2); stepDaoControl.setReturnValue(1); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java index 3ff8fdabc..ef5fcc5de 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java @@ -18,6 +18,7 @@ package org.springframework.batch.execution.repository.dao; import java.util.Date; import java.util.List; +import java.util.Properties; import org.springframework.batch.core.domain.BatchStatus; import org.springframework.batch.core.domain.Job; @@ -28,7 +29,8 @@ 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; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.ExecutionAttributes; +import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.support.PropertiesConverter; import org.springframework.dao.OptimisticLockingFailureException; @@ -60,6 +62,8 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour protected JobExecution jobExecution; protected JobParameters jobParameters = new JobParameters(); + + protected ExecutionAttributes executionAttributes; public void setJobDao(JobDao jobDao) { this.jobDao = jobDao; @@ -92,6 +96,13 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour stepExecution.setStatus(BatchStatus.STARTED); stepExecution.setStartTime(new Date(System.currentTimeMillis())); stepDao.save(stepExecution); + + executionAttributes = new ExecutionAttributes(); + executionAttributes.putString("1", "testString1"); + executionAttributes.putString("2", "testString2"); + executionAttributes.putLong("3", 3); + executionAttributes.putDouble("4", 4.4); + } public void testVersionIsNotNullForStep() throws Exception { @@ -150,7 +161,9 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour public void testUpdateStepWithStreamContext() { step1.setStatus(BatchStatus.COMPLETED); - StreamContext streamContext = new StreamContext(PropertiesConverter.stringToProperties("key1=restartData")); + Properties data = new Properties(); + data.setProperty("restart.key1", "restartData"); + ExecutionAttributes streamContext = new ExecutionAttributes(data); step1.setStreamContext(streamContext); stepDao.update(step1); StepInstance tempStep = stepDao.findStep(jobInstance, step1.getName()); @@ -163,7 +176,7 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour StepExecution execution = new StepExecution(step2, jobExecution, null); execution.setStatus(BatchStatus.STARTED); execution.setStartTime(new Date(System.currentTimeMillis())); - execution.setStreamContext(new StreamContext(PropertiesConverter.stringToProperties("key1=0,key2=5"))); + execution.setStreamContext(new ExecutionAttributes(PropertiesConverter.stringToProperties("key1=0,key2=5"))); execution.setExitStatus(new ExitStatus(false, ExitCodeExceptionClassifier.FATAL_EXCEPTION, "java.lang.Exception")); stepDao.save(execution); @@ -181,7 +194,7 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour stepExecution.setEndTime(new Date(System.currentTimeMillis())); stepExecution.setCommitCount(5); stepExecution.setTaskCount(5); - stepExecution.setStreamContext(new StreamContext()); + stepExecution.setStreamContext(new ExecutionAttributes()); stepExecution.setExitStatus(new ExitStatus(false, ExitCodeExceptionClassifier.FATAL_EXCEPTION, "java.lang.Exception")); stepDao.update(stepExecution); @@ -240,5 +253,16 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour .indexOf("" + stepExecution.getVersion()) >= 0); } } - + + public void testSaveExecutionAttributes(){ + + stepDao.save(stepExecution.getId(), executionAttributes); + ExecutionAttributes attributes = stepDao.findExecutionAttributes(stepExecution.getId()); + assertEquals(executionAttributes, attributes); + executionAttributes.putString("newString", "newString"); + stepDao.update(stepExecution.getId(), executionAttributes); + attributes = stepDao.findExecutionAttributes(stepExecution.getId()); + assertEquals(executionAttributes, attributes); + } + } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepDaoTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepDaoTests.java index 113d6e2c4..125b545e7 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepDaoTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/MapStepDaoTests.java @@ -17,6 +17,7 @@ package org.springframework.batch.execution.repository.dao; import java.util.List; +import java.util.Properties; import junit.framework.TestCase; @@ -25,8 +26,9 @@ import org.springframework.batch.core.domain.JobInstance; import org.springframework.batch.core.domain.JobParameters; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.core.domain.StepInstance; -import org.springframework.batch.item.StreamContext; -import org.springframework.batch.support.PropertiesConverter; +import org.springframework.batch.execution.repository.dao.MapStepDao; +import org.springframework.batch.item.ExecutionAttributes; +import org.springframework.batch.item.stream.GenericStreamContext; public class MapStepDaoTests extends TestCase { @@ -107,7 +109,9 @@ public class MapStepDaoTests extends TestCase { public void testSaveStreamContext() throws Exception { assertEquals(null, dao.getStreamContext(step.getId())); step.setStatus(BatchStatus.COMPLETED); - StreamContext streamContext = new StreamContext(PropertiesConverter.stringToProperties("key1=restartData")); + Properties data = new Properties(); + data.setProperty("restart.key1", "restartData"); + ExecutionAttributes streamContext = new ExecutionAttributes(data); step.setStreamContext(streamContext); dao.update(step); StepInstance tempStep = dao.findStep(job, step.getName()); diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java index 1dbce4663..eb6dd67f4 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java @@ -24,7 +24,7 @@ import junit.framework.TestCase; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.item.ItemStream; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.stream.ItemStreamAdapter; import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.support.PropertiesConverter; @@ -164,18 +164,18 @@ public class SimpleStepContextTests extends TestCase { public void close(Object key) { } - public StreamContext getStreamContext(Object key) { - return new StreamContext(PropertiesConverter.stringToProperties("foo=bar")); + public ExecutionAttributes getStreamContext(Object key) { + return new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar")); } public void open(Object key) { } - public void register(Object key, ItemStream stream, StreamContext streamContext) { + public void register(Object key, ItemStream stream, ExecutionAttributes streamContext) { map.put(key, stream); } - public void restoreFrom(Object key, StreamContext data) { + public void restoreFrom(Object key, ExecutionAttributes data) { } } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java index 1d5db23ea..2b1bd26fc 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java @@ -40,7 +40,7 @@ import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.execution.tasklet.ItemOrientedTasklet; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.ExecutionAttributes; import org.springframework.batch.item.StreamException; import org.springframework.batch.item.reader.ListItemReader; import org.springframework.batch.item.stream.ItemStreamAdapter; @@ -314,7 +314,7 @@ public class SimpleStepExecutorTests extends TestCase { StepExecution stepExecution = new StepExecution(step, jobExecutionContext); stepExecution.getStep().setStreamContext( - new StreamContext(PropertiesConverter.stringToProperties("foo=bar"))); + new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar"))); stepExecutor.execute(stepExecution); @@ -421,9 +421,9 @@ public class SimpleStepExecutorTests extends TestCase { final Map map = new HashMap(); stepExecutor.setStreamManager(new SimpleStreamManager(new ResourcelessTransactionManager()) { - public StreamContext getStreamContext(Object key) { + public ExecutionAttributes getStreamContext(Object key) { // TODO Auto-generated method stub - return new StreamContext(PropertiesConverter.stringToProperties("foo=bar")); + return new ExecutionAttributes(PropertiesConverter.stringToProperties("foo=bar")); } }); @@ -453,12 +453,12 @@ public class SimpleStepExecutorTests extends TestCase { return restoreFromCalledWithSomeContext; } - public StreamContext getStreamContext() { + public ExecutionAttributes getStreamContext() { getStreamContextCalled = true; - return new StreamContext(PropertiesConverter.stringToProperties("spam=bucket")); + return new ExecutionAttributes(PropertiesConverter.stringToProperties("spam=bucket")); } - public void restoreFrom(StreamContext data) { + public void restoreFrom(ExecutionAttributes data) { restoreFromCalled = true; restoreFromCalledWithSomeContext = data.getProperties().size() > 0; } diff --git a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/destroy.sql b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/destroy.sql index 704b88390..68c2ff272 100644 --- a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/destroy.sql +++ b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/destroy.sql @@ -3,7 +3,8 @@ DROP TABLE BATCH_STEP_EXECUTION IF EXISTS; DROP TABLE BATCH_JOB_EXECUTION IF EXISTS; DROP TABLE BATCH_STEP_INSTANCE IF EXISTS; DROP TABLE BATCH_JOB_INSTANCE IF EXISTS; -DROP TABLE BATCH_JOB_INSTANCE_PROPERTIES IF EXISTS; +DROP TABLE BATCH_JOB_INSTANCE_PARAMS IF EXISTS; +DROP TABLE BATCH_STEP_EXECUTION_ATTRS IF EXISTS; DROP TABLE BATCH_STEP_EXECUTION_SEQ IF EXISTS; DROP TABLE BATCH_STEP_SEQ IF EXISTS; diff --git a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/init.sql b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/init.sql index 7a406dbed..fdcf7a9ca 100644 --- a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/init.sql +++ b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/repository/dao/init.sql @@ -23,7 +23,7 @@ CREATE TABLE BATCH_JOB_INSTANCE_PARAMS ( KEY_NAME VARCHAR(100) NOT NULL , STRING_VAL VARCHAR(250) , DATE_VAL TIMESTAMP , - LONG_VAL VARCHAR(10) ); + LONG_VAL BIGINT ); CREATE TABLE BATCH_STEP_INSTANCE ( ID BIGINT IDENTITY PRIMARY KEY , @@ -47,16 +47,26 @@ CREATE TABLE BATCH_STEP_EXECUTION ( CONTINUABLE CHAR(1), EXIT_CODE VARCHAR(20), EXIT_MESSAGE VARCHAR(2500)); + +CREATE TABLE BATCH_STEP_EXECUTION_ATTRS ( + EXECUTION_ID BIGINT NOT NULL , + TYPE_CD VARCHAR(6) NOT NULL , + KEY_NAME VARCHAR(100) NOT NULL , + STRING_VAL VARCHAR(250) , + DATE_VAL TIMESTAMP , + LONG_VAL VARCHAR(10) , + DOUBLE_VAL DOUBLE PRECISION , + OBJECT_VAL LONGVARBINARY); -CREATE TABLE BATCH_STEP_EXECUTION_SEQ ( - ID BIGINT IDENTITY -); -CREATE TABLE BATCH_STEP_SEQ ( - ID BIGINT IDENTITY -); -CREATE TABLE BATCH_JOB_EXECUTION_SEQ ( - ID BIGINT IDENTITY -); -CREATE TABLE BATCH_JOB_SEQ ( - ID BIGINT IDENTITY -); +CREATE TABLE BATCH_STEP_EXECUTION_SEQ ( + ID BIGINT IDENTITY +); +CREATE TABLE BATCH_STEP_SEQ ( + ID BIGINT IDENTITY +); +CREATE TABLE BATCH_JOB_EXECUTION_SEQ ( + ID BIGINT IDENTITY +); +CREATE TABLE BATCH_JOB_SEQ ( + ID BIGINT IDENTITY +);