SCT-41 Introduces findAll for task explorer

resolves spring-cloud/spring-cloud-task#41

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

@@ -19,6 +19,9 @@ 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;
+
 /**
  * Offers methods that allow users to query the task executions that are available.
  *
@@ -56,7 +59,14 @@ 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 getTaskExecutionCount(String taskName);
+	public long getTaskExecutionCountByTaskName(String taskName);
+
+	/**
+	 * Retrieves current number of task executions.
+	 *
+	 * @return current number of task executions.
+	 */
+	long getTaskExecutionCount();
 
 	/**
 	 * Get a collection/page of executions
@@ -68,5 +78,13 @@ public interface TaskExplorer {
 	 */
 	public List<TaskExecution> getTaskExecutionsByName(String taskName, int start, int count);
 
+	/**
+	 * Retrieves all the task executions within the pageable constraints sorted by
+	 * start date descending, taskExecution id descending.
+	 *
+	 * @param pageable the constraints for the search
+	 * @return page containing the results from the search
+	 */
+	public Page<TaskExecution> findAll(Pageable pageable);
 
 }

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

@@ -23,12 +23,20 @@ import java.util.ArrayList;
 import java.util.Date;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.TreeMap;
 
 import javax.sql.DataSource;
 
+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.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.RowCallbackHandler;
@@ -44,6 +52,12 @@ import org.springframework.util.StringUtils;
 public class JdbcTaskExecutionDao implements TaskExecutionDao {
 
 
+	public static String SELECT_CLAUSE = "TASK_EXECUTION_ID, "
+			+ "START_TIME, END_TIME, TASK_NAME, EXIT_CODE, "
+			+ "EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE ";
+
+	public static String FROM_CLAUSE = "%PREFIX%EXECUTION";
+
 	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, STATUS_CODE) values (?, ?, ?, ?, ?, ?, ?, ?)";
@@ -68,6 +82,9 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
 			+ "TASK_PARAM from %PREFIX%EXECUTION_PARAMS where TASK_EXECUTION_ID = ?";
 
 	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 = ?";
 
 	private static final String FIND_RUNNING_TASK_EXECUTIONS = "SELECT TASK_EXECUTION_ID, "
@@ -76,7 +93,7 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
 			+ "from %PREFIX%EXECUTION where TASK_NAME = ? AND END_TIME IS NULL "
 			+ "order by TASK_EXECUTION_ID";
 
-	private static final String FIND_TASK_EXECUTIONS = "SELECT TASK_EXECUTION_ID, "
+	private static final String FIND_TASK_EXECUTIONS_BY_NAME = "SELECT TASK_EXECUTION_ID, "
 			+ "START_TIME, END_TIME, TASK_NAME, EXIT_CODE, "
 			+ "EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE "
 			+ "from %PREFIX%EXECUTION where TASK_NAME = ? "
@@ -91,9 +108,18 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
 
 	private JdbcOperations jdbcTemplate;
 
+	private DataSource dataSource;
+
+	Map<String, Order> orderMap;
+
 	public JdbcTaskExecutionDao(DataSource dataSource) {
 		Assert.notNull(dataSource);
 		this.jdbcTemplate = new JdbcTemplate(dataSource);
+		this.dataSource = dataSource;
+		orderMap = new TreeMap<>();
+		orderMap.put("START_TIME", Order.DESCENDING);
+		orderMap.put("TASK_EXECUTION_ID", Order.DESCENDING);
+
 	}
 
 	@Override
@@ -156,15 +182,25 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
 	}
 
 	@Override
-	public long getTaskExecutionCount(String taskName) {
+	public long getTaskExecutionCountByTaskName(String taskName) {
 		try {
 		return jdbcTemplate.queryForObject(
-				getQuery(TASK_EXECUTION_COUNT), new Object[] { taskName }, Long.class);
+				getQuery(TASK_EXECUTION_COUNT_BY_NAME), new Object[] { taskName }, Long.class);
 		}
 		catch (EmptyResultDataAccessException e) {
 			return 0;
 		}
+	}
 
+	@Override
+	public long getTaskExecutionCount() {
+		try {
+			return jdbcTemplate.queryForObject(
+					getQuery(TASK_EXECUTION_COUNT), new Object[] {  }, Long.class);
+		}
+		catch (EmptyResultDataAccessException e) {
+			return 0;
+		}
 	}
 
 	@Override
@@ -179,8 +215,8 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
 
 	@Override
 	public List<TaskExecution> getTaskExecutionsByName(String taskName, final int start, final int count) {
-		return jdbcTemplate.query(getQuery(FIND_TASK_EXECUTIONS),
-				new Object[]{ taskName, count, start + 1 }, new TaskExecutionRowMapper());
+		return jdbcTemplate.query(getQuery(FIND_TASK_EXECUTIONS_BY_NAME),
+				new Object[]{ taskName, count, start }, new TaskExecutionRowMapper());
 	}
 
 	@Override
@@ -188,6 +224,30 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
 		return jdbcTemplate.queryForList(getQuery(FIND_TASK_NAMES), String.class);
 	}
 
+	@Override
+	public Page<TaskExecution> findAll(Pageable pageable) {
+
+		SqlPagingQueryProviderFactoryBean factoryBean = new SqlPagingQueryProviderFactoryBean();
+		factoryBean.setSelectClause(SELECT_CLAUSE);
+		factoryBean.setFromClause(FROM_CLAUSE);
+		factoryBean.setSortKeys(orderMap);
+		factoryBean.setDataSource(dataSource);
+		PagingQueryProvider pagingQueryProvider = null;
+		try {
+			pagingQueryProvider = factoryBean.getObject();
+			pagingQueryProvider.init(dataSource);
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+		}
+		String query = pagingQueryProvider.getPageQuery(pageable);
+		List<TaskExecution> resultList = jdbcTemplate.query(
+				getQuery(query),
+				new Object[]{  },
+				new TaskExecutionRowMapper());
+		return new PageImpl<TaskExecution>(resultList, pageable, getTaskExecutionCount());
+	}
+
 	private String getQuery(String base) {
 		return StringUtils.replace(base, "%PREFIX%", tablePrefix);
 	}

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

@@ -27,6 +27,9 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
 import org.springframework.cloud.task.repository.TaskExecution;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageImpl;
