SpringCloud/spring-cloud-task
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

SCT-96 Update TaskRepository to be additive only

Browse Source 
resolves spring-cloud/spring-cloud-task#96
This commit is contained in:
Glenn Renfro
2016-03-01 13:06:42 -05:00
committed by Michael Minella
parent 83836e1eea
commit dc26f7c98d
18 changed files with 244 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-cloud-task-core/src/main/java/org/springframework/cloud/task/listener/TaskLifecycleListener.java
View File

@@ -153,7 +153,8 @@ public class TaskLifecycleListener implements ApplicationListener<ApplicationEve
this.applicationFailedEvent.getException()).getExitMessage());
}
taskExecution.setExitMessage(invokeOnTaskEnd(taskExecution).getExitMessage());
taskRepository.update(taskExecution);
taskRepository.completeTaskExecution(taskExecution.getExecutionId(), taskExecution.getExitCode(),
taskExecution.getEndTime(), taskExecution.getExitMessage());
}
else {
logger.error("An event to end a task has been received for a task that has " +
@@ -170,11 +171,8 @@ public class TaskLifecycleListener implements ApplicationListener<ApplicationEve
args = Arrays.asList(this.applicationArguments.getSourceArgs());
}
this.taskExecution = new TaskExecution(this.taskRepository.getNextExecutionId(),
null, this.taskNameResolver.getTaskName(), new Date(), null, null,
args);
this.taskRepository.createTaskExecution(this.taskExecution);
this.taskExecution = this.taskRepository.createTaskExecution(
this.taskNameResolver.getTaskName(), new Date(), args);
}
else {
logger.error("Multiple start events have been received. The first one was " +

2
spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/TaskExecution.java
View File

@@ -88,7 +88,7 @@ public class TaskExecution {
}
public Integer getExitCode() {
return exitCode;
return (exitCode == null) ? 0 : exitCode;
}
public void setExitCode(Integer exitCode) {

28
spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/TaskRepository.java
View File

@@ -16,6 +16,9 @@
package org.springframework.cloud.task.repository;
import java.util.Date;
import java.util.List;
import org.springframework.transaction.annotation.Transactional;
/**
@@ -27,23 +30,28 @@ import org.springframework.transaction.annotation.Transactional;
public interface TaskRepository {
/**
* Notifies the repository that a taskExecution needs to be updated.
* Notifies the repository that a taskExecution has completed.
*
* @param taskExecution taskExecution to be updated
* @param executionId to the task execution to be updated.
* @param exitCode to be stored for this task.
* @param endTime designated when the task completed.
* @param exitMessage to be stored for the task.
* @return the updated {@link TaskExecution}
*/
public void update(TaskExecution taskExecution);
@Transactional
TaskExecution completeTaskExecution(long executionId, Integer exitCode, Date endTime,
String exitMessage);
/**
* Notifies the repository that a taskExecution needs to be created.
*
* @param taskExecution taskExecution to be recorded
* @param taskName the name that associated with the task execution.
* @param startTime the time task began.
* @param parameters list of key/value pairs that configure the task.
* @return the initial {@link TaskExecution}
*/
@Transactional
public void createTaskExecution(TaskExecution taskExecution);
TaskExecution createTaskExecution(String taskName,
Date startTime,List<String> parameters);
/**
* Retrieves the next available execution id for a task execution.
* @return long containing the executionId.
*/
public long getNextExecutionId();
}

44
spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/dao/JdbcTaskExecutionDao.java
View File

@@ -63,9 +63,8 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
public static final String TASK_NAME_WHERE_CLAUSE = "where TASK_NAME = ? ";
private static final String SAVE_TASK_EXECUTION = "INSERT into %PREFIX%EXECUTION"
+ "(TASK_EXECUTION_ID, START_TIME, END_TIME, "
+ "TASK_NAME, EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED)"
+ "values (?, ?, ?, ?, ?, ?, ?)";
+ "(TASK_EXECUTION_ID, START_TIME, TASK_NAME, LAST_UPDATED)"
+ "values (?, ?, ?, ?)";
private static final String CREATE_TASK_PARAMETER = "INSERT into "
+ "%PREFIX%EXECUTION_PARAMS(TASK_EXECUTION_ID, TASK_PARAM ) values (?, ?)";
@@ -74,8 +73,7 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
+ "%PREFIX%EXECUTION WHERE TASK_EXECUTION_ID = ?";
private static final String UPDATE_TASK_EXECUTION = "UPDATE %PREFIX%EXECUTION set "
+ "START_TIME = ?, END_TIME = ?, TASK_NAME = ?, EXIT_CODE = ?, "
+ "EXIT_MESSAGE = ?, LAST_UPDATED = ? "
+ "END_TIME = ?, EXIT_CODE = ?, EXIT_MESSAGE = ?, LAST_UPDATED = ? "
+ "where TASK_EXECUTION_ID = ?";
private static final String GET_EXECUTION_BY_ID = "SELECT TASK_EXECUTION_ID, " +
@@ -119,36 +117,38 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
}
@Override
public void saveTaskExecution(TaskExecution taskExecution) {
Object[] parameters = new Object[]{ taskExecution.getExecutionId(),
taskExecution.getStartTime(), taskExecution.getEndTime(),
taskExecution.getTaskName(), taskExecution.getExitCode(),
taskExecution.getExitMessage(), new Date()};
public TaskExecution createTaskExecution(String taskName,
Date startTime, List<String> parameters) {
long taskExecutionId = getNextExecutionId();
TaskExecution taskExecution = new TaskExecution(taskExecutionId, null, taskName,
startTime, null, null, parameters);
Object[] queryParameters = new Object[]{ taskExecutionId, startTime, taskName, new Date()};
jdbcTemplate.update(
getQuery(SAVE_TASK_EXECUTION),
parameters,
new int[]{ Types.BIGINT, Types.TIMESTAMP, Types.TIMESTAMP,
Types.VARCHAR, Types.INTEGER, Types.VARCHAR, Types.TIMESTAMP });
insertTaskParameters(taskExecution.getExecutionId(), taskExecution.getParameters());
queryParameters,
new int[]{ Types.BIGINT, Types.TIMESTAMP, Types.VARCHAR, Types.TIMESTAMP });
insertTaskParameters(taskExecutionId, parameters);
return taskExecution;
}
@Override
public void updateTaskExecution(TaskExecution taskExecution) {
public void completeTaskExecution(long taskExecutionId, Integer exitCode, Date endTime,
String exitMessage) {
// Check if given TaskExecution's Id already exists, if none is found
// it is invalid and an exception should be thrown.
if (jdbcTemplate.queryForObject(getQuery(CHECK_TASK_EXECUTION_EXISTS), Integer.class,
new Object[]{ taskExecution.getExecutionId() }) != 1) {
throw new IllegalStateException("Invalid TaskExecution, ID " + taskExecution.getExecutionId() + " not found.");
new Object[]{ taskExecutionId}) != 1) {
throw new IllegalStateException("Invalid TaskExecution, ID " + taskExecutionId + " not found.");
}
Object[] parameters = new Object[]{ taskExecution.getStartTime(), taskExecution.getEndTime(),
taskExecution.getTaskName(), taskExecution.getExitCode(),
taskExecution.getExitMessage(), new Date(), taskExecution.getExecutionId()};
Object[] parameters = new Object[]{ endTime, exitCode, exitMessage, new Date(),
taskExecutionId};
jdbcTemplate.update(
getQuery(UPDATE_TASK_EXECUTION),
parameters,
new int[]{ Types.TIMESTAMP, Types.TIMESTAMP, Types.VARCHAR, Types.INTEGER,
Types.VARCHAR, Types.TIMESTAMP, Types.BIGINT});
new int[]{ Types.TIMESTAMP, Types.INTEGER, Types.VARCHAR, Types.TIMESTAMP,
Types.BIGINT});
}
/**

17
spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/dao/MapTaskExecutionDao.java
View File

@@ -19,6 +19,7 @@ package org.springframework.cloud.task.repository.dao;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -48,13 +49,21 @@ public class MapTaskExecutionDao implements TaskExecutionDao {
}
@Override
public void saveTaskExecution(TaskExecution taskExecution) {
taskExecutions.put(taskExecution.getExecutionId(), taskExecution);
public TaskExecution createTaskExecution(String taskName,
Date startTime, List<String> parameters) {
long taskExecutionId = getNextExecutionId();
TaskExecution taskExecution = new TaskExecution(taskExecutionId, null, taskName,
startTime, null, null, parameters);
taskExecutions.put(taskExecutionId, taskExecution);
return taskExecution;
}
@Override
public void updateTaskExecution(TaskExecution taskExecution) {
taskExecutions.put(taskExecution.getExecutionId(), taskExecution);
public void completeTaskExecution(long executionId, Integer exitCode, Date endTime, String exitMessage) {
TaskExecution taskExecution= taskExecutions.get(executionId);
taskExecution.setEndTime(endTime);
taskExecution.setExitCode(exitCode);
taskExecution.setExitMessage(exitMessage);
}
@Override

21
spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/dao/TaskExecutionDao.java
View File

@@ -16,6 +16,7 @@
package org.springframework.cloud.task.repository.dao;
import java.util.Date;
import java.util.List;
import org.springframework.cloud.task.repository.TaskExecution;
@@ -32,16 +33,23 @@ public interface TaskExecutionDao {
/**
* Save a new {@link TaskExecution}.
*
* @param taskExecution the taskExecution to be stored.
* @param taskName the name that associated with the task execution.
* @param startTime the time task began.
* @param parameters list of key/value pairs that configure the task.
* @return A fully qualified {@linkTaskExecution} instance.
*/
void saveTaskExecution(TaskExecution taskExecution);
TaskExecution createTaskExecution( String taskName,
Date startTime, List<String> parameters);
/**
* Update and existing {@link TaskExecution}.
*
* @param taskExecution the taskExecution to be updated.
* @param executionId the id of the taskExecution to be updated.
* @param exitCode the status of the task upon completion.
* @param endTime the time the task completed.
* @param exitMessage the message assigned to the task upon completion.
*/
void updateTaskExecution(TaskExecution taskExecution);
void completeTaskExecution(long executionId, Integer exitCode, Date endTime, String exitMessage);
/**
* Retrieves a task execution from the task repository.
@@ -107,9 +115,4 @@ public interface TaskExecutionDao {
public Page<TaskExecution> findAll(Pageable pageable);
/**
* Retrieves the next available execution id for a task execution.
* @return long containing the executionId.
*/
public long getNextExecutionId();
}

