diff --git a/spring-cloud-task-core/pom.xml b/spring-cloud-task-core/pom.xml index cb269590..c408fd6c 100755 --- a/spring-cloud-task-core/pom.xml +++ b/spring-cloud-task-core/pom.xml @@ -60,6 +60,16 @@ org.springframework.data spring-data-commons + + org.springframework.integration + spring-integration-core + true + + + org.springframework.integration + spring-integration-jdbc + true + javax javaee-api diff --git a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/EnableTask.java b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/EnableTask.java index ac61e8d8..26d99364 100644 --- a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/EnableTask.java +++ b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/EnableTask.java @@ -59,6 +59,6 @@ import org.springframework.context.annotation.Import; @Retention(RetentionPolicy.RUNTIME) @Documented @Inherited -@Import(SimpleTaskConfiguration.class) +@Import({ SimpleTaskConfiguration.class, SingleTaskConfiguration.class }) public @interface EnableTask { } diff --git a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/SingleInstanceTaskListener.java b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/SingleInstanceTaskListener.java new file mode 100644 index 00000000..0c89d1a3 --- /dev/null +++ b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/SingleInstanceTaskListener.java @@ -0,0 +1,154 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * 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.configuration; + +import javax.sql.DataSource; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.cloud.task.listener.TaskExecutionException; +import org.springframework.cloud.task.listener.annotation.AfterTask; +import org.springframework.cloud.task.listener.annotation.BeforeTask; +import org.springframework.cloud.task.listener.annotation.FailedTask; +import org.springframework.cloud.task.repository.TaskExecution; +import org.springframework.cloud.task.repository.TaskNameResolver; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationListener; +import org.springframework.integration.jdbc.lock.DefaultLockRepository; +import org.springframework.integration.jdbc.lock.JdbcLockRegistry; +import org.springframework.integration.leader.DefaultCandidate; +import org.springframework.integration.leader.event.OnFailedToAcquireMutexEvent; +import org.springframework.integration.leader.event.OnGrantedEvent; +import org.springframework.integration.support.leader.LockRegistryLeaderInitiator; +import org.springframework.integration.support.locks.LockRegistry; + +/** + * When spring.cloud.task.singleInstanceEnabled is set to true this listener will create a lock for the task + * based on the spring.cloud.task.name. If a lock already exists this Listener will throw + * a TaskExecutionException. If this listener is added manually, then it should + * be added as the first listener in the chain. + * + * @author Glenn Renfro + * @since 2.0.0 + */ +public class SingleInstanceTaskListener implements ApplicationListener { + + private final static Log logger = LogFactory.getLog(SingleInstanceTaskListener.class); + + private LockRegistry lockRegistry; + + private LockRegistryLeaderInitiator lockRegistryLeaderInitiator; + + private TaskNameResolver taskNameResolver; + + private ApplicationEventPublisher applicationEventPublisher; + + private boolean lockReady; + + private boolean lockFailed; + + private DataSource dataSource; + + private TaskProperties taskProperties; + + public SingleInstanceTaskListener(LockRegistry lockRegistry, + TaskNameResolver taskNameResolver, + TaskProperties taskProperties, + ApplicationEventPublisher applicationEventPublisher) { + this.lockRegistry = lockRegistry; + this.taskNameResolver = taskNameResolver; + this.taskProperties = taskProperties; + this.lockRegistryLeaderInitiator = new LockRegistryLeaderInitiator(this.lockRegistry); + this.applicationEventPublisher = applicationEventPublisher; + } + + public SingleInstanceTaskListener(DataSource dataSource, + TaskNameResolver taskNameResolver, + TaskProperties taskProperties, + ApplicationEventPublisher applicationEventPublisher) { + this.taskNameResolver = taskNameResolver; + this.applicationEventPublisher = applicationEventPublisher; + this.dataSource = dataSource; + this.taskProperties = taskProperties; + } + + @BeforeTask + public void lockTask(TaskExecution taskExecution) { + if(this.lockRegistry == null ) { + this.lockRegistry = getDefaultLockRegistry(taskExecution.getExecutionId()); + } + this.lockRegistryLeaderInitiator = new LockRegistryLeaderInitiator( + this.lockRegistry, + new DefaultCandidate(String.valueOf(taskExecution.getExecutionId()), + taskNameResolver.getTaskName())); + this.lockRegistryLeaderInitiator.setApplicationEventPublisher(this.applicationEventPublisher); + this.lockRegistryLeaderInitiator.setPublishFailedEvents(true); + this.lockRegistryLeaderInitiator.start(); + while (!this.lockReady) { + try { + Thread.sleep(this.taskProperties.getSingleInstanceLockCheckInterval()); + } + catch (InterruptedException ex) { + logger.warn("Thread Sleep Failed", ex); + } + if (this.lockFailed) { + String errorMessage = String.format( + "Task with name \"%s\" is already running.", + this.taskNameResolver.getTaskName()); + try { + this.lockRegistryLeaderInitiator.destroy(); + } + catch (Exception exception) { + throw new TaskExecutionException("Failed to destroy lock.", exception); + } + throw new TaskExecutionException(errorMessage); + } + } + } + + @AfterTask + public void unlockTaskOnEnd(TaskExecution taskExecution) throws Exception { + this.lockRegistryLeaderInitiator.destroy(); + } + + @FailedTask + public void unlockTaskOnError(TaskExecution taskExecution) throws Exception { + this.lockRegistryLeaderInitiator.destroy(); + } + + @Override + public void onApplicationEvent(ApplicationEvent applicationEvent) { + if (applicationEvent instanceof OnGrantedEvent) { + this.lockReady = true; + } + else if (applicationEvent instanceof OnFailedToAcquireMutexEvent) { + this.lockFailed = true; + } + } + + private LockRegistry getDefaultLockRegistry( long executionId) { + DefaultLockRepository lockRepository = + new DefaultLockRepository(this.dataSource, String.valueOf( + executionId)); + lockRepository.setPrefix(this.taskProperties.getTablePrefix()); + lockRepository.setTimeToLive(this.taskProperties.getSingleInstanceLockTtl()); + lockRepository.afterPropertiesSet(); + return new JdbcLockRegistry(lockRepository); + } +} diff --git a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/SingleTaskConfiguration.java b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/SingleTaskConfiguration.java new file mode 100644 index 00000000..4a599732 --- /dev/null +++ b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/SingleTaskConfiguration.java @@ -0,0 +1,72 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * 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.configuration; + +import java.util.Collection; + +import javax.sql.DataSource; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.cloud.task.repository.support.SimpleTaskNameResolver; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; +import org.springframework.integration.support.locks.PassThruLockRegistry; +import org.springframework.util.CollectionUtils; + +/** + * Autoconfiguration of {@link SingleInstanceTaskListener}. + * + * @author Glenn Renfro + * @since 2.0.0 + */ + +@Order(Ordered.HIGHEST_PRECEDENCE) +@Configuration +@ConditionalOnProperty(prefix = "spring.cloud.task", name = "singleInstanceEnabled", havingValue = "true") +public class SingleTaskConfiguration { + + @Autowired + private TaskProperties taskProperties; + + @Autowired + private SimpleTaskNameResolver taskNameResolver; + + @Autowired + private ApplicationEventPublisher applicationEventPublisher; + + @Autowired + private TaskConfigurer taskConfigurer; + + + @Bean + public SingleInstanceTaskListener taskListener() { + if (taskConfigurer.getTaskDataSource() == null) { + return new SingleInstanceTaskListener(new PassThruLockRegistry(), + this.taskNameResolver, this.taskProperties, this.applicationEventPublisher); + } + + return new SingleInstanceTaskListener(taskConfigurer.getTaskDataSource(), + this.taskNameResolver, + this.taskProperties, + this.applicationEventPublisher); + } +} diff --git a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/TaskProperties.java b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/TaskProperties.java index 5f2a87e2..37c116e3 100644 --- a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/TaskProperties.java +++ b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/TaskProperties.java @@ -62,6 +62,28 @@ public class TaskProperties { */ private Boolean closecontextEnabled = false; + /** + * When set to true it + * will check to see if a task execution with the same task name is already + * running. If a task is still running then it will throw a + * {@link org.springframework.cloud.task.listener.TaskExecutionException}. + * When task execution ends the lock is released. + */ + private boolean singleInstanceEnabled = false; + + /** + * Declares the maximum amount of time (in millis) that a task execution can + * hold a lock to prevent another task from executing with a specific task + * name when the singleInstanceEnabled is set to true. Default time is: Integer.MAX_VALUE. + */ + private int singleInstanceLockTtl = Integer.MAX_VALUE; + + /** + * Declares the time (in millis) that a task execution will wait between + * checks. Default time is: 500 millis. + */ + private int singleInstanceLockCheckInterval = 500; + public String getExternalExecutionId() { return externalExecutionId; } @@ -123,4 +145,28 @@ public class TaskProperties { public void setParentExecutionId(Long parentExecutionId) { this.parentExecutionId = parentExecutionId; } + + public boolean getSingleInstanceEnabled() { + return singleInstanceEnabled; + } + + public void setSingleInstanceEnabled(boolean singleInstanceEnabled) { + this.singleInstanceEnabled = singleInstanceEnabled; + } + + public int getSingleInstanceLockTtl() { + return singleInstanceLockTtl; + } + + public void setSingleInstanceLockTtl(int singleInstanceLockTtl) { + this.singleInstanceLockTtl = singleInstanceLockTtl; + } + + public int getSingleInstanceLockCheckInterval() { + return singleInstanceLockCheckInterval; + } + + public void setSingleInstanceLockCheckInterval(int singleInstanceLockCheckInterval) { + this.singleInstanceLockCheckInterval = singleInstanceLockCheckInterval; + } } diff --git a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/listener/TaskLifecycleListener.java b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/listener/TaskLifecycleListener.java index db7971ad..9a426ed3 100644 --- a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/listener/TaskLifecycleListener.java +++ b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/listener/TaskLifecycleListener.java @@ -260,7 +260,9 @@ public class TaskLifecycleListener implements ApplicationListener starterList = new ArrayList<>(taskExecutionListeners); + Collections.reverse(starterList); + for (TaskExecutionListener taskExecutionListener : starterList) { taskExecutionListener.onTaskStartup(listenerTaskExecution); } } diff --git a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-db2.sql b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-db2.sql index 7e7834e5..e4cb2fd7 100644 --- a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-db2.sql +++ b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-db2.sql @@ -26,4 +26,12 @@ CREATE TABLE TASK_TASK_BATCH ( references TASK_EXECUTION(TASK_EXECUTION_ID) ) ; -CREATE SEQUENCE TASK_SEQ AS BIGINT START WITH 0 MINVALUE 0 MAXVALUE 9223372036854775807 NOCACHE NOCYCLE; \ No newline at end of file +CREATE SEQUENCE TASK_SEQ AS BIGINT START WITH 0 MINVALUE 0 MAXVALUE 9223372036854775807 NOCACHE NOCYCLE; + +CREATE TABLE TASK_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-h2.sql b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-h2.sql index eb633d9d..d2fad98d 100644 --- a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-h2.sql +++ b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-h2.sql @@ -27,3 +27,11 @@ CREATE TABLE TASK_TASK_BATCH ( ) ; CREATE SEQUENCE TASK_SEQ ; + +CREATE TABLE TASK_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-hsqldb.sql b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-hsqldb.sql index 4ffa6734..554f8faa 100644 --- a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-hsqldb.sql +++ b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-hsqldb.sql @@ -29,3 +29,11 @@ CREATE TABLE TASK_TASK_BATCH ( CREATE TABLE TASK_SEQ ( ID BIGINT IDENTITY ); + +CREATE TABLE TASK_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-mysql.sql b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-mysql.sql index 00599fa7..f2734d7e 100644 --- a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-mysql.sql +++ b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-mysql.sql @@ -33,3 +33,11 @@ CREATE TABLE TASK_SEQ ( ) ENGINE=InnoDB; INSERT INTO TASK_SEQ (ID, UNIQUE_KEY) select * from (select 0 as ID, '0' as UNIQUE_KEY) as tmp; + +CREATE TABLE TASK_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE DATETIME(6) NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +) ENGINE=InnoDB; diff --git a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-oracle10g.sql b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-oracle10g.sql index b9aea638..91a72903 100644 --- a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-oracle10g.sql +++ b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-oracle10g.sql @@ -26,4 +26,12 @@ CREATE TABLE TASK_TASK_BATCH ( references TASK_EXECUTION(TASK_EXECUTION_ID) ) ; -CREATE SEQUENCE TASK_SEQ START WITH 0 MINVALUE 0 MAXVALUE 9223372036854775807 NOCACHE NOCYCLE; \ No newline at end of file +CREATE SEQUENCE TASK_SEQ START WITH 0 MINVALUE 0 MAXVALUE 9223372036854775807 NOCACHE NOCYCLE; + +CREATE TABLE TASK_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR2(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-postgresql.sql b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-postgresql.sql index 6d558b4f..fde999d4 100644 --- a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-postgresql.sql +++ b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-postgresql.sql @@ -26,4 +26,12 @@ CREATE TABLE TASK_TASK_BATCH ( references TASK_EXECUTION(TASK_EXECUTION_ID) ) ; -CREATE SEQUENCE TASK_SEQ MAXVALUE 9223372036854775807 NO CYCLE; \ No newline at end of file +CREATE SEQUENCE TASK_SEQ MAXVALUE 9223372036854775807 NO CYCLE; + +CREATE TABLE TASK_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE TIMESTAMP NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-sqlserver.sql b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-sqlserver.sql index 3985cd96..e055c203 100644 --- a/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-sqlserver.sql +++ b/spring-cloud-task-core/src/main/resources/org/springframework/cloud/task/schema-sqlserver.sql @@ -26,3 +26,11 @@ CREATE TABLE TASK_TASK_BATCH ( ) ; CREATE TABLE TASK_SEQ (ID BIGINT IDENTITY); + +CREATE TABLE TASK_LOCK ( + LOCK_KEY CHAR(36), + REGION VARCHAR(100), + CLIENT_ID CHAR(36), + CREATED_DATE DATETIME NOT NULL, + constraint LOCK_PK primary key (LOCK_KEY, REGION) +); diff --git a/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/SimpleSingleTaskAutoConfigurationTests.java b/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/SimpleSingleTaskAutoConfigurationTests.java new file mode 100644 index 00000000..5b73a8ce --- /dev/null +++ b/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/SimpleSingleTaskAutoConfigurationTests.java @@ -0,0 +1,62 @@ +/* + * Copyright 2017 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; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.task.configuration.SingleTaskConfiguration; +import org.springframework.cloud.task.configuration.SimpleTaskConfiguration; +import org.springframework.cloud.task.configuration.SingleInstanceTaskListener; +import org.springframework.cloud.task.configuration.TaskProperties; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +/** + * Verifies that the beans created by the SimpleSingleTaskAutoConfigurationConfiguration + * specifically that PassThruRegistry was selected. + * + * @author Glenn Renfro + * @since 2.0.0 + */ +@RunWith(SpringRunner.class) +@ContextConfiguration(classes = {TaskProperties.class, SimpleTaskConfiguration.class, SingleTaskConfiguration.class}) +@TestPropertySource(properties = { + "spring.cloud.task.singleInstanceEnabled=true", +}) +public class SimpleSingleTaskAutoConfigurationTests { + @Autowired + private ConfigurableApplicationContext context; + + @Test + public void testConfiguration() throws Exception { + + SingleInstanceTaskListener singleInstanceTaskListener = this.context.getBean(SingleInstanceTaskListener.class); + + assertNotNull("singleInstanceTaskListener should not be null", singleInstanceTaskListener); + + assertEquals(singleInstanceTaskListener.getClass(), SingleInstanceTaskListener.class); + + } + +} diff --git a/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/SimpleSingleTaskAutoConfigurationWithDataSourceTests.java b/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/SimpleSingleTaskAutoConfigurationWithDataSourceTests.java new file mode 100644 index 00000000..f40fe9f1 --- /dev/null +++ b/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/SimpleSingleTaskAutoConfigurationWithDataSourceTests.java @@ -0,0 +1,63 @@ +/* + * Copyright 2017 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; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.jdbc.EmbeddedDataSourceConfiguration; +import org.springframework.cloud.task.configuration.SingleTaskConfiguration; +import org.springframework.cloud.task.configuration.SimpleTaskConfiguration; +import org.springframework.cloud.task.configuration.SingleInstanceTaskListener; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +/** + * Verifies that the beans created by the SimpleSingleTaskAutoConfigurationConfiguration + * specifically that the JdbcLockRegistry was selected. + * + * @author Glenn Renfro + * @since 2.0.0 + */ +@RunWith(SpringRunner.class) +@ContextConfiguration(classes = {SimpleTaskConfiguration.class, + SingleTaskConfiguration.class, + EmbeddedDataSourceConfiguration.class}) +@TestPropertySource(properties = { + "spring.cloud.task.singleInstanceEnabled=true", +}) +public class SimpleSingleTaskAutoConfigurationWithDataSourceTests { + + @Autowired + private ConfigurableApplicationContext context; + + @Test + public void testConfiguration() throws Exception { + + SingleInstanceTaskListener singleInstanceTaskListener = this.context.getBean(SingleInstanceTaskListener.class); + + assertNotNull("singleInstanceTaskListener should not be null", singleInstanceTaskListener); + + assertEquals(singleInstanceTaskListener.getClass(), SingleInstanceTaskListener.class); + } +} diff --git a/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/listener/TaskLifecycleListenerTests.java b/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/listener/TaskLifecycleListenerTests.java index f208a267..1c7dae95 100644 --- a/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/listener/TaskLifecycleListenerTests.java +++ b/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/listener/TaskLifecycleListenerTests.java @@ -16,6 +16,7 @@ package org.springframework.cloud.task.listener; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -76,6 +77,10 @@ public class TaskLifecycleListenerTests { context = new AnnotationConfigApplicationContext(); context.setId("testTask"); context.register(TestDefaultConfiguration.class, PropertyPlaceholderAutoConfiguration.class); + TestListener.getStartupOrderList().clear(); + TestListener.getFailOrderList().clear(); + TestListener.getEndOrderList().clear(); + } @After @@ -125,6 +130,9 @@ public class TaskLifecycleListenerTests { @Test public void testTaskFailedWithExitCodeEvent() { final int exitCode = 10; + context.register(TestListener.class); + context.register(TestListener2.class); + context.refresh(); RuntimeException exception = new RuntimeException("This was expected"); SpringApplication application = new SpringApplication(); @@ -134,6 +142,18 @@ public class TaskLifecycleListenerTests { context.publishEvent(new ApplicationReadyEvent(application, new String[0], context)); verifyTaskExecution(0, true, exitCode, exception, null); + assertEquals(2, TestListener.getStartupOrderList().size()); + assertEquals(Integer.valueOf(2), TestListener.getStartupOrderList().get(0)); + assertEquals(Integer.valueOf(1), TestListener.getStartupOrderList().get(1)); + + assertEquals(2, TestListener.getEndOrderList().size()); + assertEquals(Integer.valueOf(1), TestListener.getEndOrderList().get(0)); + assertEquals(Integer.valueOf(2), TestListener.getEndOrderList().get(1)); + + assertEquals(2, TestListener.getFailOrderList().size()); + assertEquals(Integer.valueOf(1), TestListener.getFailOrderList().get(0)); + assertEquals(Integer.valueOf(2), TestListener.getFailOrderList().get(1)); + } @Test @@ -300,4 +320,53 @@ public class TaskLifecycleListenerTests { throw new UnsupportedOperationException("Not supported at this time."); } } + + private static class TestListener2 extends TestListener { + + } + + private static class TestListener implements TaskExecutionListener { + + private static int currentCount = 0; + + private int id = 0; + + static List startupOrderList = new ArrayList<>(); + + static List endOrderList = new ArrayList<>(); + + static List failOrderList = new ArrayList<>(); + + public TestListener() { + currentCount++; + id = currentCount; + } + + @Override + public void onTaskStartup(TaskExecution taskExecution) { + startupOrderList.add(id); + } + + @Override + public void onTaskEnd(TaskExecution taskExecution) { + endOrderList.add(id); + } + + @Override + public void onTaskFailed(TaskExecution taskExecution, Throwable throwable) { + failOrderList.add(id); + } + + public static List getStartupOrderList() { + return startupOrderList; + } + + public static List getEndOrderList() { + return endOrderList; + } + + public static List getFailOrderList() { + return failOrderList; + } + } } diff --git a/spring-cloud-task-docs/src/main/asciidoc/features.adoc b/spring-cloud-task-docs/src/main/asciidoc/features.adoc index 087249d1..464d0cd4 100644 --- a/spring-cloud-task-docs/src/main/asciidoc/features.adoc +++ b/spring-cloud-task-docs/src/main/asciidoc/features.adoc @@ -344,3 +344,30 @@ will be stored in the repo, else if a failure occurs then the `exitMessage` from the `onTaskFailed` will be stored. Also if a user sets the `exitMessage` with a `onTaskEnd` listener then the `exitMessage` from the `onTaskEnd` will supersede the exit messages from both the `onTaskStartup` and `onTaskFailed`. + +=== Restricting Spring Cloud Task Instances + +Allows a user to establish that only one task with a given task name can be run +at a time. To do this the user establishes the <> +and sets `spring.cloud.task.singleInstanceEnabled=true` for each task execution. +While the first task execution is running, any other time user tries to run +a task with the same <> and +`spring.cloud.task.singleInstanceEnabled=true` the task will fail with the following +error message `Task with name "application" is already running.` The default +for `spring.cloud.task.singleInstanceEnabled` is `false`. +``` + spring.cloud.task.singleInstanceEnabled= +``` +In order for this feature to be used you must include the following Spring +Integration dependencies to your application: +[source,xml] +--- + + org.springframework.integration + spring-integration-core + + + org.springframework.integration + spring-integration-jdbc + +--- \ No newline at end of file diff --git a/spring-cloud-task-integration-tests/pom.xml b/spring-cloud-task-integration-tests/pom.xml index 4841cde5..14d509e4 100644 --- a/spring-cloud-task-integration-tests/pom.xml +++ b/spring-cloud-task-integration-tests/pom.xml @@ -84,6 +84,16 @@ h2 test + + org.springframework.integration + spring-integration-core + test + + + org.springframework.integration + spring-integration-jdbc + test + diff --git a/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/executionid/TaskStartTests.java b/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/executionid/TaskStartTests.java index 30214b37..ea0e7ef6 100644 --- a/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/executionid/TaskStartTests.java +++ b/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/executionid/TaskStartTests.java @@ -19,9 +19,12 @@ package org.springframework.cloud.task.executionid; import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.ArrayList; import java.util.Date; import java.util.HashMap; import java.util.Map; +import java.util.UUID; + import javax.sql.DataSource; import org.h2.tools.Server; @@ -51,6 +54,7 @@ import org.springframework.core.io.ClassPathResource; import org.springframework.data.domain.Page; import org.springframework.data.domain.PageRequest; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.simple.SimpleJdbcInsert; import org.springframework.jdbc.datasource.DriverManagerDataSource; import org.springframework.jdbc.datasource.init.DataSourceInitializer; import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator; @@ -122,6 +126,7 @@ public class TaskStartTests { template.execute("DROP TABLE IF EXISTS TASK_SEQ"); template.execute("DROP TABLE IF EXISTS TASK_EXECUTION_PARAMS"); template.execute("DROP TABLE IF EXISTS TASK_EXECUTION"); + template.execute("DROP TABLE IF EXISTS TASK_LOCK"); template.execute("DROP TABLE IF EXISTS BATCH_STEP_EXECUTION_SEQ"); template.execute("DROP TABLE IF EXISTS BATCH_STEP_EXECUTION_CONTEXT"); template.execute("DROP TABLE IF EXISTS BATCH_STEP_EXECUTION"); @@ -184,6 +189,43 @@ public class TaskStartTests { taskRepository.completeTaskExecution(1, 0, new Date(),""); this.applicationContext = getTaskApplication(1).run(new String[0]); } + + @Test + public void testDuplicateTaskExecutionWithSingleInstanceEnabled() throws Exception { + String params[] = {"--spring.cloud.task.singleInstanceEnabled=true", + "--spring.cloud.task.name=foo"}; + boolean testFailed = false; + try { + taskRepository.createTaskExecution(); + assertEquals("Only one row is expected", 1, taskExplorer.getTaskExecutionCount()); + enableLock("foo"); + getTaskApplication(1).run(params); + } + catch (ApplicationContextException taskException) { + assertEquals("Failed to start bean 'taskLifecycleListener'; nested " + + "exception is org.springframework.cloud.task." + + "listener.TaskExecutionException: Failed to process " + + "@BeforeTask or @AfterTask annotation because: Task with name \"foo\" is already running.", + taskException.getMessage()); + testFailed = true; + } + assertTrue("Expected TaskExecutionException for because of " + + "singleInstanceEnabled is enabled", testFailed); + + } + + @Test + public void testDuplicateTaskExecutionWithSingleInstanceDisabled() throws Exception { + taskRepository.createTaskExecution(); + TaskExecution execution = taskRepository.createTaskExecution(); + taskRepository.startTaskExecution(execution.getExecutionId(), "bar", + new Date(), new ArrayList<>(), ""); + String params[] = {"--spring.cloud.task.name=bar"}; + enableLock("bar"); + this.applicationContext = getTaskApplication(1).run(params); + assertTrue(waitForDBToBePopulated()); + } + private SpringApplication getTaskApplication(Integer executionId) { SpringApplication myapp = new SpringApplication(TaskStartApplication.class); Map myMap = new HashMap<>(); @@ -217,6 +259,16 @@ public class TaskStartTests { return isDbPopulated; } + private void enableLock(String lockKey) { + SimpleJdbcInsert taskLockInsert = new SimpleJdbcInsert(this.dataSource).withTableName("TASK_LOCK"); + Map taskLockParams = new HashMap<>(); + taskLockParams.put("LOCK_KEY", UUID.nameUUIDFromBytes(lockKey.getBytes()).toString()); + taskLockParams.put("REGION", "DEFAULT"); + taskLockParams.put("CLIENT_ID", "aClientID"); + taskLockParams.put("CREATED_DATE", new Date()); + taskLockInsert.execute(taskLockParams); + } + @Configuration public static class TaskLauncherConfiguration { private static Server defaultServer;