Add listener to persist relationship between job and task

This introduces a listener that stores the association between a Spring
Batch job and the task it was executed within.

Resolves spring-cloud/spring-cloud-task#46

Merge Changes based on code review.

spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/TaskExplorer.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.task.repository;
 
 import java.util.List;
+import java.util.Set;
 
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
@@ -25,6 +26,7 @@ import org.springframework.data.domain.Pageable;
  * Offers methods that allow users to query the task executions that are available.
  *
  * @author Glenn Renfro
+ * @author Michael Minella
  */
 public interface TaskExplorer {
 
@@ -34,7 +36,7 @@ public interface TaskExplorer {
 	 * @param executionId the task execution id
 	 * @return the {@link TaskExecution} with this id, or null if not found
 	 */
-	public TaskExecution getTaskExecution(long executionId);
+	TaskExecution getTaskExecution(long executionId);
 
 
 	/**
@@ -44,14 +46,14 @@ public interface TaskExplorer {
 	 * @param pageable the constraints for the search
 	 * @return the set of running executions for tasks with the specified name
 	 */
-	public Page<TaskExecution> findRunningTaskExecutions(String taskName, Pageable pageable);
+	Page<TaskExecution> findRunningTaskExecutions(String taskName, Pageable pageable);
 
 	/**
 	 * Retrieve a list of available task names.
 	 *
 	 * @return the set of task names that have been executed
 	 */
-	public List<String> getTaskNames();
+	List<String> getTaskNames();
 
 	/**
 	 * Get number of executions for a taskName.
@@ -59,7 +61,7 @@ public interface TaskExplorer {
 	 * @param taskName the name of the task to be searched
 	 * @return the number of running tasks that have the taskname specified
 	 */
-	public long getTaskExecutionCountByTaskName(String taskName);
+	long getTaskExecutionCountByTaskName(String taskName);
 
 	/**
 	 * Retrieves current number of task executions.
@@ -75,7 +77,7 @@ public interface TaskExplorer {
 	 * @param pageable the constraints for the search
 	 * @return list of task executions
 	 */
-	public Page<TaskExecution> findTaskExecutionsByName(String taskName, Pageable pageable);
+	Page<TaskExecution> findTaskExecutionsByName(String taskName, Pageable pageable);
 
 	/**
 	 * Retrieves all the task executions within the pageable constraints sorted by
@@ -84,6 +86,23 @@ public interface TaskExplorer {
 	 * @param pageable the constraints for the search
 	 * @return page containing the results from the search
 	 */
-	public Page<TaskExecution> findAll(Pageable pageable);
+	Page<TaskExecution> findAll(Pageable pageable);
 
+	/**
+	 * Returns the id of the TaskExecution that the requested Spring Batch job execution
+	 * was executed within the context of.  Returns null if none were found.
+	 *
+	 * @param jobExecutionId the id of the JobExecution
+	 * @return the id of the {@link TaskExecution}
+	 */
+	Long getTaskExecutionIdByJobExecutionId(long jobExecutionId);
+
+	/**
+	 * Returns a Set of JobExecution ids for the jobs that were executed within the scope
+	 * of the requested task.
+	 *
+	 * @param taskExecutionId id of the {@link TaskExecution}
+	 * @return a <code>Set</code> of the ids of the job executions executed within the task.
+	 */
+	Set<Long> getJobExecutionIdsByTaskExecutionId(long taskExecutionId);
 }

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

@@ -20,10 +20,13 @@ import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Types;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.TreeMap;
+import java.util.TreeSet;
 
 import javax.sql.DataSource;
 
@@ -31,12 +34,14 @@ import org.springframework.batch.item.database.Order;
 import org.springframework.cloud.task.repository.TaskExecution;
 import org.springframework.cloud.task.repository.database.PagingQueryProvider;
 import org.springframework.cloud.task.repository.database.support.SqlPagingQueryProviderFactoryBean;
+import org.springframework.dao.DataAccessException;
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageImpl;
 import org.springframework.data.domain.Pageable;
 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.support.incrementer.DataFieldMaxValueIncrementer;
@@ -95,7 +100,10 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
 
 	private static final String FIND_TASK_NAMES = "SELECT distinct TASK_NAME from %PREFIX%EXECUTION order by TASK_NAME";
 
-	private static final String DEFAULT_TABLE_PREFIX = "TASK_";
+	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 = ?";
+
+	public static final String DEFAULT_TABLE_PREFIX = "TASK_";
 
 	private String tablePrefix = DEFAULT_TABLE_PREFIX;
 
@@ -241,6 +249,43 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
 		return taskIncrementer.nextLongValue();
 	}
 