+import org.springframework.data.domain.Pageable;
 
 /**
  * Stores Task Execution Information to a in-memory map.
@@ -57,7 +60,7 @@ public class MapTaskExecutionDao implements TaskExecutionDao {
 	}
 
 	@Override
-	public long getTaskExecutionCount(String taskName) {
+	public long getTaskExecutionCountByTaskName(String taskName) {
 		int count = 0;
 		for (Map.Entry<String, TaskExecution> entry : taskExecutions.entrySet()) {
 			if (entry.getValue().getTaskName().equals(taskName)) {
@@ -67,6 +70,11 @@ public class MapTaskExecutionDao implements TaskExecutionDao {
 		return count;
 	}
 
+	@Override
+	public long getTaskExecutionCount() {
+		return taskExecutions.size();
+	}
+
 	@Override
 	public Set<TaskExecution> findRunningTaskExecutions(String taskName) {
 		Set<TaskExecution> result = getTaskExecutionTreeSet();
@@ -101,6 +109,19 @@ public class MapTaskExecutionDao implements TaskExecutionDao {
 		return new ArrayList<String>(result);
 	}
 
+	@Override
+	public Page<TaskExecution> findAll(Pageable pageable) {
+		TreeSet<TaskExecution> sortedSet = getTaskExecutionTreeSet();
+		sortedSet.addAll(taskExecutions.values());
+		List<TaskExecution> result = new ArrayList<>(sortedSet.descendingSet());
+		int toIndex = (pageable.getOffset() + pageable.getPageSize() > result.size()) ?
+				result.size() : pageable.getOffset() + pageable.getPageSize();
+		return new PageImpl<TaskExecution>(
+				result.subList(pageable.getOffset(), toIndex),
+				pageable,
+				getTaskExecutionCount());
+	}
+
 	public Map<String, TaskExecution> getTaskExecutions() {
 		return Collections.unmodifiableMap(taskExecutions);
 	}
@@ -109,7 +130,11 @@ public class MapTaskExecutionDao implements TaskExecutionDao {
 		return new TreeSet<TaskExecution>(new Comparator<TaskExecution>() {
 			@Override
 			public int compare(TaskExecution e1, TaskExecution e2) {
-				return e1.getExecutionId().compareTo(e2.getExecutionId());
+				int result = e1.getStartTime().compareTo(e2.getStartTime());
+				if (result == 0){
+					result = e1.getExecutionId().compareTo(e2.getExecutionId());
+				}
+				return result;
 			}
 		});
 	}

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

@@ -20,6 +20,8 @@ import java.util.List;
 import java.util.Set;
 
 import org.springframework.cloud.task.repository.TaskExecution;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
 
 /**
  * Data Access Object for task executions.
@@ -54,9 +56,16 @@ public interface TaskExecutionDao {
 	 * Retrieves current number of task executions for a taskName.
 	 *
 	 * @param taskName the name of the task to search for in the repository.
+	 * @return current number of task executions for the taskName.
+	 */
+	long getTaskExecutionCountByTaskName(String taskName);
+
+	/**
+	 * Retrieves current number of task executions.
+	 *
 	 * @return current number of task executions.
 	 */
-	long getTaskExecutionCount(String taskName);
+	long getTaskExecutionCount();
 
 	/**
 	 * Retrieves a set of task executions that are running for a taskName.
@@ -83,4 +92,12 @@ public interface TaskExecutionDao {
 	 * @return a list of distinct task names from the task repository..
 	 */
 	public List<String> getTaskNames();
+
+	/**
+	 * Retrieves all the task executions within the pageable constraints.
+	 * @param pageable the constraints for the search
+	 * @return page containing the results from the search
+	 */
+
+	public Page<TaskExecution> findAll(Pageable pageable);
 }

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

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database;
+
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.springframework.batch.item.database.Order;
+import org.springframework.data.domain.Pageable;
+
+/**
+ * Interface defining the functionality to be provided for generating paging queries.
+ * @author Glenn Renfro
+ */
+public interface PagingQueryProvider {
+
+	/**
+	 * Initialize the query provider using the provided {@link DataSource} if necessary.
+	 *
+	 * @param dataSource DataSource to use for any initialization
+	 */
+	void init(DataSource dataSource) throws Exception;
+
+	/**
+	 * The number of parameters that are declared in the query
+	 * @return number of parameters
+	 */
+	int getParameterCount();
+
+	/**
+	 * Indicate whether the generated queries use named parameter syntax.
+	 *
+	 * @return true if named parameter syntax is used
+	 */
+	boolean isUsingNamedParameters();
+
+	/**
+	 * The sort keys.  A Map of the columns that make up the key and a Boolean indicating ascending or descending
+	 * (ascending = true).
+	 *
+	 * @return the sort keys used to order the query
+	 */
+	Map<String, Order> getSortKeys();
+
+	/**
+	 *
+	 * Generate the query that will provide the jump to item query.
+	 *
+	 * @param pageable the coordinates to pull the next page from the datasource
+	 * @return the generated query
+	 */
+	String getPageQuery(Pageable pageable);
+}

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

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.springframework.batch.item.database.JdbcParameterUtils;
+import org.springframework.batch.item.database.Order;
+import org.springframework.cloud.task.repository.database.PagingQueryProvider;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Abstract SQL Paging Query Provider to serve as a base class for all provided
+ * SQL paging query providers.
+ *
+ * Any implementation must provide a way to specify the select clause, from
+ * clause and optionally a where clause. It is recommended that there should be an index for
+ * the sort key to provide better performance.
+ *
+ * Provides properties and preparation for the mandatory "selectClause" and
+ * "fromClause" as well as for the optional "whereClause".
+ *
+ * @author Glenn Renfro
+ */
+public abstract class AbstractSqlPagingQueryProvider implements PagingQueryProvider {
+
+	private String selectClause;
+
+	private String fromClause;
+
+	private String whereClause;
+
+	private Map<String, Order> sortKeys = new LinkedHashMap<String, Order>();
+
+	private int parameterCount;
+
+	private boolean usingNamedParameters;
+
+	/**
+	 * @param selectClause SELECT clause part of SQL query string
+	 */
+	public void setSelectClause(String selectClause) {
+		this.selectClause = removeKeyWord("select", selectClause);
+	}
+
+	/**
+	 *
+	 * @return SQL SELECT clause part of SQL query string
+	 */
+	protected String getSelectClause() {
+		return selectClause;
+	}
+
+	/**
+	 * @param fromClause FROM clause part of SQL query string
+	 */
+	public void setFromClause(String fromClause) {
+		this.fromClause = removeKeyWord("from", fromClause);
+	}
+
+	/**
+	 *
+	 * @return SQL FROM clause part of SQL query string
+	 */
+	protected String getFromClause() {
+		return fromClause;
+	}
+
+	/**
+	 * @param whereClause WHERE clause part of SQL query string
+	 */
+	public void setWhereClause(String whereClause) {
+		if (StringUtils.hasText(whereClause)) {
+			this.whereClause = removeKeyWord("where", whereClause);
+		}
+		else {
+			this.whereClause = null;
+		}
+	}
+
+	/**
+	 *
+	 * @return SQL WHERE clause part of SQL query string
+	 */
+	protected String getWhereClause() {
+		return whereClause;
+	}
+
+	/**
+	 * @param sortKeys key to use to sort and limit page content
+	 */
+	public void setSortKeys(Map<String, Order> sortKeys) {
+		this.sortKeys = sortKeys;
+	}
+
+	/**
+	 * A Map&lt;String, Order&gt; of sort columns as the key and {@link Order} for ascending/descending.
+	 *
+	 * @return sortKey key to use to sort and limit page content
+	 */
+	@Override
+	public Map<String, Order> getSortKeys() {
+		return sortKeys;
+	}
+
+	@Override
+	public int getParameterCount() {
+		return parameterCount;
+	}
+
+	@Override
+	public boolean isUsingNamedParameters() {
+		return usingNamedParameters;
+	}
+
+	@Override
+	public void init(DataSource dataSource) throws Exception {
+		Assert.notNull(dataSource);
+		Assert.hasLength(selectClause, "selectClause must be specified");
+		Assert.hasLength(fromClause, "fromClause must be specified");
+		Assert.notEmpty(sortKeys, "sortKey must be specified");
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(selectClause);
+		sql.append(" FROM ").append(fromClause);
+		if (whereClause != null) {
+			sql.append(" WHERE ").append(whereClause);
+		}
+		List<String> namedParameters = new ArrayList<String>();
+		parameterCount = JdbcParameterUtils.countParameterPlaceholders(sql.toString(), namedParameters);
+		if (namedParameters.size() > 0) {
+			if (parameterCount != namedParameters.size()) {
+				throw new InvalidDataAccessApiUsageException(
+						"You can't use both named parameters and classic \"?\" placeholders: " + sql);
+			}
+			usingNamedParameters = true;
+		}
+	}
+	private String removeKeyWord(String keyWord, String clause) {
+		String temp = clause.trim();
+		String keyWordString = keyWord + " ";
+		if (temp.toLowerCase().startsWith(keyWordString) && temp.length() > keyWordString.length()) {
+			return temp.substring(keyWordString.length());
+		}
+		else {
+			return temp;
+		}
+	}
+}

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import org.springframework.cloud.task.repository.database.PagingQueryProvider;
+import org.springframework.data.domain.Pageable;
+
+/**
+ * HSQLDB implementation of a {@link PagingQueryProvider} using database specific features.
+ *
+ * @author Glenn Renfro
+ */
+public class HsqlPagingQueryProvider extends AbstractSqlPagingQueryProvider {
+
+	@Override
+	public String getPageQuery(Pageable pageable) {
+		String topClause = new StringBuilder().append("LIMIT ")
+				.append(pageable.getOffset()).append(" ")
+				.append(pageable.getPageSize()).toString();
+		return SqlPagingQueryUtils.generateTopJumpToQuery(this, topClause);
+	}
+
+}

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import org.springframework.cloud.task.repository.database.PagingQueryProvider;
+import org.springframework.data.domain.Pageable;
+
+/**
+ * MySQL implementation of a {@link PagingQueryProvider} using database specific features.
+ *
+ * @author Glenn Renfro
+ */
+public class MySqlPagingQueryProvider extends AbstractSqlPagingQueryProvider {
+	@Override
+	public String getPageQuery(Pageable pageable) {
+		String topClause = new StringBuilder().append("LIMIT ")
+				.append(pageable.getOffset()).append(", ")
+				.append(pageable.getPageSize()).toString();
+		return SqlPagingQueryUtils.generateLimitJumpToQuery(this, topClause);
+	}
+
+}

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import org.springframework.cloud.task.repository.database.PagingQueryProvider;
+import org.springframework.data.domain.Pageable;
+
+/**
+ * Oracle implementation of a {@link PagingQueryProvider} using database specific features.
+ *
+ * @author Glenn Renfro
+ */
+public class OraclePagingQueryProvider extends AbstractSqlPagingQueryProvider {
+
+	@Override
+	public String getPageQuery(Pageable pageable) {
+		int offset = pageable.getOffset()+1;
+		return SqlPagingQueryUtils.generateRowNumSqlQueryWithNesting(this, getSelectClause(), getSelectClause(), false, "TMP_ROW_NUM >= "
+				+ offset + " AND TMP_ROW_NUM < " + (offset+pageable.getPageSize()));
+	}
+}

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import org.springframework.cloud.task.repository.database.PagingQueryProvider;
+import org.springframework.data.domain.Pageable;
+
+/**
+ * Postgres implementation of a  {@link PagingQueryProvider} using database specific features.
+ *
+ * @author Glenn Renfro
+ */
+public class PostgresPagingQueryProvider extends AbstractSqlPagingQueryProvider {
+
+	@Override
+	public String getPageQuery(Pageable pageable) {
+		String limitClause = new StringBuilder().append("LIMIT ").
+				append(pageable.getPageSize()).append(" OFFSET ").
+				append(pageable.getOffset()).toString();
+		return SqlPagingQueryUtils.generateLimitJumpToQuery(this, limitClause);
+	}
+}

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

