gh-424 Add the ability to retrieve the last TaskExecution for Tasks
- Add the ability to retrieve the last TaskExecution to: - MapTaskExecutionDao - JdbcTaskExecutionDao - Refactor `JdbcTaskExecutionDao` and use `NamedParameterJdbcTemplate` for all persistence store calls - Add the following TaskExecution methods to TaskExplorer: - getLatestTaskExecutionsByTaskNames - getLatestTaskExecutionForTaskName - Add Tests - Ensure that code is JDK7 compatible due to backporting needs - Ensure commit backports to 1.2.x Polishing on tests during merge
This commit is contained in:
Gunnar Hillert
committed by
Glenn Renfro
Glenn Renfro
parent
4c5ce965fd
commit
de1ebfb351
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2015 the original author or authors.
|
||||
* Copyright 2015-2018 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.
|
||||
@@ -27,6 +27,7 @@ import org.springframework.data.domain.Pageable;
|
||||
*
|
||||
* @author Glenn Renfro
|
||||
* @author Michael Minella
|
||||
* @author Gunnar Hillert
|
||||
*/
|
||||
public interface TaskExplorer {
|
||||
|
||||
@@ -105,4 +106,34 @@ public interface TaskExplorer {
|
||||
* @return a <code>Set</code> of the ids of the job executions executed within the task.
|
||||
*/
|
||||
Set<Long> getJobExecutionIdsByTaskExecutionId(long taskExecutionId);
|
||||
|
||||
/**
|
||||
* Returns a {@link List} of the latest {@link TaskExecution} for 1 or more task names.
|
||||
*
|
||||
* Latest is defined by the most recent start time. A {@link TaskExecution} does not have to be finished
|
||||
* (The results may including pending {@link TaskExecution}s).
|
||||
*
|
||||
* It is theoretically possible that a {@link TaskExecution} with the same name to have more than 1
|
||||
* {@link TaskExecution} for the exact same start time. In that case the {@link TaskExecution} with the
|
||||
* highest Task Execution ID is returned.
|
||||
*
|
||||
* This method will not consider end times in its calculations. Thus, when a task execution {@code A} starts
|
||||
* after task execution {@code B} but finishes BEFORE task execution {@code A}, then task execution {@code B}
|
||||
* is being returned.
|
||||
*
|
||||
* @param taskNames At least 1 task name must be provided
|
||||
* @return List of TaskExecutions. May be empty but never null.
|
||||
*/
|
||||
List<TaskExecution> getLatestTaskExecutionsByTaskNames(String... taskNames);
|
||||
|
||||
|
||||
/**
|
||||
* Returns the latest task execution for a given task name. Will ultimately apply the same algorithm underneath
|
||||
* as {@link #getLatestTaskExecutionsByTaskNames(String...)} but will only return a single result.
|
||||
*
|
||||
* @param taskName Must not be null or empty
|
||||
* @return The latest Task Execution or null
|
||||
* @see #getLatestTaskExecutionsByTaskNames(String...)
|
||||
*/
|
||||
TaskExecution getLatestTaskExecutionForTaskName(String taskName);
|
||||
}
|
||||
|
||||
@@ -24,6 +24,7 @@ import java.util.Collections;
|
||||
import java.util.Date;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
|
||||
@@ -40,11 +41,11 @@ import org.springframework.data.domain.Page;
|
||||
import org.springframework.data.domain.PageImpl;
|
||||
import org.springframework.data.domain.Pageable;
|
||||
import org.springframework.data.domain.Sort;
|
||||
import org.springframework.jdbc.core.JdbcOperations;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.core.ResultSetExtractor;
|
||||
import org.springframework.jdbc.core.RowCallbackHandler;
|
||||
import org.springframework.jdbc.core.RowMapper;
|
||||
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
|
||||
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
|
||||
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
@@ -67,61 +68,75 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
public static final String FROM_CLAUSE = "%PREFIX%EXECUTION";
|
||||
|
||||
public static final String RUNNING_TASK_WHERE_CLAUSE =
|
||||
"where TASK_NAME = ? AND END_TIME IS NULL ";
|
||||
"where TASK_NAME = :taskName AND END_TIME IS NULL ";
|
||||
|
||||
public static final String TASK_NAME_WHERE_CLAUSE = "where TASK_NAME = ? ";
|
||||
public static final String TASK_NAME_WHERE_CLAUSE = "where TASK_NAME = :taskName ";
|
||||
|
||||
private static final String SAVE_TASK_EXECUTION = "INSERT into %PREFIX%EXECUTION"
|
||||
+ "(TASK_EXECUTION_ID, START_TIME, TASK_NAME, LAST_UPDATED, EXTERNAL_EXECUTION_ID, PARENT_EXECUTION_ID)"
|
||||
+ "values (?, ?, ?, ?, ?, ?)";
|
||||
+ "values (:taskExecutionId, :startTime, :taskName, :lastUpdated, :externalExecutionId, :parentExecutionId)";
|
||||
|
||||
private static final String CREATE_TASK_ARGUMENT = "INSERT into "
|
||||
+ "%PREFIX%EXECUTION_PARAMS(TASK_EXECUTION_ID, TASK_PARAM ) values (?, ?)";
|
||||
+ "%PREFIX%EXECUTION_PARAMS(TASK_EXECUTION_ID, TASK_PARAM ) values (:taskExecutionId, :taskParam)";
|
||||
|
||||
private static final String START_TASK_EXECUTION_PREFIX = "UPDATE %PREFIX%EXECUTION set "
|
||||
+ "START_TIME = ?, TASK_NAME = ?, LAST_UPDATED = ?";
|
||||
+ "START_TIME = :startTime, TASK_NAME = :taskName, LAST_UPDATED = :lastUpdated";
|
||||
|
||||
private static final String START_TASK_EXECUTION_EXTERNAL_ID_SUFFIX = ", "
|
||||
+ "EXTERNAL_EXECUTION_ID = ?, PARENT_EXECUTION_ID = ? where TASK_EXECUTION_ID = ?";
|
||||
+ "EXTERNAL_EXECUTION_ID = :externalExecutionId, PARENT_EXECUTION_ID = :parentExecutionId where TASK_EXECUTION_ID = :taskExecutionId";
|
||||
|
||||
private static final String START_TASK_EXECUTION_SUFFIX = ", PARENT_EXECUTION_ID = ? where TASK_EXECUTION_ID = ?";
|
||||
private static final String START_TASK_EXECUTION_SUFFIX = ", PARENT_EXECUTION_ID = :parentExecutionId where TASK_EXECUTION_ID = :taskExecutionId";
|
||||
|
||||
private static final String CHECK_TASK_EXECUTION_EXISTS = "SELECT COUNT(*) FROM "
|
||||
+ "%PREFIX%EXECUTION WHERE TASK_EXECUTION_ID = ?";
|
||||
+ "%PREFIX%EXECUTION WHERE TASK_EXECUTION_ID = :taskExecutionId";
|
||||
|
||||
private static final String UPDATE_TASK_EXECUTION = "UPDATE %PREFIX%EXECUTION set "
|
||||
+ "END_TIME = ?, EXIT_CODE = ?, EXIT_MESSAGE = ?, ERROR_MESSAGE = ?, "
|
||||
+ "LAST_UPDATED = ? where TASK_EXECUTION_ID = ?";
|
||||
+ "END_TIME = :endTime, EXIT_CODE = :exitCode, EXIT_MESSAGE = :exitMessage, ERROR_MESSAGE = :errorMessage, "
|
||||
+ "LAST_UPDATED = :lastUpdated where TASK_EXECUTION_ID = :taskExecutionId";
|
||||
|
||||
private static final String UPDATE_TASK_EXECUTION_EXTERNAL_EXECUTION_ID = "UPDATE %PREFIX%EXECUTION set "
|
||||
+ "EXTERNAL_EXECUTION_ID = ? where TASK_EXECUTION_ID = ?";
|
||||
+ "EXTERNAL_EXECUTION_ID = :externalExecutionId where TASK_EXECUTION_ID = :taskExecutionId";
|
||||
|
||||
private static final String GET_EXECUTION_BY_ID = "SELECT TASK_EXECUTION_ID, " +
|
||||
"START_TIME, END_TIME, TASK_NAME, EXIT_CODE, "
|
||||
+ "EXIT_MESSAGE, ERROR_MESSAGE, LAST_UPDATED, EXTERNAL_EXECUTION_ID, "
|
||||
+ "PARENT_EXECUTION_ID "
|
||||
+ "from %PREFIX%EXECUTION where TASK_EXECUTION_ID = ?";
|
||||
+ "from %PREFIX%EXECUTION where TASK_EXECUTION_ID = :taskExecutionId";
|
||||
|
||||
private static final String FIND_ARGUMENT_FROM_ID = "SELECT TASK_EXECUTION_ID, "
|
||||
+ "TASK_PARAM from %PREFIX%EXECUTION_PARAMS where TASK_EXECUTION_ID = ?";
|
||||
+ "TASK_PARAM from %PREFIX%EXECUTION_PARAMS where TASK_EXECUTION_ID = :taskExecutionId";
|
||||
|
||||
private static final String TASK_EXECUTION_COUNT = "SELECT COUNT(*) FROM " +
|
||||
"%PREFIX%EXECUTION ";
|
||||
|
||||
private static final String TASK_EXECUTION_COUNT_BY_NAME = "SELECT COUNT(*) FROM " +
|
||||
"%PREFIX%EXECUTION where TASK_NAME = ?";
|
||||
"%PREFIX%EXECUTION where TASK_NAME = :taskName";
|
||||
|
||||
private static final String RUNNING_TASK_EXECUTION_COUNT_BY_NAME = "SELECT COUNT(*) FROM " +
|
||||
"%PREFIX%EXECUTION where TASK_NAME = ? AND END_TIME IS NULL ";
|
||||
"%PREFIX%EXECUTION where TASK_NAME = :taskName AND END_TIME IS NULL ";
|
||||
|
||||
private static final String LAST_TASK_EXECUTIONS_BY_TASK_NAMES =
|
||||
"select TE2.* from (" +
|
||||
"select MAX(TE.TASK_EXECUTION_ID) as TASK_EXECUTION_ID, TE.TASK_NAME, TE.START_TIME from (" +
|
||||
"select TASK_NAME, MAX(START_TIME) as START_TIME" +
|
||||
" FROM %PREFIX%EXECUTION where TASK_NAME in (:taskNames)" +
|
||||
" GROUP BY TASK_NAME" +
|
||||
") TE_MAX " +
|
||||
"inner join %PREFIX%EXECUTION TE ON TE.TASK_NAME = TE_MAX.TASK_NAME AND TE.START_TIME = TE_MAX.START_TIME " +
|
||||
"group by TE.TASK_NAME, TE.START_TIME" +
|
||||
") TE1 " +
|
||||
"inner join %PREFIX%EXECUTION TE2 ON TE1.TASK_EXECUTION_ID = TE2.TASK_EXECUTION_ID " +
|
||||
"order by TE2.START_TIME DESC, TE2.TASK_EXECUTION_ID DESC";
|
||||
|
||||
private static final String FIND_TASK_NAMES = "SELECT distinct TASK_NAME from %PREFIX%EXECUTION order by TASK_NAME";
|
||||
|
||||
private static final String FIND_TASK_EXECUTION_BY_JOB_EXECUTION_ID = "SELECT TASK_EXECUTION_ID FROM %PREFIX%TASK_BATCH WHERE JOB_EXECUTION_ID = ?";
|
||||
private static final String FIND_JOB_EXECUTION_BY_TASK_EXECUTION_ID = "SELECT JOB_EXECUTION_ID FROM %PREFIX%TASK_BATCH WHERE TASK_EXECUTION_ID = ?";
|
||||
private static final String FIND_TASK_EXECUTION_BY_JOB_EXECUTION_ID = "SELECT TASK_EXECUTION_ID FROM %PREFIX%TASK_BATCH WHERE JOB_EXECUTION_ID = :jobExecutionId";
|
||||
|
||||
private static final String FIND_JOB_EXECUTION_BY_TASK_EXECUTION_ID = "SELECT JOB_EXECUTION_ID FROM %PREFIX%TASK_BATCH WHERE TASK_EXECUTION_ID = :taskExecutionId";
|
||||
|
||||
private String tablePrefix = TaskProperties.DEFAULT_TABLE_PREFIX;
|
||||
|
||||
private JdbcOperations jdbcTemplate;
|
||||
private final NamedParameterJdbcTemplate jdbcTemplate;
|
||||
|
||||
private DataSource dataSource;
|
||||
|
||||
@@ -146,8 +161,8 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
* @param dataSource used by the dao to execute queries and update the tables.
|
||||
*/
|
||||
public JdbcTaskExecutionDao(DataSource dataSource) {
|
||||
Assert.notNull(dataSource);
|
||||
this.jdbcTemplate = new JdbcTemplate(dataSource);
|
||||
Assert.notNull(dataSource, "The dataSource must not be null.");
|
||||
this.jdbcTemplate = new NamedParameterJdbcTemplate(dataSource);
|
||||
this.dataSource = dataSource;
|
||||
orderMap = new LinkedHashMap<>();
|
||||
orderMap.put("START_TIME", Order.DESCENDING);
|
||||
@@ -170,14 +185,17 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
TaskExecution taskExecution = new TaskExecution(nextExecutionId, null, taskName,
|
||||
startTime, null, null, arguments, null, externalExecutionId);
|
||||
|
||||
Object[] queryParameters = new Object[]{ nextExecutionId, startTime,
|
||||
taskName, new Date(), externalExecutionId,
|
||||
parentExecutionId};
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("taskExecutionId", nextExecutionId, Types.BIGINT)
|
||||
.addValue("startTime", startTime, Types.TIMESTAMP)
|
||||
.addValue("taskName", taskName, Types.VARCHAR)
|
||||
.addValue("lastUpdated", new Date(), Types.TIMESTAMP)
|
||||
.addValue("externalExecutionId", externalExecutionId, Types.VARCHAR)
|
||||
.addValue("parentExecutionId", parentExecutionId, Types.BIGINT);
|
||||
|
||||
jdbcTemplate.update(
|
||||
getQuery(SAVE_TASK_EXECUTION),
|
||||
queryParameters,
|
||||
new int[]{ Types.BIGINT, Types.TIMESTAMP, Types.VARCHAR,
|
||||
Types.TIMESTAMP, Types.VARCHAR, Types.BIGINT});
|
||||
queryParameters);
|
||||
insertTaskArguments(nextExecutionId, arguments);
|
||||
return taskExecution;
|
||||
}
|
||||
@@ -196,25 +214,25 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
String externalExecutionId, Long parentExecutionId) {
|
||||
TaskExecution taskExecution = new TaskExecution(executionId, null, taskName,
|
||||
startTime, null, null, arguments,null, externalExecutionId, parentExecutionId);
|
||||
Object[] queryParameters;
|
||||
int[] argTypes;
|
||||
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("startTime", startTime, Types.TIMESTAMP)
|
||||
.addValue("taskName", taskName, Types.VARCHAR)
|
||||
.addValue("lastUpdated", new Date(), Types.TIMESTAMP)
|
||||
.addValue("parentExecutionId", parentExecutionId, Types.BIGINT)
|
||||
.addValue("taskExecutionId", executionId, Types.BIGINT);
|
||||
|
||||
String updateString = START_TASK_EXECUTION_PREFIX;
|
||||
|
||||
if(externalExecutionId == null) {
|
||||
queryParameters = new Object[]{startTime, taskName,
|
||||
new Date(), parentExecutionId, executionId};
|
||||
updateString += START_TASK_EXECUTION_SUFFIX;
|
||||
argTypes = new int[]{Types.TIMESTAMP, Types.VARCHAR,
|
||||
Types.TIMESTAMP, Types.BIGINT, Types.BIGINT};
|
||||
}
|
||||
else {
|
||||
queryParameters = new Object[]{ startTime, taskName,
|
||||
new Date(), externalExecutionId, parentExecutionId, executionId};
|
||||
argTypes = new int[]{ Types.TIMESTAMP, Types.VARCHAR,
|
||||
Types.TIMESTAMP, Types.VARCHAR, Types.BIGINT, Types.BIGINT };
|
||||
updateString += START_TASK_EXECUTION_EXTERNAL_ID_SUFFIX;
|
||||
queryParameters.addValue("externalExecutionId", externalExecutionId, Types.VARCHAR);
|
||||
}
|
||||
|
||||
jdbcTemplate.update(getQuery(updateString), queryParameters, argTypes);
|
||||
jdbcTemplate.update(getQuery(updateString), queryParameters);
|
||||
insertTaskArguments(executionId, arguments);
|
||||
return taskExecution;
|
||||
}
|
||||
@@ -222,20 +240,26 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
@Override
|
||||
public void completeTaskExecution(long taskExecutionId, Integer exitCode, Date endTime,
|
||||
String exitMessage, String errorMessage) {
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("taskExecutionId", taskExecutionId, Types.BIGINT);
|
||||
|
||||
// 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,
|
||||
taskExecutionId) != 1) {
|
||||
if (jdbcTemplate.queryForObject(getQuery(CHECK_TASK_EXECUTION_EXISTS), queryParameters, Integer.class) != 1) {
|
||||
throw new IllegalStateException("Invalid TaskExecution, ID " + taskExecutionId + " not found.");
|
||||
}
|
||||
|
||||
Object[] parameters = new Object[]{ endTime, exitCode, exitMessage, errorMessage, new Date(),
|
||||
taskExecutionId};
|
||||
final MapSqlParameterSource parameters = new MapSqlParameterSource()
|
||||
.addValue("endTime", endTime, Types.TIMESTAMP)
|
||||
.addValue("exitCode", exitCode, Types.INTEGER)
|
||||
.addValue("exitMessage", exitMessage, Types.VARCHAR)
|
||||
.addValue("errorMessage", errorMessage, Types.VARCHAR)
|
||||
.addValue("lastUpdated", new Date(), Types.TIMESTAMP)
|
||||
.addValue("taskExecutionId", taskExecutionId, Types.BIGINT);
|
||||
|
||||
jdbcTemplate.update(
|
||||
getQuery(UPDATE_TASK_EXECUTION),
|
||||
parameters,
|
||||
new int[]{ Types.TIMESTAMP, Types.INTEGER, Types.VARCHAR, Types.VARCHAR, Types.TIMESTAMP,
|
||||
Types.BIGINT});
|
||||
parameters);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -246,9 +270,12 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
|
||||
@Override
|
||||
public TaskExecution getTaskExecution(long executionId) {
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("taskExecutionId", executionId, Types.BIGINT);
|
||||
|
||||
try {
|
||||
TaskExecution taskExecution = jdbcTemplate.queryForObject(getQuery(GET_EXECUTION_BY_ID),
|
||||
new TaskExecutionRowMapper(), executionId);
|
||||
queryParameters, new TaskExecutionRowMapper());
|
||||
taskExecution.setArguments(getTaskArguments(executionId));
|
||||
return taskExecution;
|
||||
}
|
||||
@@ -259,9 +286,13 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
|
||||
@Override
|
||||
public long getTaskExecutionCountByTaskName(String taskName) {
|
||||
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("taskName", taskName, Types.VARCHAR);
|
||||
|
||||
try {
|
||||
return jdbcTemplate.queryForObject(
|
||||
getQuery(TASK_EXECUTION_COUNT_BY_NAME), new Object[] { taskName }, Long.class);
|
||||
return jdbcTemplate.queryForObject(
|
||||
getQuery(TASK_EXECUTION_COUNT_BY_NAME), queryParameters, Long.class);
|
||||
}
|
||||
catch (EmptyResultDataAccessException e) {
|
||||
return 0;
|
||||
@@ -270,20 +301,65 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
|
||||
@Override
|
||||
public long getRunningTaskExecutionCountByTaskName(String taskName) {
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("taskName", taskName, Types.VARCHAR);
|
||||
|
||||
try {
|
||||
return jdbcTemplate.queryForObject(
|
||||
getQuery(RUNNING_TASK_EXECUTION_COUNT_BY_NAME), new Object[] { taskName }, Long.class);
|
||||
getQuery(RUNNING_TASK_EXECUTION_COUNT_BY_NAME), queryParameters, Long.class);
|
||||
}
|
||||
catch (EmptyResultDataAccessException e) {
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<TaskExecution> getLatestTaskExecutionsByTaskNames(String... taskNames) {
|
||||
Assert.notEmpty(taskNames, "At least 1 task name must be provided.");
|
||||
|
||||
final List<String> taskNamesAsList = new ArrayList<>();
|
||||
|
||||
for (String taskName : taskNames) {
|
||||
if (StringUtils.hasText(taskName)) {
|
||||
taskNamesAsList.add(taskName);
|
||||
}
|
||||
}
|
||||
|
||||
Assert.isTrue(taskNamesAsList.size() == taskNames.length,
|
||||
String.format("Task names must not contain any empty elements but %s of %s were empty or null.",
|
||||
taskNames.length - taskNamesAsList.size(), taskNames.length));
|
||||
|
||||
try {
|
||||
final Map<String, List<String>> paramMap = Collections.singletonMap("taskNames", taskNamesAsList);
|
||||
return this.jdbcTemplate.query(
|
||||
getQuery(LAST_TASK_EXECUTIONS_BY_TASK_NAMES), paramMap, new TaskExecutionRowMapper());
|
||||
}
|
||||
catch (EmptyResultDataAccessException e) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public TaskExecution getLatestTaskExecutionForTaskName(String taskName) {
|
||||
Assert.hasText(taskName, "The task name must not be empty.");
|
||||
final List<TaskExecution> taskExecutions = this.getLatestTaskExecutionsByTaskNames(taskName);
|
||||
if (taskExecutions.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
else if (taskExecutions.size() == 1) {
|
||||
return taskExecutions.get(0);
|
||||
}
|
||||
else {
|
||||
throw new IllegalStateException("Only expected a single TaskExecution but received " + taskExecutions.size());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public long getTaskExecutionCount() {
|
||||
|
||||
try {
|
||||
return jdbcTemplate.queryForObject(
|
||||
getQuery(TASK_EXECUTION_COUNT), new Object[] { }, Long.class);
|
||||
getQuery(TASK_EXECUTION_COUNT), new MapSqlParameterSource(), Long.class);
|
||||
}
|
||||
catch (EmptyResultDataAccessException e) {
|
||||
return 0;
|
||||
@@ -293,26 +369,26 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
@Override
|
||||
public Page<TaskExecution> findRunningTaskExecutions(String taskName, Pageable pageable) {
|
||||
return queryForPageableResults(pageable, SELECT_CLAUSE, FROM_CLAUSE,
|
||||
RUNNING_TASK_WHERE_CLAUSE, new Object[]{ taskName },
|
||||
RUNNING_TASK_WHERE_CLAUSE, new MapSqlParameterSource("taskName", taskName),
|
||||
getRunningTaskExecutionCountByTaskName(taskName));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Page<TaskExecution> findTaskExecutionsByName(String taskName, Pageable pageable) {
|
||||
return queryForPageableResults(pageable, SELECT_CLAUSE, FROM_CLAUSE,
|
||||
TASK_NAME_WHERE_CLAUSE, new Object[]{ taskName },
|
||||
TASK_NAME_WHERE_CLAUSE, new MapSqlParameterSource("taskName", taskName),
|
||||
getTaskExecutionCountByTaskName(taskName));
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> getTaskNames() {
|
||||
return jdbcTemplate.queryForList(getQuery(FIND_TASK_NAMES), String.class);
|
||||
return jdbcTemplate.queryForList(getQuery(FIND_TASK_NAMES), new MapSqlParameterSource(), String.class);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Page<TaskExecution> findAll(Pageable pageable) {
|
||||
return queryForPageableResults(pageable, SELECT_CLAUSE, FROM_CLAUSE, null,
|
||||
new Object[]{ }, getTaskExecutionCount());
|
||||
new MapSqlParameterSource(), getTaskExecutionCount());
|
||||
}
|
||||
|
||||
public void setTaskIncrementer(DataFieldMaxValueIncrementer taskIncrementer) {
|
||||
@@ -325,10 +401,13 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
|
||||
@Override
|
||||
public Long getTaskExecutionIdByJobExecutionId(long jobExecutionId) {
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("jobExecutionId", jobExecutionId, Types.BIGINT);
|
||||
|
||||
try {
|
||||
return jdbcTemplate.queryForObject(
|
||||
getQuery(FIND_TASK_EXECUTION_BY_JOB_EXECUTION_ID),
|
||||
new Object[] { jobExecutionId },
|
||||
queryParameters,
|
||||
Long.class);
|
||||
}
|
||||
catch (EmptyResultDataAccessException e) {
|
||||
@@ -338,10 +417,13 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
|
||||
@Override
|
||||
public Set<Long> getJobExecutionIdsByTaskExecutionId(long taskExecutionId) {
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("taskExecutionId", taskExecutionId, Types.BIGINT);
|
||||
|
||||
try {
|
||||
return jdbcTemplate.query(
|
||||
getQuery(FIND_JOB_EXECUTION_BY_TASK_EXECUTION_ID),
|
||||
new Object[] {taskExecutionId},
|
||||
queryParameters,
|
||||
new ResultSetExtractor<Set<Long>>() {
|
||||
@Override
|
||||
public Set<Long> extractData(ResultSet resultSet) throws SQLException, DataAccessException {
|
||||
@@ -362,12 +444,13 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
|
||||
@Override
|
||||
public void updateExternalExecutionId(long taskExecutionId, String externalExecutionId) {
|
||||
Object[] parameters = new Object[]{externalExecutionId,
|
||||
taskExecutionId};
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("externalExecutionId", externalExecutionId, Types.VARCHAR)
|
||||
.addValue("taskExecutionId", taskExecutionId, Types.BIGINT);
|
||||
|
||||
if (jdbcTemplate.update(
|
||||
getQuery(UPDATE_TASK_EXECUTION_EXTERNAL_EXECUTION_ID),
|
||||
parameters,
|
||||
new int[]{Types.VARCHAR, Types.BIGINT}) != 1) {
|
||||
queryParameters) != 1) {
|
||||
throw new IllegalStateException("Invalid TaskExecution, ID "
|
||||
+ taskExecutionId + " not found.");
|
||||
}
|
||||
@@ -377,7 +460,7 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
String selectClause,
|
||||
String fromClause,
|
||||
String whereClause,
|
||||
Object[] queryParam,
|
||||
MapSqlParameterSource queryParameters,
|
||||
long totalCount){
|
||||
SqlPagingQueryProviderFactoryBean factoryBean = new SqlPagingQueryProviderFactoryBean();
|
||||
factoryBean.setSelectClause(selectClause);
|
||||
@@ -413,7 +496,7 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
String query = pagingQueryProvider.getPageQuery(pageable);
|
||||
List<TaskExecution> resultList = jdbcTemplate.query(
|
||||
getQuery(query),
|
||||
queryParam,
|
||||
queryParameters,
|
||||
new TaskExecutionRowMapper());
|
||||
return new PageImpl<>(resultList, pageable, totalCount);
|
||||
}
|
||||
@@ -439,13 +522,14 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
* Convenience method that inserts an individual records into the
|
||||
* TASK_EXECUTION_PARAMS table.
|
||||
*/
|
||||
private void insertArgument(long executionId, String param) {
|
||||
int[] argTypes = new int[]{ Types.BIGINT, Types.VARCHAR };
|
||||
Object[] args = new Object[]{ executionId, param };
|
||||
jdbcTemplate.update(getQuery(CREATE_TASK_ARGUMENT), args, argTypes);
|
||||
private void insertArgument(long taskExecutionId, String taskParam) {
|
||||
final MapSqlParameterSource queryParameters = new MapSqlParameterSource()
|
||||
.addValue("taskExecutionId", taskExecutionId, Types.BIGINT)
|
||||
.addValue("taskParam", taskParam, Types.VARCHAR);
|
||||
jdbcTemplate.update(getQuery(CREATE_TASK_ARGUMENT), queryParameters);
|
||||
}
|
||||
|
||||
private List<String> getTaskArguments(long executionId){
|
||||
private List<String> getTaskArguments(long taskExecutionId){
|
||||
final List<String> params= new ArrayList<>();
|
||||
RowCallbackHandler handler = new RowCallbackHandler() {
|
||||
@Override
|
||||
@@ -453,8 +537,7 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
params.add(rs.getString(2));
|
||||
}
|
||||
};
|
||||
|
||||
jdbcTemplate.query(getQuery(FIND_ARGUMENT_FROM_ID), new Object[] { executionId },
|
||||
jdbcTemplate.query(getQuery(FIND_ARGUMENT_FROM_ID), new MapSqlParameterSource("taskExecutionId", taskExecutionId),
|
||||
handler);
|
||||
return params;
|
||||
}
|
||||
@@ -486,10 +569,9 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
|
||||
parentExecutionId);
|
||||
}
|
||||
|
||||
private Integer getNullableExitCode(ResultSet rs) throws SQLException {
|
||||
int exitCode = rs.getInt("EXIT_CODE");
|
||||
return !rs.wasNull() ? exitCode : null;
|
||||
}
|
||||
private Integer getNullableExitCode(ResultSet rs) throws SQLException {
|
||||
int exitCode = rs.getInt("EXIT_CODE");
|
||||
return !rs.wasNull() ? exitCode : null;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2015-2017 the original author or authors.
|
||||
* Copyright 2015-2018 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.
|
||||
@@ -20,6 +20,7 @@ import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.Comparator;
|
||||
import java.util.Date;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
@@ -33,11 +34,13 @@ import org.springframework.data.domain.Page;
|
||||
import org.springframework.data.domain.PageImpl;
|
||||
import org.springframework.data.domain.Pageable;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* Stores Task Execution Information to a in-memory map.
|
||||
*
|
||||
* @author Glenn Renfro
|
||||
* @author Gunnar Hillert
|
||||
*/
|
||||
public class MapTaskExecutionDao implements TaskExecutionDao {
|
||||
|
||||
@@ -251,4 +254,72 @@ public class MapTaskExecutionDao implements TaskExecutionDao {
|
||||
executionList.subList((int)pageable.getOffset(), (int)toIndex),
|
||||
pageable, maxSize);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<TaskExecution> getLatestTaskExecutionsByTaskNames(String... taskNames) {
|
||||
|
||||
Assert.notEmpty(taskNames, "At least 1 task name must be provided.");
|
||||
|
||||
final List<String> taskNamesAsList = new ArrayList<>();
|
||||
|
||||
for (String taskName : taskNames) {
|
||||
if (StringUtils.hasText(taskName)) {
|
||||
taskNamesAsList.add(taskName);
|
||||
}
|
||||
}
|
||||
|
||||
Assert.isTrue(taskNamesAsList.size() == taskNames.length,
|
||||
String.format("Task names must not contain any empty elements but %s of %s were empty or null.",
|
||||
taskNames.length - taskNamesAsList.size(), taskNames.length));
|
||||
|
||||
final Map<String, TaskExecution> tempTaskExecutions = new HashMap<>();
|
||||
|
||||
for (Map.Entry<Long, TaskExecution> taskExecutionMapEntry : this.taskExecutions.entrySet()) {
|
||||
if (!taskNamesAsList.contains(taskExecutionMapEntry.getValue().getTaskName())) {
|
||||
continue;
|
||||
}
|
||||
|
||||
final TaskExecution tempTaskExecution = tempTaskExecutions.get(taskExecutionMapEntry.getValue().getTaskName());
|
||||
if (tempTaskExecution == null
|
||||
|| tempTaskExecution.getStartTime().before(taskExecutionMapEntry.getValue().getStartTime())
|
||||
|| (
|
||||
tempTaskExecution.getStartTime().equals(taskExecutionMapEntry.getValue().getStartTime())
|
||||
&& tempTaskExecution.getExecutionId() < taskExecutionMapEntry.getValue().getExecutionId()
|
||||
)
|
||||
) {
|
||||
tempTaskExecutions.put(taskExecutionMapEntry.getValue().getTaskName(), taskExecutionMapEntry.getValue());
|
||||
}
|
||||
}
|
||||
final List<TaskExecution> latestTaskExecutions = new ArrayList<>(tempTaskExecutions.values());
|
||||
Collections.sort(latestTaskExecutions, new TaskExecutionComparator());
|
||||
return latestTaskExecutions;
|
||||
}
|
||||
|
||||
@Override
|
||||
public TaskExecution getLatestTaskExecutionForTaskName(String taskName) {
|
||||
Assert.hasText(taskName, "The task name must not be empty.");
|
||||
final List<TaskExecution> taskExecutions = this.getLatestTaskExecutionsByTaskNames(taskName);
|
||||
if (taskExecutions.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
else if (taskExecutions.size() == 1) {
|
||||
return taskExecutions.get(0);
|
||||
}
|
||||
else {
|
||||
throw new IllegalStateException("Only expected a single TaskExecution but received " + taskExecutions.size());
|
||||
}
|
||||
}
|
||||
|
||||
private class TaskExecutionComparator implements Comparator<TaskExecution> {
|
||||
@Override
|
||||
public int compare(TaskExecution firstTaskExecution, TaskExecution secondTaskExecution) {
|
||||
if (firstTaskExecution.getStartTime().equals(secondTaskExecution.getStartTime())) {
|
||||
return Long.compare(firstTaskExecution.getExecutionId(), secondTaskExecution.getExecutionId());
|
||||
}
|
||||
else {
|
||||
return secondTaskExecution.getStartTime().compareTo(firstTaskExecution.getStartTime());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2015-2017 the original author or authors.
|
||||
* Copyright 2015-2018 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.
|
||||
@@ -28,6 +28,8 @@ import org.springframework.data.domain.Pageable;
|
||||
* Data Access Object for task executions.
|
||||
*
|
||||
* @author Glenn Renfro
|
||||
* @author Gunnar Hillert
|
||||
*
|
||||
*/
|
||||
public interface TaskExecutionDao {
|
||||
|
||||
@@ -201,4 +203,33 @@ public interface TaskExecutionDao {
|
||||
*/
|
||||
void updateExternalExecutionId(long taskExecutionId,
|
||||
String externalExecutionId);
|
||||
|
||||
/**
|
||||
* Returns a {@link List} of the latest {@link TaskExecution} for 1 or more task names.
|
||||
*
|
||||
* Latest is defined by the most recent start time. A {@link TaskExecution} does not have to be finished
|
||||
* (The results may including pending {@link TaskExecution}s).
|
||||
*
|
||||
* It is theoretically possible that a {@link TaskExecution} with the same name to have more than 1
|
||||
* {@link TaskExecution} for the exact same start time. In that case the {@link TaskExecution} with the
|
||||
* highest Task Execution ID is returned.
|
||||
*
|
||||
* This method will not consider end times in its calculations. Thus, when a task execution {@code A} starts
|
||||
* after task execution {@code B} but finishes BEFORE task execution {@code A}, then task execution {@code B}
|
||||
* is being returned.
|
||||
*
|
||||
* @param taskNames At least 1 task name must be provided
|
||||
* @return List of TaskExecutions. May be empty but never null.
|
||||
*/
|
||||
List<TaskExecution> getLatestTaskExecutionsByTaskNames(String... taskNames);
|
||||
|
||||
/**
|
||||
* Returns the latest task execution for a given task name. Will ultimately apply the same algorithm underneath
|
||||
* as {@link #getLatestTaskExecutionsByTaskNames(String...)} but will only return a single result.
|
||||
*
|
||||
* @param taskName Must not be null or empty
|
||||
* @return The latest Task Execution or null
|
||||
* @see #getLatestTaskExecutionsByTaskNames(String...)
|
||||
*/
|
||||
TaskExecution getLatestTaskExecutionForTaskName(String taskName);
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2015 the original author or authors.
|
||||
* Copyright 2015-2018 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.
|
||||
@@ -31,6 +31,7 @@ import org.springframework.util.Assert;
|
||||
*
|
||||
* @author Glenn Renfro
|
||||
* @author Michael Minella
|
||||
* @author Gunnar Hillert
|
||||
*/
|
||||
public class SimpleTaskExplorer implements TaskExplorer {
|
||||
|
||||
@@ -92,4 +93,14 @@ public class SimpleTaskExplorer implements TaskExplorer {
|
||||
return taskExecutionDao.getJobExecutionIdsByTaskExecutionId(taskExecutionId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<TaskExecution> getLatestTaskExecutionsByTaskNames(String... taskNames) {
|
||||
return taskExecutionDao.getLatestTaskExecutionsByTaskNames(taskNames);
|
||||
}
|
||||
|
||||
@Override
|
||||
public TaskExecution getLatestTaskExecutionForTaskName(String taskName) {
|
||||
return taskExecutionDao.getLatestTaskExecutionForTaskName(taskName);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user