Prevent a task from starting if an inst with the same name is running.

resolves #81

Using LockRegistryLeaderInitiator to do leadership election.

When task is started and singleInstanceEnabled isset to true then we use leader election
to determine if a task needs to be started.

Error Event Name had to be updated
This commit is contained in:
Glenn Renfro
2017-09-27 17:39:29 -04:00
committed by Michael Minella
parent 94e074d841
commit d8a73ba183
19 changed files with 628 additions and 5 deletions

View File

@@ -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 {
}

View File

@@ -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<ApplicationEvent> {
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);
}
}

View File

@@ -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);
}
}

View File

@@ -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;
}
}

View File

@@ -260,7 +260,9 @@ public class TaskLifecycleListener implements ApplicationListener<ApplicationEve
TaskExecution listenerTaskExecution = getTaskExecutionCopy(taskExecution);
if (this.taskExecutionListeners != null) {
try {
for (TaskExecutionListener taskExecutionListener : this.taskExecutionListeners) {
List<TaskExecutionListener> starterList = new ArrayList<>(taskExecutionListeners);
Collections.reverse(starterList);
for (TaskExecutionListener taskExecutionListener : starterList) {
taskExecutionListener.onTaskStartup(listenerTaskExecution);
}
}

View File

@@ -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;
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)
);

View File

@@ -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)
);

View File

@@ -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)
);

View File

@@ -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;

View File

@@ -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;
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)
);

View File

@@ -26,4 +26,12 @@ CREATE TABLE TASK_TASK_BATCH (
references TASK_EXECUTION(TASK_EXECUTION_ID)
) ;
CREATE SEQUENCE TASK_SEQ MAXVALUE 9223372036854775807 NO CYCLE;
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)
);

View File

@@ -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)
);

View File

@@ -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);
}
}

View File

@@ -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);
}
}

View File

@@ -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<Integer> startupOrderList = new ArrayList<>();
static List<Integer> endOrderList = new ArrayList<>();
static List<Integer> 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<Integer> getStartupOrderList() {
return startupOrderList;
}
public static List<Integer> getEndOrderList() {
return endOrderList;
}
public static List<Integer> getFailOrderList() {
return failOrderList;
}
}
}