@@ -0,0 +1,168 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+
+import static org.springframework.cloud.task.repository.support.DatabaseType.HSQL;
+import static org.springframework.cloud.task.repository.support.DatabaseType.MYSQL;
+import static org.springframework.cloud.task.repository.support.DatabaseType.ORACLE;
+import static org.springframework.cloud.task.repository.support.DatabaseType.POSTGRES;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.sql.DataSource;
+
+import org.springframework.batch.item.database.Order;
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.cloud.task.repository.database.PagingQueryProvider;
+import org.springframework.cloud.task.repository.support.DatabaseType;
+import org.springframework.jdbc.support.MetaDataAccessException;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Factory bean for {@link PagingQueryProvider} interface. The database type
+ * will be determined from the data source if not provided explicitly. Valid
+ * types are given by the {@link DatabaseType} enum.
+ *
+ * @author Glenn Renfro
+ */
+public class SqlPagingQueryProviderFactoryBean implements FactoryBean<PagingQueryProvider> {
+
+	private DataSource dataSource;
+
+	private String databaseType;
+
+	private String fromClause;
+
+	private String whereClause;
+
+	private String selectClause;
+
+	private Map<String, Order> sortKeys;
+
+	private Map<DatabaseType, AbstractSqlPagingQueryProvider> providers = new HashMap<DatabaseType, AbstractSqlPagingQueryProvider>();
+
+
+	{
+		providers.put(HSQL, new HsqlPagingQueryProvider());
+		providers.put(MYSQL, new MySqlPagingQueryProvider());
+		providers.put(POSTGRES, new PostgresPagingQueryProvider());
+		providers.put(ORACLE, new OraclePagingQueryProvider());
+	}
+
+	/**
+	 * @param databaseType the databaseType to set
+	 */
+	public void setDatabaseType(String databaseType) {
+		Assert.hasText(databaseType, "databaseType must not be empty nor null");
+		this.databaseType = databaseType;
+	}
+
+	/**
+	 * @param dataSource the dataSource to set
+	 */
+	public void setDataSource(DataSource dataSource) {
+		Assert.notNull(dataSource, "dataSource must not be null");
+		this.dataSource = dataSource;
+	}
+
+	/**
+	 * @param fromClause the fromClause to set
+	 */
+	public void setFromClause(String fromClause) {
+		Assert.hasText(fromClause, "fromClause must not be empty nor null");
+		this.fromClause = fromClause;
+	}
+
+	/**
+	 * @param whereClause the whereClause to set
+	 */
+	public void setWhereClause(String whereClause) {
+		this.whereClause = whereClause;
+	}
+
+	/**
+	 * @param selectClause the selectClause to set
+	 */
+	public void setSelectClause(String selectClause) {
+		Assert.hasText(selectClause, "selectClause must not be empty nor null");
+		this.selectClause = selectClause;
+	}
+
+	/**
+	 * @param sortKeys the sortKeys to set
+	 */
+	public void setSortKeys(Map<String, Order> sortKeys) {
+		this.sortKeys = sortKeys;
+	}
+
+	/**
+	 * Get a {@link PagingQueryProvider} instance using the provided properties
+	 * and appropriate for the given database type.
+	 *
+	 * @see FactoryBean#getObject()
+	 */
+	@Override
+	public PagingQueryProvider getObject() throws Exception {
+
+		DatabaseType type;
+		try {
+			type = databaseType != null ? DatabaseType.valueOf(databaseType.toUpperCase()) : DatabaseType
+					.fromMetaData(dataSource);
+		}
+		catch (MetaDataAccessException e) {
+			throw new IllegalArgumentException(
+					"Could not inspect meta data for database type.  You have to supply it explicitly.", e);
+		}
+
+		AbstractSqlPagingQueryProvider provider = providers.get(type);
+		Assert.state(provider != null, "Should not happen: missing PagingQueryProvider for DatabaseType=" + type);
+
+		provider.setFromClause(fromClause);
+		provider.setWhereClause(whereClause);
+		provider.setSortKeys(sortKeys);
+		if (StringUtils.hasText(selectClause)) {
+			provider.setSelectClause(selectClause);
+		}
+		provider.init(dataSource);
+
+		return provider;
+
+	}
+
+	/**
+	 * Always returns {@link PagingQueryProvider}.
+	 *
+	 * @see FactoryBean#getObjectType()
+	 */
+	@Override
+	public Class<PagingQueryProvider> getObjectType() {
+		return PagingQueryProvider.class;
+	}
+
+	/**
+	 * Always returns true.
+	 *
+	 * @see FactoryBean#isSingleton()
+	 */
+	@Override
+	public boolean isSingleton() {
+		return true;
+	}
+}

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

