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

Updating task execution id next val to use sequence for sqlserver

Browse Source 
resolves TASK-782

Updated PR to resolve issue when no task tables exist

Updated PR based on review comments.
This does not include updated tests.
This resolves a problem where task would fail because the incrementer selection occured before the tables were created.

Added unit test for SqlServerSequenceIncrementer

updated based on code review
This commit is contained in:
Glenn Renfro
2021-05-07 14:11:37 -04:00
committed by Glenn Renfro
parent 5d26173837
commit 17415eaed5
5 changed files with 162 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
docs/src/main/asciidoc/appendix-task-repository-schema.adoc
View File

@@ -83,3 +83,16 @@ Used for the `single-instance-enabled` feature discussed <<features-single-insta
NOTE: The DDL for setting up tables for each database type can be found https://github.com/spring-cloud/spring-cloud-task/tree/master/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task[here].
--
=== SQL Server
By default Spring Cloud Task uses a sequence table for determining the `TASK_EXECUTION_ID` for the `TASK_EXECUTION` table.
However, when launching multiple tasks simultaneously while using SQL Server, this can cause a deadlock to occur on the `TASK_SEQ` table.
The resolution is to drop the `TASK_EXECUTION_SEQ` table and create a sequence using the same name. For example:
```
DROP TABLE TASK_SEQ;
CREATE SEQUENCE [DBO].[TASK_SEQ] AS BIGINT
START WITH 1
INCREMENT BY 1;
```
NOTE: Set the `START WITH` to a higher value than your current execution id.

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2020 the original author or authors.
* Copyright 2015-2021 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.
@@ -152,17 +152,6 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
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 final NamedParameterJdbcTemplate jdbcTemplate;
private String tablePrefix = TaskProperties.DEFAULT_TABLE_PREFIX;
private DataSource dataSource;
private LinkedHashMap<String, Order> orderMap;
private DataFieldMaxValueIncrementer taskIncrementer;
private static final Set<String> validSortColumns = new HashSet<>(10);
static {
@@ -178,6 +167,12 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
validSortColumns.add("PARENT_EXECUTION_ID");
}
private final NamedParameterJdbcTemplate jdbcTemplate;
private String tablePrefix = TaskProperties.DEFAULT_TABLE_PREFIX;
private DataSource dataSource;
private LinkedHashMap<String, Order> orderMap;
private DataFieldMaxValueIncrementer taskIncrementer;
/**
* Initializes the JdbcTaskExecutionDao.
* @param dataSource used by the dao to execute queries and update the tables.
@@ -530,10 +525,11 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
for (Sort.Order sortOrder : sort) {
if (validSortColumns.contains(sortOrder.getProperty().toUpperCase())) {
sortOrderMap.put(sortOrder.getProperty(),
sortOrder.isAscending() ? Order.ASCENDING : Order.DESCENDING);
sortOrder.isAscending() ? Order.ASCENDING : Order.DESCENDING);
}
else {
throw new IllegalArgumentException(String.format("Invalid sort option selected: %s", sortOrder.getProperty()));
throw new IllegalArgumentException(
String.format("Invalid sort option selected: %s", sortOrder.getProperty()));
}
}
}
@@ -576,8 +572,8 @@ public class JdbcTaskExecutionDao implements TaskExecutionDao {
}
/**
* Convenience method that inserts an individual records into the
* TASK_EXECUTION_PARAMS table.
* Convenience method that inserts an individual records into the TASK_EXECUTION_PARAMS
* table.
* @param taskExecutionId id of a task execution
* @param taskParam task parameters
*/