+	@Override
+	public Long getTaskExecutionIdByJobExecutionId(long jobExecutionId) {
+		try {
+			return jdbcTemplate.queryForObject(
+					getQuery(FIND_TASK_EXECUTION_BY_JOB_EXECUTION_ID),
+					new Object[] { jobExecutionId },
+					Long.class);
+		}
+		catch (EmptyResultDataAccessException e) {
+			return null;
+		}
+	}
+
+	@Override
+	public Set<Long> getJobExecutionIdsByTaskExecutionId(long taskExecutionId) {
+		try {
+			return jdbcTemplate.query(
+					getQuery(FIND_JOB_EXECUTION_BY_TASK_EXECUTION_ID),
+					new Object[] {taskExecutionId},
+					new ResultSetExtractor<Set<Long>>() {
+						@Override
+						public Set<Long> extractData(ResultSet resultSet) throws SQLException, DataAccessException {
+							Set<Long> jobExecutionIds = new TreeSet<>();
+
+							while(resultSet.next()) {
+								jobExecutionIds.add(resultSet.getLong("JOB_EXECUTION_ID"));
+							}
+
+							return jobExecutionIds;
+						}
+					});
+		}
+		catch (DataAccessException e) {
+			return Collections.emptySet();
+		}
+	}
+
 	private Page<TaskExecution> queryForPageableResults(Pageable pageable,
 														String selectClause,
 														String fromClause,

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

@@ -42,10 +42,13 @@ public class MapTaskExecutionDao implements TaskExecutionDao {
 
 	private ConcurrentMap<Long, TaskExecution> taskExecutions;
 
+	private ConcurrentMap<Long, Set<Long>> batchJobAssociations;
+
 	private final AtomicLong currentId = new AtomicLong(0L);
 
 	public MapTaskExecutionDao() {
 		taskExecutions = new ConcurrentHashMap<>();
+		batchJobAssociations = new ConcurrentHashMap<>();
 	}
 
 	@Override
@@ -68,7 +71,7 @@ public class MapTaskExecutionDao implements TaskExecutionDao {
 
 	@Override
 	public TaskExecution getTaskExecution(long executionId) {
-		return taskExecutions.get(executionId);
+			return taskExecutions.get(executionId);
 	}
 
 	@Override
@@ -149,6 +152,38 @@ public class MapTaskExecutionDao implements TaskExecutionDao {
 		return currentId.getAndIncrement();
 	}
 
+	@Override
+	public Long getTaskExecutionIdByJobExecutionId(long jobExecutionId) {
+		Long taskId = null;
+
+		found:
+
+		for (Map.Entry<Long, Set<Long>> association : batchJobAssociations.entrySet()) {
+			for (Long curJobExecutionId : association.getValue()) {
+				if(curJobExecutionId.equals(jobExecutionId)) {
+					taskId = association.getKey();
+					break found;
+				}
+			}
+		}
+
+		return taskId;
+	}
+
+	@Override
+	public Set<Long> getJobExecutionIdsByTaskExecutionId(long taskExecutionId) {
+		if(batchJobAssociations.containsKey(taskExecutionId)) {
+			return Collections.unmodifiableSet(batchJobAssociations.get(taskExecutionId));
+		}
+		else {
+			return new TreeSet<>();
+		}
+	}
+
+	public ConcurrentMap<Long, Set<Long>> getBatchJobAssociations() {
+		return batchJobAssociations;
+	}
+
 	private TreeSet<TaskExecution> getTaskExecutionTreeSet() {
 		return new TreeSet<>(new Comparator<TaskExecution>() {
 			@Override

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

@@ -18,6 +18,7 @@ package org.springframework.cloud.task.repository.dao;
 
 import java.util.Date;
 import java.util.List;
+import java.util.Set;
 
 import org.springframework.cloud.task.repository.TaskExecution;
 import org.springframework.data.domain.Page;
@@ -36,7 +37,7 @@ public interface TaskExecutionDao {
 	 * @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.
+	 * @return A fully qualified {@link TaskExecution} instance.
 	 */
 	TaskExecution createTaskExecution( String taskName,
 						   Date startTime, List<String> parameters);
@@ -105,7 +106,7 @@ public interface TaskExecutionDao {
 	 *
 	 * @return a list of distinct task names from the task repository..
 	 */
-	public List<String> getTaskNames();
+	List<String> getTaskNames();
 
 	/**
 	 * Retrieves all the task executions within the pageable constraints.
@@ -113,6 +114,27 @@ public interface TaskExecutionDao {
 	 * @return page containing the results from the search
 	 */
 
-	public Page<TaskExecution> findAll(Pageable pageable);
+	Page<TaskExecution> findAll(Pageable pageable);
 
+	/**
+	 * Retrieves the next available execution id for a task execution.
+	 * @return long containing the executionId.
+	 */
+	long getNextExecutionId();
+
+	/**
+	 * Returns the id of the TaskExecution that the requested Spring Batch job execution
+	 * was executed within the context of.  Returns null if non were found.
+	 *
+	 * @param jobExecutionId the id of the JobExecution
+	 * @return the id of the {@link TaskExecution}
+	 */
+	Long getTaskExecutionIdByJobExecutionId(long jobExecutionId);
+
+	/**
+	 * Returns the job execution ids associated with a task execution id.
+	 * @param taskExecutionId id of the {@link TaskExecution}
+	 * @return a <code>Set</code> of the ids of the job executions executed within the task.
+	 */
+	Set<Long> getJobExecutionIdsByTaskExecutionId(long taskExecutionId);
 }

spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/database/support/SqlPagingQueryUtils.java

@@ -71,7 +71,6 @@ public class SqlPagingQueryUtils {
 	 * Generates WHERE clause for queries that require sub selects.
 	 *
 	 * @param provider the paging query provider that will provide the base where clause
-	 * @return a String for the where clause.
 	 */
 	public static void buildWhereClause( AbstractSqlPagingQueryProvider provider,
 						boolean remainingPageQuery, StringBuilder sql) {

spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/support/SimpleTaskExplorer.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.task.repository.support;
 
 import java.util.List;
+import java.util.Set;
 
 import org.springframework.cloud.task.repository.TaskExecution;
 import org.springframework.cloud.task.repository.TaskExplorer;
@@ -29,8 +30,9 @@ import org.springframework.util.Assert;
  * TaskExplorer for that gathers task information from a task repository.
  *
  * @author Glenn Renfro
+ * @author Michael Minella
  */
-public class SimpleTaskExplorer implements TaskExplorer{
+public class SimpleTaskExplorer implements TaskExplorer {
 
 	private TaskExecutionDao taskExecutionDao;
 
@@ -80,4 +82,14 @@ public class SimpleTaskExplorer implements TaskExplorer{
 		return taskExecutionDao.findAll(pageable);
 	}
 
+	@Override
+	public Long getTaskExecutionIdByJobExecutionId(long jobExecutionId) {
+		return taskExecutionDao.getTaskExecutionIdByJobExecutionId(jobExecutionId);
+	}
+
+	@Override
+	public Set<Long> getJobExecutionIdsByTaskExecutionId(long taskExecutionId) {
+		return taskExecutionDao.getJobExecutionIdsByTaskExecutionId(taskExecutionId);
+	}
+
 }

spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-h2.sql

@@ -16,4 +16,11 @@ CREATE TABLE TASK_EXECUTION_PARAMS  (
 	references TASK_EXECUTION(TASK_EXECUTION_ID)
 ) ;
 
+CREATE TABLE TASK_TASK_BATCH (
+  TASK_EXECUTION_ID BIGINT NOT NULL ,
+  JOB_EXECUTION_ID BIGINT NOT NULL ,
+	constraint TASK_EXEC_BATCH_FK foreign key (TASK_EXECUTION_ID)
+	references TASK_EXECUTION(TASK_EXECUTION_ID)
+) ;
+
 CREATE SEQUENCE TASK_SEQ ;

spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-hsqldb.sql

@@ -16,6 +16,13 @@ CREATE TABLE TASK_EXECUTION_PARAMS  (
 	references TASK_EXECUTION(TASK_EXECUTION_ID)
 ) ;
 
+CREATE TABLE TASK_TASK_BATCH (
+  TASK_EXECUTION_ID BIGINT NOT NULL ,
+  JOB_EXECUTION_ID BIGINT NOT NULL ,
+	constraint TASK_EXEC_BATCH_FK foreign key (TASK_EXECUTION_ID)
+	references TASK_EXECUTION(TASK_EXECUTION_ID)
+) ;
+
 CREATE TABLE TASK_SEQ (
 	ID BIGINT IDENTITY
 );

spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-mysql.sql

@@ -16,6 +16,12 @@ CREATE TABLE TASK_EXECUTION_PARAMS  (
 	references TASK_EXECUTION(TASK_EXECUTION_ID)
 ) ENGINE=InnoDB;
 
+CREATE TABLE TASK_TASK_BATCH (
+  TASK_EXECUTION_ID BIGINT NOT NULL ,
+  JOB_EXECUTION_ID BIGINT NOT NULL ,
+	constraint TASK_EXEC_BATCH_FK foreign key (TASK_EXECUTION_ID)
+	references TASK_EXECUTION(TASK_EXECUTION_ID)
+)  ENGINE=InnoDB;
 
 CREATE TABLE TASK_SEQ (
 	ID BIGINT NOT NULL,

spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-oracle10g.sql

@@ -16,4 +16,11 @@ CREATE TABLE TASK_EXECUTION_PARAMS  (
 	references TASK_EXECUTION(TASK_EXECUTION_ID)
 ) ;
 
+CREATE TABLE TASK_TASK_BATCH (
+  TASK_EXECUTION_ID NUMBER NOT NULL ,
+  JOB_EXECUTION_ID NUMBER NOT NULL ,
+	constraint TASK_EXEC_BATCH_FK foreign key (TASK_EXECUTION_ID)
+	references TASK_EXECUTION(TASK_EXECUTION_ID)
+) ;
+
 CREATE SEQUENCE TASK_SEQ START WITH 0 MINVALUE 0 MAXVALUE 9223372036854775807 NOCACHE NOCYCLE;

spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-postgresql.sql

@@ -16,4 +16,11 @@ CREATE TABLE TASK_EXECUTION_PARAMS  (
 	references TASK_EXECUTION(TASK_EXECUTION_ID)
 ) ;
 
+CREATE TABLE TASK_TASK_BATCH (
+  TASK_EXECUTION_ID BIGINT NOT NULL ,
+  JOB_EXECUTION_ID BIGINT NOT NULL ,
+	constraint TASK_EXEC_BATCH_FK foreign key (TASK_EXECUTION_ID)
+	references TASK_EXECUTION(TASK_EXECUTION_ID)
+) ;
+
 CREATE SEQUENCE TASK_SEQ MAXVALUE 9223372036854775807 NO CYCLE;

spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-sqlserver.sql

@@ -15,4 +15,11 @@ CREATE TABLE TASK_EXECUTION_PARAMS  (
   references TASK_EXECUTION(TASK_EXECUTION_ID)
 ) ;
 
+CREATE TABLE TASK_TASK_BATCH (
+  TASK_EXECUTION_ID BIGINT NOT NULL ,
+  JOB_EXECUTION_ID BIGINT NOT NULL ,
+	constraint TASK_EXEC_BATCH_FK foreign key (TASK_EXECUTION_ID)
+	references TASK_EXECUTION(TASK_EXECUTION_ID)
+) ;
+
 CREATE TABLE TASK_SEQ (ID BIGINT IDENTITY);