@@ -0,0 +1,143 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import java.util.Map;
+
+import org.springframework.batch.item.database.Order;
+
+/**
+ * Utility class that generates the actual SQL statements used by query
+ * providers.
+ *
+ * @author Glenn Renfro
+ */
+public class SqlPagingQueryUtils {
+
+	/**
+	 * Generate SQL query string using a LIMIT clause
+	 *
+	 * @param provider {@link AbstractSqlPagingQueryProvider} providing the
+	 * implementation specifics
+	 * @param limitClause the implementation specific top clause to be used
+	 * @return the generated query
+	 */
+	public static String generateLimitJumpToQuery(AbstractSqlPagingQueryProvider provider, String limitClause) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(provider.getSelectClause());
+		sql.append(" FROM ").append(provider.getFromClause());
+		sql.append(provider.getWhereClause() == null ? "" : " WHERE " + provider.getWhereClause());
+		sql.append(" ORDER BY ").append(buildSortClause(provider));
+		sql.append(" " + limitClause);
+
+		return sql.toString();
+	}
+
+	/**
+	 * Generate SQL query string using a TOP clause
+	 *
+	 * @param provider {@link AbstractSqlPagingQueryProvider} providing the
+	 * implementation specifics
+	 * @param topClause the implementation specific top clause to be used
+	 * @return the generated query
+	 */
+	public static String generateTopJumpToQuery(AbstractSqlPagingQueryProvider provider, String topClause) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(topClause).append(" ").append(provider.getSelectClause());
+		sql.append(" FROM ").append(provider.getFromClause());
+		sql.append(provider.getWhereClause() == null ? "" : " WHERE " + provider.getWhereClause());
+		sql.append(" ORDER BY ").append(buildSortClause(provider));
+
+		return sql.toString();
+	}
+
+	public static String generateRowNumSqlQueryWithNesting(AbstractSqlPagingQueryProvider provider,
+														   String selectClause, boolean remainingPageQuery, String rowNumClause) {
+		return generateRowNumSqlQueryWithNesting(provider, selectClause, selectClause, remainingPageQuery, rowNumClause);
+	}
+
+	public static String generateRowNumSqlQueryWithNesting(AbstractSqlPagingQueryProvider provider,
+														   String innerSelectClause, String outerSelectClause, boolean remainingPageQuery, String rowNumClause) {
+
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(outerSelectClause).append(" FROM (SELECT ").append(outerSelectClause)
+				.append(", ").append("ROWNUM as TMP_ROW_NUM");
+		sql.append(" FROM (SELECT ").append(innerSelectClause).append(" FROM ").append(provider.getFromClause());
+		buildWhereClause(provider, remainingPageQuery, sql);
+		sql.append(" ORDER BY ").append(buildSortClause(provider));
+		sql.append(")) WHERE ").append(rowNumClause);
+
+		return sql.toString();
+
+	}
+
+	private static void buildWhereClause(AbstractSqlPagingQueryProvider provider, boolean remainingPageQuery,
+										 StringBuilder sql) {
+		if (remainingPageQuery) {
+			sql.append(" WHERE ");
+			if (provider.getWhereClause() != null) {
+				sql.append("(");
+				sql.append(provider.getWhereClause());
+				sql.append(") AND ");
+			}
+
+		}
+		else {
+			sql.append(provider.getWhereClause() == null ? "" : " WHERE " + provider.getWhereClause());
+		}
+	}
+
+	/**
+	 * Generates ORDER BY attributes based on the sort keys.
+	 *
+	 * @param provider {@link AbstractSqlPagingQueryProvider} providing the
+	 * implementation specifics
+	 * @return a String that can be appended to an ORDER BY clause.
+	 */
+	public static String buildSortClause(AbstractSqlPagingQueryProvider provider) {
+		return buildSortClause(provider.getSortKeys());
+	}
+
+	/**
+	 * Generates ORDER BY attributes based on the sort keys.
+	 *
+	 * @param sortKeys generates order by clause from map
+	 * @return a String that can be appended to an ORDER BY clause.
+	 */
+	public static String buildSortClause(Map<String, Order> sortKeys) {
+		StringBuilder builder = new StringBuilder();
+		String prefix = "";
+
+		for (Map.Entry<String, Order> sortKey : sortKeys.entrySet()) {
+			builder.append(prefix);
+
+			prefix = ", ";
+
+			builder.append(sortKey.getKey());
+
+			if(sortKey.getValue() != null && sortKey.getValue() == Order.DESCENDING) {
+				builder.append(" DESC");
+			}
+			else {
+				builder.append(" ASC");
+			}
+		}
+
+		return builder.toString();
+	}
+
+}

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

@@ -1,52 +0,0 @@
-/*
- * Copyright 2015 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.cloud.task.repository.support;
-
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-import org.springframework.cloud.task.repository.TaskExecution;
-import org.springframework.cloud.task.repository.TaskExplorer;
-
-/**
- * Provides a no-op TaskExplorer for development purposes.
- *
- * @author Michael Minella
- */
-public class NoOpTaskExplorer implements TaskExplorer {
-
-	public TaskExecution getTaskExecution(String executionId) {
-		return null;
-	}
-
-	public Set<TaskExecution> findRunningTaskExecutions(String taskName) {
-		return new HashSet<TaskExecution>(0);
-	}
-
-	public List<String> getTaskNames() {
-		return new ArrayList<String>(0);
-	}
-
-	public long getTaskExecutionCount(String taskName) {
-		return 0;
-	}
-
-	public List<TaskExecution> getTaskExecutionsByName(String taskName, int start, int count) {
-		return new ArrayList<TaskExecution>(0);
-	}
-}

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