37
spring-cloud-task-core/src/main/java/org/springframework/cloud/task/repository/support/SqlServerSequenceMaxValueIncrementer.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* Copyright 2021-2021 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
*
* https://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 javax.sql.DataSource;
import org.springframework.jdbc.support.incrementer.AbstractSequenceMaxValueIncrementer;
/**
* Incrementer using SQL Server's sequence.
* @author Glenn Renfro
* @since 2.3.2
*/
public class SqlServerSequenceMaxValueIncrementer extends AbstractSequenceMaxValueIncrementer {
SqlServerSequenceMaxValueIncrementer(DataSource dataSource, String incrementerName) {
super(dataSource, incrementerName);
}
@Override
protected String getSequenceQuery() {
return "select next value for " + getIncrementerName();
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2019 the original author or authors.
* Copyright 2015-2021 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.
@@ -16,17 +16,24 @@
package org.springframework.cloud.task.repository.support;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import javax.sql.DataSource;
import org.springframework.batch.item.database.support.DataFieldMaxValueIncrementerFactory;
import org.springframework.batch.item.database.support.DefaultDataFieldMaxValueIncrementerFactory;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.cloud.task.configuration.TaskProperties;
import org.springframework.cloud.task.listener.TaskException;
import org.springframework.cloud.task.repository.dao.JdbcTaskExecutionDao;
import org.springframework.cloud.task.repository.dao.MapTaskExecutionDao;
import org.springframework.cloud.task.repository.dao.TaskExecutionDao;
import org.springframework.jdbc.support.MetaDataAccessException;
import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* A {@link FactoryBean} implementation that creates the appropriate
@@ -81,7 +88,26 @@ public class TaskExecutionDaoFactoryBean implements FactoryBean<TaskExecutionDao
this.dao = new MapTaskExecutionDao();
}
}
if (this.dataSource != null) {
String databaseType = null;
try {
databaseType = DatabaseType.fromMetaData(dataSource).name();
}
catch (MetaDataAccessException e) {
throw new IllegalStateException(e);
}
if (StringUtils.hasText(databaseType) && databaseType.equals("SQLSERVER")) {
String incrementerName = this.tablePrefix + "SEQ";
DataFieldMaxValueIncrementerFactory incrementerFactory = new DefaultDataFieldMaxValueIncrementerFactory(
dataSource);
DataFieldMaxValueIncrementer incrementer = incrementerFactory
.getIncrementer(databaseType, incrementerName);
if (!isSqlServerTableSequenceAvailable(incrementerName)) {
incrementer = new SqlServerSequenceMaxValueIncrementer(dataSource, this.tablePrefix + "SEQ");
}
((JdbcTaskExecutionDao) this.dao).setTaskIncrementer(incrementer);
}
}
return this.dao;
}
@@ -107,7 +133,27 @@ public class TaskExecutionDaoFactoryBean implements FactoryBean<TaskExecutionDao
throw new IllegalStateException(e);
}
((JdbcTaskExecutionDao) this.dao).setTaskIncrementer(incrementerFactory
.getIncrementer(databaseType, this.tablePrefix + "SEQ"));
.getIncrementer(databaseType, this.tablePrefix + "SEQ"));
}
private boolean isSqlServerTableSequenceAvailable(String incrementerName) {
boolean result = false;
DatabaseMetaData metaData = null;
try {
metaData = dataSource.getConnection().getMetaData();
String[] types = { "TABLE" };
ResultSet tables = metaData.getTables(null, null, "%", types);
while (tables.next()) {
if (tables.getString("TABLE_NAME").equals(incrementerName)) {
result = true;
break;
}
}
}
catch (SQLException sqe) {
throw new TaskException(sqe.getMessage());
}
return result;
}
}

51
spring-cloud-task-core/src/test/java/org/springframework/cloud/task/repository/support/SqlServerSequenceMaxValueIncrementerTests.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* Copyright 2020-2020 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
*
* https://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 javax.sql.DataSource;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import static org.assertj.core.api.Assertions.assertThat;
public class SqlServerSequenceMaxValueIncrementerTests {
private ConfigurableApplicationContext context;
@AfterEach
public void tearDown() {
if (this.context != null) {
this.context.close();
}
}
@Test
public void testDefaultDataSourceConfiguration() throws Exception {
this.context = new AnnotationConfigApplicationContext(
TaskExecutionDaoFactoryBeanTests.DefaultDataSourceConfiguration.class);
DataSource dataSource = this.context.getBean(DataSource.class);
SqlServerSequenceMaxValueIncrementer incrementer = new SqlServerSequenceMaxValueIncrementer(dataSource, "foo");
assertThat(incrementer.getSequenceQuery()).isEqualTo("select next value for foo");
assertThat(incrementer.getIncrementerName()).isEqualTo("foo");
}
}