66
spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/support/SimpleTaskRepository.java
View File

@@ -16,6 +16,9 @@
package org.springframework.cloud.task.repository.support;
import java.util.Date;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,26 +52,32 @@ public class SimpleTaskRepository implements TaskRepository {
}
@Override
public void update(TaskExecution taskExecution) {
public TaskExecution completeTaskExecution(long executionId, Integer exitCode, Date endTime,
String exitMessage) {
initialize();
validateTaskExecution(taskExecution);
taskExecutionDao.updateTaskExecution(taskExecution);
logger.info("Updating: " + taskExecution.toString());
validateExitInformation(executionId, exitCode, endTime);
exitMessage = trimExitMessage(exitMessage);
taskExecutionDao.completeTaskExecution(executionId, exitCode, endTime, exitMessage);
logger.debug("Updating: TaskExecution with executionId="+executionId
+ " with the following {"
+ "exitCode=" + exitCode
+ ", endTime=" + endTime
+ ", exitMessage='" + exitMessage + '\''
+ '}');
return taskExecutionDao.getTaskExecution(executionId);
}
@Override
public void createTaskExecution(TaskExecution taskExecution) {
public TaskExecution createTaskExecution(String taskName,
Date startTime,List<String> parameters) {
initialize();
validateTaskExecution(taskExecution);
taskExecutionDao.saveTaskExecution(taskExecution);
validateCreateInformation(startTime, taskName);
TaskExecution taskExecution =
taskExecutionDao.createTaskExecution(taskName, startTime, parameters);
logger.info("Creating: " + taskExecution.toString());
}
@Override
public long getNextExecutionId() {
initialize();
return taskExecutionDao.getNextExecutionId();
return taskExecution;
}
/**
@@ -93,23 +102,30 @@ public class SimpleTaskRepository implements TaskRepository {
}
/**
* Validate TaskExecution. At a minimum a startTime.
*
* @param taskExecution the taskExecution to be evaluagted.
* Validate startTime and taskName are valid.
*/
private void validateTaskExecution(TaskExecution taskExecution) {
Assert.notNull(taskExecution, "taskExecution should not be null");
Assert.notNull(taskExecution.getStartTime(), "TaskExecution start time cannot be null.");
private void validateCreateInformation(Date startTime, String taskName) {
Assert.notNull(startTime, "TaskExecution start time cannot be null.");
if (taskExecution.getTaskName() != null &&
taskExecution.getTaskName().length() > MAX_TASK_NAME_SIZE) {
if (taskName != null &&
taskName.length() > MAX_TASK_NAME_SIZE) {
throw new IllegalArgumentException("TaskName length exceeds "
+ MAX_TASK_NAME_SIZE + " characters");
}
//Trim the exit message
if(taskExecution.getExitMessage() != null &&
taskExecution.getExitMessage().length() > MAX_EXIT_MESSAGE_SIZE){
taskExecution.setExitMessage(taskExecution.getExitMessage().substring(0, MAX_EXIT_MESSAGE_SIZE - 1));
}
private void validateExitInformation(long executionId, Integer exitCode, Date endTime){
Assert.notNull(exitCode, "exitCode should not be null");
Assert.isTrue(exitCode >= 0, "exit code must be greater than or equal to zero");
Assert.notNull(endTime, "TaskExecution endTime cannot be null.");
}
private String trimExitMessage(String exitMessage){
String result = exitMessage;
if(exitMessage != null &&
exitMessage.length() > MAX_EXIT_MESSAGE_SIZE) {
result = exitMessage.substring(0, MAX_EXIT_MESSAGE_SIZE - 1);
}
return result;
}
}