@@ -22,6 +22,8 @@ import java.util.Set;
 import org.springframework.cloud.task.repository.TaskExecution;
 import org.springframework.cloud.task.repository.TaskExplorer;
 import org.springframework.cloud.task.repository.dao.TaskExecutionDao;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
 import org.springframework.util.Assert;
 
 /**
@@ -54,12 +56,23 @@ public class SimpleTaskExplorer implements TaskExplorer{
 	}
 
 	@Override
-	public long getTaskExecutionCount(String taskName) {
-		return taskExecutionDao.getTaskExecutionCount(taskName);
+	public long getTaskExecutionCountByTaskName(String taskName) {
+		return taskExecutionDao.getTaskExecutionCountByTaskName(taskName);
+	}
+
+	@Override
+	public long getTaskExecutionCount() {
+		return taskExecutionDao.getTaskExecutionCount();
 	}
 
 	@Override
 	public List<TaskExecution> getTaskExecutionsByName(String taskName, int start, int count) {
 		return taskExecutionDao.getTaskExecutionsByName(taskName, start, count);
 	}
+
+	@Override
+	public Page<TaskExecution> findAll(Pageable pageable) {
+		return taskExecutionDao.findAll(pageable);
+	}
+
 }

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

@@ -29,6 +29,11 @@ import org.springframework.util.Assert;
  */
 public class SimpleTaskRepository implements TaskRepository {
 
+	public static final int MAX_EXIT_MESSAGE_SIZE = 2500;
+	public static final int MAX_TASK_NAME_SIZE = 100;
+	public static final int MAX_STATUS_CODE_SIZE = 10;
+	public static final int MAX_EXECUTION_ID_SIZE = 100;
+
 	private final static Logger logger = LoggerFactory.getLogger(SimpleTaskRepository.class);
 
 	private TaskExecutionDao taskExecutionDao;
@@ -67,7 +72,27 @@ public class SimpleTaskRepository implements TaskRepository {
 	 */
 	private void validateTaskExecution(TaskExecution taskExecution) {
 		Assert.notNull(taskExecution, "taskExecution should not be null");
-		Assert.notNull(taskExecution.getExecutionId(), "taskExecutionId should not be null");
+		Assert.hasText(taskExecution.getExecutionId(), "taskExecutionId should not be null");
 		Assert.notNull(taskExecution.getStartTime(), "TaskExecution start time cannot be null.");
+
+		if (taskExecution.getTaskName() != null &&
+				taskExecution.getTaskName().length() > MAX_TASK_NAME_SIZE) {
+			throw new IllegalArgumentException("TaskName length exceeds "
+					+ MAX_TASK_NAME_SIZE + " characters");
+		}
+		if (taskExecution.getStatusCode() != null &&
+				taskExecution.getStatusCode().length() > MAX_STATUS_CODE_SIZE) {
+			throw new IllegalArgumentException("StatusCode length exceeds "
+					+ MAX_STATUS_CODE_SIZE + " characters");
+		}
+		if (taskExecution.getExecutionId().length() > MAX_EXECUTION_ID_SIZE) {
+			throw new IllegalArgumentException("ExecutionID length exceeds "
+					+ MAX_EXECUTION_ID_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));
+		}
 	}
 }

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/repository/database/support/FindAllPagingQueryProviderTests.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.springframework.cloud.task.util.TestDBUtils;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+
+/**
+ * @author Glenn Renfro
+ */
+@RunWith(Parameterized.class)
+public class FindAllPagingQueryProviderTests {
+
+	private String databaseProductName;
+	private String expectedQuery;
+	private Pageable pageable = new PageRequest(0, 10);
+
+	@Parameterized.Parameters
+	public static Collection<Object[]> data() {
+		return Arrays.asList(new Object[][]{
+				{"Oracle", "SELECT TASK_EXECUTION_ID, START_TIME, END_TIME, TASK_NAME, "
+						+ "EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE FROM "
+						+ "(SELECT TASK_EXECUTION_ID, START_TIME, END_TIME, TASK_NAME, "
+						+ "EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE, ROWNUM as "
+						+ "TMP_ROW_NUM FROM (SELECT TASK_EXECUTION_ID, START_TIME, "
+						+ "END_TIME, TASK_NAME, EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, "
+						+ "STATUS_CODE FROM %PREFIX%EXECUTION ORDER BY START_TIME DESC, "
+						+ "TASK_EXECUTION_ID DESC)) WHERE TMP_ROW_NUM >= 1 AND "
+						+ "TMP_ROW_NUM < 11"},
+				{"HSQL Database Engine","SELECT LIMIT 0 10 TASK_EXECUTION_ID, "
+						+ "START_TIME, END_TIME, TASK_NAME, EXIT_CODE, EXIT_MESSAGE, "
+						+ "LAST_UPDATED, STATUS_CODE FROM %PREFIX%EXECUTION ORDER BY "
+						+ "START_TIME DESC, TASK_EXECUTION_ID DESC"},
+				{"PostgreSQL","SELECT TASK_EXECUTION_ID, START_TIME, END_TIME, "
+						+ "TASK_NAME, EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE "
+						+ "FROM %PREFIX%EXECUTION ORDER BY START_TIME DESC, "
+						+ "TASK_EXECUTION_ID DESC LIMIT 10 OFFSET 0"},
+				{"MySQL","SELECT TASK_EXECUTION_ID, START_TIME, END_TIME, TASK_NAME, "
+						+ "EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE FROM "
+						+ "%PREFIX%EXECUTION ORDER BY START_TIME DESC, "
+						+ "TASK_EXECUTION_ID DESC LIMIT 0, 10"}
+		});
+	}
+
+	public FindAllPagingQueryProviderTests(String databaseProductName, String expectedQuery) {
+		this.databaseProductName = databaseProductName;
+		this.expectedQuery = expectedQuery;
+	}
+
+	@Test
+	public void testGeneratedQuery() throws Exception{
+		String actualQuery = TestDBUtils.getPagingQueryProvider(databaseProductName).getPageQuery(pageable);
+		assertEquals(String.format(
+				"the generated query for %s, was not the expected query",
+				databaseProductName), expectedQuery, actualQuery);
+	}
+
+}

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/repository/database/support/InvalidPagingQueryProviderTests.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import org.junit.Test;
+import org.springframework.cloud.task.util.TestDBUtils;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+
+/**
+ * @author Glenn Renfro
+ */
+public class InvalidPagingQueryProviderTests {
+
+	@Test(expected = IllegalStateException.class)
+	public void testInvalidDatabase() throws Exception{
+		Pageable pageable = new PageRequest(0, 10);
+		String actualQuery = TestDBUtils.getPagingQueryProvider("Invalid").getPageQuery(pageable);
+	}
+}

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/repository/database/support/SqlPagingQueryProviderFactoryBeanTests.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Map;
+import java.util.TreeMap;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.item.database.Order;
+import org.springframework.cloud.task.repository.dao.JdbcTaskExecutionDao;
+import org.springframework.cloud.task.repository.database.PagingQueryProvider;
+import org.springframework.cloud.task.util.TestDBUtils;
+
+/**
+ * @author Glenn Renfro
+ */
+public class SqlPagingQueryProviderFactoryBeanTests {
+
+	private SqlPagingQueryProviderFactoryBean factoryBean;
+
+	@Before
+	public void setup() throws Exception{
+		factoryBean = new SqlPagingQueryProviderFactoryBean();
+		factoryBean.setDataSource(TestDBUtils.getMockDataSource("MySQL"));
+		factoryBean.setDatabaseType("Oracle");
+		factoryBean.setSelectClause(JdbcTaskExecutionDao.SELECT_CLAUSE);
+		factoryBean.setFromClause(JdbcTaskExecutionDao.FROM_CLAUSE);
+		Map<String, Order> orderMap = new TreeMap<>();
+		orderMap.put("START_TIME", Order.DESCENDING);
+		orderMap.put("TASK_EXECUTION_ID", Order.DESCENDING);
+		factoryBean.setSortKeys(orderMap);
+
+	}
+
+	@Test
+	public void testDatabaseType() throws Exception{
+		PagingQueryProvider pagingQueryProvider = factoryBean.getObject();
+		assertThat(pagingQueryProvider, instanceOf(OraclePagingQueryProvider.class));
+	}
+
+	@Test
+	public void testIsSingleton(){
+		assertTrue(factoryBean.isSingleton());
+	}
+
+}

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/repository/database/support/WhereClausePagingQueryProviderTests.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.repository.database.support;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.springframework.cloud.task.repository.database.PagingQueryProvider;
+import org.springframework.cloud.task.util.TestDBUtils;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+
+/**
+ * @author Glenn Renfro
+ */
+@RunWith(Parameterized.class)
+public class WhereClausePagingQueryProviderTests {
+
+	private String databaseProductName;
+	private String expectedQuery;
+	private Pageable pageable = new PageRequest(0, 10);
+
+	@Parameterized.Parameters
+	public static Collection<Object[]> data() {
+		return Arrays.asList(new Object[][]{
+				{"Oracle", "SELECT TASK_EXECUTION_ID, START_TIME, END_TIME, TASK_NAME, "
+						+ "EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE FROM "
+						+ "(SELECT TASK_EXECUTION_ID, START_TIME, END_TIME, TASK_NAME, "
+						+ "EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE, ROWNUM as "
+						+ "TMP_ROW_NUM FROM (SELECT TASK_EXECUTION_ID, START_TIME, "
+						+ "END_TIME, TASK_NAME, EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, "
+						+ "STATUS_CODE FROM %PREFIX%EXECUTION "
+						+ "WHERE TASK_EXECUTION_ID = '0000' ORDER BY START_TIME DESC, "
+						+ "TASK_EXECUTION_ID DESC)) WHERE TMP_ROW_NUM >= 1 AND "
+						+ "TMP_ROW_NUM < 11"},
+				{"HSQL Database Engine","SELECT LIMIT 0 10 TASK_EXECUTION_ID, "
+						+ "START_TIME, END_TIME, TASK_NAME, EXIT_CODE, EXIT_MESSAGE, "
+						+ "LAST_UPDATED, STATUS_CODE FROM %PREFIX%EXECUTION "
+						+ "WHERE TASK_EXECUTION_ID = '0000' ORDER BY "
+						+ "START_TIME DESC, TASK_EXECUTION_ID DESC"},
+				{"PostgreSQL","SELECT TASK_EXECUTION_ID, START_TIME, END_TIME, "
+						+ "TASK_NAME, EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE "
+						+ "FROM %PREFIX%EXECUTION WHERE TASK_EXECUTION_ID = '0000' "
+						+ "ORDER BY START_TIME DESC, "
+						+ "TASK_EXECUTION_ID DESC LIMIT 10 OFFSET 0"},
+				{"MySQL","SELECT TASK_EXECUTION_ID, START_TIME, END_TIME, TASK_NAME, "
+						+ "EXIT_CODE, EXIT_MESSAGE, LAST_UPDATED, STATUS_CODE FROM "
+						+ "%PREFIX%EXECUTION WHERE TASK_EXECUTION_ID = '0000' "
+						+ "ORDER BY START_TIME DESC, "
+						+ "TASK_EXECUTION_ID DESC LIMIT 0, 10"}
+		});
+	}
+
+	public WhereClausePagingQueryProviderTests(String databaseProductName, String expectedQuery) {
+		this.databaseProductName = databaseProductName;
+		this.expectedQuery = expectedQuery;
+	}
+
+	@Test
+	public void testGeneratedQuery() throws Exception{
+		PagingQueryProvider pagingQueryProvider =
+				TestDBUtils.getPagingQueryProvider(databaseProductName,
+						"TASK_EXECUTION_ID = '0000'");
+		String actualQuery = pagingQueryProvider.getPageQuery(pageable);
+		assertEquals(String.format(
+				"the generated query for %s, was not the expected query",
+				databaseProductName), expectedQuery, actualQuery);
+	}
+}

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/repository/support/DatabaseTypeTests.java

@@ -19,6 +19,7 @@ import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.springframework.cloud.task.repository.support.DatabaseType.HSQL;
+import static org.springframework.cloud.task.repository.support.DatabaseType.MYSQL;
 import static org.springframework.cloud.task.repository.support.DatabaseType.ORACLE;
 import static org.springframework.cloud.task.repository.support.DatabaseType.POSTGRES;
 import static org.springframework.cloud.task.repository.support.DatabaseType.fromProductName;
@@ -45,6 +46,7 @@ public class DatabaseTypeTests {
 		assertEquals(HSQL, fromProductName("HSQL Database Engine"));
 		assertEquals(ORACLE, fromProductName("Oracle"));
 		assertEquals(POSTGRES, fromProductName("PostgreSQL"));
+		assertEquals(MYSQL, fromProductName("MySQL"));
 	}
 
 	@Test(expected = IllegalArgumentException.class)
@@ -70,6 +72,12 @@ public class DatabaseTypeTests {
 		assertEquals(POSTGRES, DatabaseType.fromMetaData(ds));
 	}
 
+	@Test
+	public void testFromMetaDataForMySQL() throws Exception {
+		DataSource ds = getMockDataSource("MySQL");
+		assertEquals(MYSQL, DatabaseType.fromMetaData(ds));
+	}
+
 	public  DataSource getMockDataSource(String databaseProductName) throws Exception {
 		DatabaseMetaData dmd = mock(DatabaseMetaData.class);
 		DataSource ds = mock(DataSource.class);

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/repository/support/NoOpTaskExplorerTests.java

@@ -1,59 +0,0 @@
-/*
- * Copyright 2015 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.cloud.task.repository.support;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-
-import org.junit.Before;
-import org.junit.Test;
-
-import org.springframework.cloud.task.repository.TaskExplorer;
-
-public class NoOpTaskExplorerTests {
-
-	private TaskExplorer taskExplorer;
-
-	@Before
-	public void setUp() throws Exception {
-		taskExplorer = new NoOpTaskExplorer();
-	}
-
-	@Test
-	public void testGetTaskExecution() throws Exception {
-		assertNull(taskExplorer.getTaskExecution("abc"));
-	}
-
-	@Test
-	public void testFindRunningTaskExecutions() throws Exception {
-		assertEquals(taskExplorer.findRunningTaskExecutions("foo").size(), 0);
-	}
-
-	@Test
-	public void testGetTaskNames() throws Exception {
-		assertEquals(taskExplorer.getTaskNames().size(), 0);
-	}
-
-	@Test
-	public void testGetTaskExecutionCount() throws Exception {
-		assertEquals(taskExplorer.getTaskExecutionCount("foo"), 0);
-	}
-
-	@Test
-	public void testGetTaskExecutionsByName() throws Exception {
-		assertEquals(taskExplorer.getTaskExecutionsByName("foo", 0, 100).size(), 0);
-	}
-}

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/repository/support/SimpleTaskExplorerTests.java

@@ -16,19 +16,23 @@
 
 package org.springframework.cloud.task.repository.support;
 
+import static junit.framework.Assert.assertEquals;
 import static junit.framework.Assert.assertNotNull;
+import static junit.framework.Assert.assertNull;
 import static junit.framework.Assert.assertTrue;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Comparator;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.UUID;
 
 import javax.sql.DataSource;
@@ -40,7 +44,6 @@ import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
 import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
@@ -52,6 +55,9 @@ import org.springframework.cloud.task.repository.dao.TaskExecutionDao;
 import org.springframework.cloud.task.util.TestVerifierUtils;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.core.io.ResourceLoader;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
 
 /**
  * @author Glenn Renfro
@@ -77,23 +83,24 @@ public class SimpleTaskExplorerTests {
 
 	@Parameterized.Parameters
 	public static Collection<Object> data() {
-		return Arrays.asList(new Object[] {
-				 DaoType.jdbc ,  DaoType.map  });
+		return Arrays.asList(new Object[]{
+				DaoType.jdbc, DaoType.map });
 	}
 
 	public SimpleTaskExplorerTests(DaoType testType) {
 		this.testType = testType;
 	}
+
 	@Rule
 	public ExpectedException expected = ExpectedException.none();
 
 	@Before
 	public void testDefaultContext() throws Exception {
 
-		if(testType == DaoType.jdbc){
+		if (testType == DaoType.jdbc) {
 			initializeJdbcExplorerTest();
 		}
-		else{
+		else {
 			initializeMapExplorerTest();
 		}
 
@@ -109,13 +116,7 @@ public class SimpleTaskExplorerTests {
 
 	@Test
 	public void getTaskExecution() {
-		final int TEST_COUNT = 5;
-		Map<String, TaskExecution> expectedResults = new HashMap<>();
-		for (int i = 0; i < TEST_COUNT; i++) {
-			TaskExecution expectedTaskExecution = createAndSaveTaskExecution();
-			expectedResults.put(expectedTaskExecution.getExecutionId(),
-					expectedTaskExecution);
-		}
+		Map<String, TaskExecution> expectedResults = createSampleDataSet(5);
 		for (String taskExecutionId : expectedResults.keySet()) {
 			TaskExecution actualTaskExecution =
 					taskExplorer.getTaskExecution(taskExecutionId);
@@ -129,39 +130,36 @@ public class SimpleTaskExplorerTests {
 
 	@Test
 	public void taskExecutionNotFound() {
-		final int TEST_COUNT = 5;
-		Map<String, TaskExecution> expectedResults = new HashMap<>();
-		for (int i = 0; i < TEST_COUNT; i++) {
-			TaskExecution expectedTaskExecution = createAndSaveTaskExecution();
-			expectedResults.put(expectedTaskExecution.getExecutionId(),
-					expectedTaskExecution);
-		}
+		Map<String, TaskExecution> expectedResults = createSampleDataSet(5);
 
 		TaskExecution actualTaskExecution =
 				taskExplorer.getTaskExecution("NO_EXECUTION_PRESENT");
-			assertNull(String.format(
-					"expected null for actualTaskExecution %s", testType),
-					actualTaskExecution);
+		assertNull(String.format(
+				"expected null for actualTaskExecution %s", testType),
+				actualTaskExecution);
 	}
 
 	@Test
 	public void getTaskCountByTaskName() {
-		final int TEST_COUNT = 5;
-		Map<String, TaskExecution> expectedResults = new HashMap<>();
-		for (int i = 0; i < TEST_COUNT; i++) {
-			TaskExecution expectedTaskExecution = createAndSaveTaskExecution();
-			expectedResults.put(expectedTaskExecution.getExecutionId(),
-					expectedTaskExecution);
-		}
+		Map<String, TaskExecution> expectedResults = createSampleDataSet(5);
 		for (Map.Entry<String, TaskExecution> entry : expectedResults.entrySet()) {
 			String taskName = entry.getValue().getTaskName();
 			assertEquals(String.format(
 					"task count for task name did not match expected result for testType %s",
 					testType),
-					1, taskExplorer.getTaskExecutionCount(taskName));
+					1, taskExplorer.getTaskExecutionCountByTaskName(taskName));
 		}
 	}
 
+	@Test
+	public void getTaskCount() {
+		Map<String, TaskExecution> expectedResults = createSampleDataSet(33);
+		assertEquals(String.format(
+				"task count did not match expected result for test Type %s",
+				testType),
+				33, taskExplorer.getTaskExecutionCount());
+	}
+
 	@Test
 	public void findRunningTasks() {
 		final int TEST_COUNT = 2;
@@ -227,7 +225,7 @@ public class SimpleTaskExplorerTests {
 					expectedResults.containsKey(result.getExecutionId()));
 			assertEquals(
 					String.format("taskName for taskExecution is incorrect for testType %s",
-					testType), TASK_NAME, result.getTaskName());
+							testType), TASK_NAME, result.getTaskName());
 		}
 	}
 
@@ -246,13 +244,83 @@ public class SimpleTaskExplorerTests {
 		}
 	}
 
+	@Test
+	public void findAllExecutionsOffBoundry() {
+		Pageable pageable = new PageRequest(0, 10);
+		verifyPageResults(pageable, 103);
+	}
+
+	@Test
+	public void findAllExecutionsOffBoundryByOne() {
+		Pageable pageable = new PageRequest(0, 10);
+		verifyPageResults(pageable, 101);
+	}
+
+	@Test
+	public void findAllExecutionsOnBoundry() {
+		Pageable pageable = new PageRequest(0, 10);
+		verifyPageResults(pageable, 100);
+	}
+
+	@Test
+	public void findAllExecutionsNoResult() {
+		Pageable pageable = new PageRequest(0, 10);
+		verifyPageResults(pageable, 0);
+	}
+
+	private void verifyPageResults(Pageable pageable, int totalNumberOfExecs) {
+		Map<String, TaskExecution> expectedResults = createSampleDataSet(totalNumberOfExecs);
+		List<String> sortedExecIds = getSortedOfTaskExecIds(expectedResults);
+		Iterator<String> expectedTaskExecutionIter = sortedExecIds.iterator();
+		//Verify pageable totals
+		Page taskPage = taskExplorer.findAll(pageable);
+		int pagesExpected = (int) Math.ceil(totalNumberOfExecs / ((double) pageable.getPageSize()));
+		assertEquals("actual page count return was not the expected total",
+				pagesExpected,
+				taskPage.getTotalPages());
+		assertEquals("actual element count was not the expected count", totalNumberOfExecs,
+				taskPage.getTotalElements());
+
+		//Verify pagination
+		Pageable actualPageable = new PageRequest(0, pageable.getPageSize());
+		boolean hasMorePages = taskPage.hasContent();
+		int pageNumber = 0;
+		int elementCount = 0;
+		while (hasMorePages) {
+			taskPage = taskExplorer.findAll(actualPageable);
+			hasMorePages = taskPage.hasNext();
+			List<TaskExecution> actualTaskExecutions = taskPage.getContent();
+			int expectedPageSize = pageable.getPageSize();
+			if (!hasMorePages && pageable.getPageSize() != actualTaskExecutions.size()) {
+				expectedPageSize = totalNumberOfExecs % pageable.getPageSize();
+			}
+			assertEquals(
+					String.format("Element count on page did not match on the %n page",
+							pageNumber), expectedPageSize, actualTaskExecutions.size());
+			for (TaskExecution actualExecution : actualTaskExecutions) {
+				assertEquals(String.format("Element on page %n did not match expected",
+						pageNumber), expectedTaskExecutionIter.next(),
+						actualExecution.getExecutionId());
+				TestVerifierUtils.verifyTaskExecution(
+						expectedResults.get(actualExecution.getExecutionId()),
+						actualExecution);
+				elementCount++;
+			}
+			actualPageable = taskPage.nextPageable();
+			pageNumber++;
+		}
+		//Verify actual totals
+		assertEquals("Pages processed did not equal expected", pagesExpected, pageNumber);
+		assertEquals("Elements processed did not equal expected,", totalNumberOfExecs, elementCount);
+	}
+
 	private TaskExecution createAndSaveTaskExecution() {
 		TaskExecution taskExecution = TestVerifierUtils.createSampleTaskExecution();
 		dao.saveTaskExecution(taskExecution);
 		return taskExecution;
 	}
 
-	private void initializeJdbcExplorerTest(){
+	private void initializeJdbcExplorerTest() {
 		this.context = new AnnotationConfigApplicationContext();
 		this.context.register(TestConfiguration.class,
 				EmbeddedDataSourceConfiguration.class,
@@ -273,6 +341,39 @@ public class SimpleTaskExplorerTests {
 				AutowireCapableBeanFactory.AUTOWIRE_BY_TYPE, false);
 	}
 
+	private Map<String, TaskExecution> createSampleDataSet(int count){
+		Map<String, TaskExecution> expectedResults = new HashMap<>();
+		for (int i = 0; i < count; i++) {
+			TaskExecution expectedTaskExecution = createAndSaveTaskExecution();
+			expectedResults.put(expectedTaskExecution.getExecutionId(),
+					expectedTaskExecution);
+		}
+		return expectedResults;
+	}
+
+	private List<String> getSortedOfTaskExecIds(Map<String, TaskExecution> taskExecutionMap){
+		List<String> sortedExecIds = new ArrayList<>(taskExecutionMap.size());
+		TreeSet sortedSet = getTreeSet();
+		sortedSet.addAll(taskExecutionMap.values());
+		Iterator <TaskExecution> iterator = sortedSet.descendingIterator();
+		while(iterator.hasNext()){
+			sortedExecIds.add(iterator.next().getExecutionId());
+		}
+		return sortedExecIds;
+	}
+
+	private TreeSet getTreeSet(){
+		return new TreeSet<TaskExecution>(new Comparator<TaskExecution>() {
+			@Override
+			public int compare(TaskExecution e1, TaskExecution e2) {
+				int result = e1.getStartTime().compareTo(e2.getStartTime());
+				if (result == 0){
+					result = e1.getExecutionId().compareTo(e2.getExecutionId());
+				}
+				return result;
+			}
+		});
+	}
 
 	private enum DaoType{jdbc, map}
 }

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/repository/support/SimpleTaskRepositoryJdbcTests.java

@@ -93,5 +93,33 @@ public class SimpleTaskRepositoryJdbcTests {
 				dataSource, expectedTaskExecution.getExecutionId());
 		TestVerifierUtils.verifyTaskExecution(expectedTaskExecution, actualTaskExecution);
 	}
+
+	@Test
+	public void testCreateTaskExecutionNoParamMaxExitMessageSize(){
+		TaskExecution expectedTaskExecution = TestVerifierUtils.createSampleTaskExecutionNoParam();
+		expectedTaskExecution.setExitMessage(new String(new char[SimpleTaskRepository.MAX_EXIT_MESSAGE_SIZE+1]));
+		taskRepository.createTaskExecution(expectedTaskExecution);
+	}
+
+	@Test(expected=IllegalArgumentException.class)
+	public void testCreateTaskExecutionNoParamMaxTaskName(){
+		TaskExecution expectedTaskExecution = TestVerifierUtils.createSampleTaskExecutionNoParam();
+		expectedTaskExecution.setTaskName(new String(new char[SimpleTaskRepository.MAX_TASK_NAME_SIZE+1]));
+		taskRepository.createTaskExecution(expectedTaskExecution);
+	}
+
+	@Test(expected=IllegalArgumentException.class)
+	public void testCreateTaskExecutionNoParamMaxStatus(){
+		TaskExecution expectedTaskExecution = TestVerifierUtils.createSampleTaskExecutionNoParam();
+		expectedTaskExecution.setStatusCode(new String(new char[SimpleTaskRepository.MAX_STATUS_CODE_SIZE+1]));
+		taskRepository.createTaskExecution(expectedTaskExecution);
+	}
+
+	@Test(expected=IllegalArgumentException.class)
+	public void testCreateTaskExecutionNoParamMaxExecutionId(){
+		TaskExecution expectedTaskExecution = TestVerifierUtils.createSampleTaskExecutionNoParam();
+		expectedTaskExecution.setExecutionId(new String(new char[SimpleTaskRepository.MAX_EXECUTION_ID_SIZE+1]));
+		taskRepository.createTaskExecution(expectedTaskExecution);
+	}
 }
 

spring-cloud-task-core/src/test/java/org/springframework/cloud/task/util/TestDBUtils.java

@@ -17,16 +17,25 @@
 package org.springframework.cloud.task.util;
 
 import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;
 
 import javax.sql.DataSource;
 
+import org.springframework.batch.item.database.Order;
 import org.springframework.cloud.task.repository.TaskExecution;
+import org.springframework.cloud.task.repository.dao.JdbcTaskExecutionDao;
+import org.springframework.cloud.task.repository.database.PagingQueryProvider;
+import org.springframework.cloud.task.repository.database.support.SqlPagingQueryProviderFactoryBean;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.core.RowMapper;
 
@@ -39,10 +48,10 @@ import org.springframework.jdbc.core.RowMapper;
 public class TestDBUtils {
 
 	/**
-	 * Retrieves the TaskExecution from the datasource
+	 * Retrieves the TaskExecution from the datasource.
 	 *
-	 * @param dataSource      The datasource from which to retrieve the taskExecution
-	 * @param taskExecutionId The id of the task to search .
+	 * @param dataSource      The datasource from which to retrieve the taskExecution.
+	 * @param taskExecutionId The id of the task to search.
 	 * @return taskExecution
 	 */
 	public static TaskExecution getTaskExecutionFromDB(DataSource dataSource,
@@ -73,6 +82,58 @@ public class TestDBUtils {
 		return taskExecution;
 	}
 
+	/**
+	 * Create a pagingQueryProvider specific database type with a findAll.
+	 * @param databaseProductName of the database.
+	 * @return a PagingQueryPovider that will return all the requested information.
+	 * @throws Exception
+	 */
+	public static PagingQueryProvider getPagingQueryProvider(String databaseProductName) throws Exception{
+		return getPagingQueryProvider(databaseProductName, null);
+	}
+
+	/**
+	 * Create a pagingQueryProvider specific database type with a query containing a where clause.
+	 * @param databaseProductName of the database.
+	 * @param whereClause  to be applied to the query.
+	 * @return a PagingQueryProvider that will return the requested information.
+	 * @throws Exception
+	 */
+	public static PagingQueryProvider getPagingQueryProvider(String databaseProductName,
+															 String whereClause) throws Exception{
+		DataSource dataSource = getMockDataSource(databaseProductName);
+		Map<String, Order> orderMap = new TreeMap<>();
+		orderMap.put("START_TIME", Order.DESCENDING);
+		orderMap.put("TASK_EXECUTION_ID", Order.DESCENDING);
+		SqlPagingQueryProviderFactoryBean factoryBean = new SqlPagingQueryProviderFactoryBean();
+		factoryBean.setSelectClause(JdbcTaskExecutionDao.SELECT_CLAUSE);
+		factoryBean.setFromClause(JdbcTaskExecutionDao.FROM_CLAUSE);
+		if(whereClause != null){
+			factoryBean.setWhereClause(whereClause);
+		}
+		factoryBean.setSortKeys(orderMap);
+		factoryBean.setDataSource(dataSource);
+		PagingQueryProvider pagingQueryProvider = null;
+		try {
+			pagingQueryProvider = factoryBean.getObject();
+			pagingQueryProvider.init(dataSource);
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+		}
+		return pagingQueryProvider;
+	}
+
+	public static DataSource getMockDataSource(String databaseProductName) throws Exception {
+		DatabaseMetaData dmd = mock(DatabaseMetaData.class);
+		DataSource ds = mock(DataSource.class);
+		Connection con = mock(Connection.class);
+		when(ds.getConnection()).thenReturn(con);
+		when(con.getMetaData()).thenReturn(dmd);
+		when(dmd.getDatabaseProductName()).thenReturn(databaseProductName);
+		return ds;
+	}
+
 	private static void populateParamsToDB(DataSource dataSource, TaskExecution taskExecution) {
 		String sql = "SELECT * FROM TASK_EXECUTION_PARAMS WHERE TASK_EXECUTION_ID = '"
 				+ taskExecution.getExecutionId() + "'";