natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

remove taskInstancnCacheManager (#7636) 

Co-authored-by: caishunfeng <534328519@qq.com>
This commit is contained in:
wind 2021-12-27 10:55:09 +08:00 committed by GitHub
parent 9e068d23e2
commit 744577f252
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 1 additions and 437 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/cache/TaskInstanceCacheManager.java vendored
View File

@ -1,64 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.apache.dolphinscheduler.server.master.cache;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
/**
* task instance state manager
*/
public interface TaskInstanceCacheManager {
/**
* get taskInstance by taskInstance id
*
* @param taskInstanceId taskInstanceId
* @return taskInstance
*/
TaskInstance getByTaskInstanceId(Integer taskInstanceId);
/**
* cache taskInstance
*
* @param taskExecutionContext taskExecutionContext
*/
void cacheTaskInstance(TaskExecutionContext taskExecutionContext);
/**
* cache taskInstance
*
* @param taskAckCommand taskAckCommand
*/
void cacheTaskInstance(TaskExecuteAckCommand taskAckCommand);
/**
* cache taskInstance
*
* @param taskExecuteResponseCommand taskExecuteResponseCommand
*/
void cacheTaskInstance(TaskExecuteResponseCommand taskExecuteResponseCommand);
/**
* remove taskInstance by taskInstanceId
* @param taskInstanceId taskInstanceId
*/
void removeByTaskInstanceId(Integer taskInstanceId);
}

155
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/cache/impl/TaskInstanceCacheManagerImpl.java vendored
View File

@ -1,155 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.apache.dolphinscheduler.server.master.cache.impl;
import static org.apache.dolphinscheduler.common.Constants.CACHE_REFRESH_TIME_MILLIS;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.server.master.cache.TaskInstanceCacheManager;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* taskInstance state manager
*/
@Component
public class TaskInstanceCacheManagerImpl implements TaskInstanceCacheManager {
/**
* taskInstance cache
*/
private Map<Integer,TaskInstance> taskInstanceCache = new ConcurrentHashMap<>();
/**
* process service
*/
@Autowired
private ProcessService processService;
/**
* taskInstance cache refresh timer
*/
private Timer refreshTaskInstanceTimer = null;
@PostConstruct
public void init() {
//issue#5539 add thread to fetch task state from database in a fixed rate
this.refreshTaskInstanceTimer = new Timer(true);
refreshTaskInstanceTimer.scheduleAtFixedRate(
new RefreshTaskInstanceTimerTask(), CACHE_REFRESH_TIME_MILLIS, CACHE_REFRESH_TIME_MILLIS
);
}
@PreDestroy
public void close() {
this.refreshTaskInstanceTimer.cancel();
}
/**
* get taskInstance by taskInstance id
*
* @param taskInstanceId taskInstanceId
* @return taskInstance
*/
@Override
public TaskInstance getByTaskInstanceId(Integer taskInstanceId) {
return taskInstanceCache.computeIfAbsent(taskInstanceId, k -> processService.findTaskInstanceById(taskInstanceId));
}
/**
* cache taskInstance
*
* @param taskExecutionContext taskExecutionContext
*/
@Override
public void cacheTaskInstance(TaskExecutionContext taskExecutionContext) {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(taskExecutionContext.getTaskInstanceId());
taskInstance.setName(taskExecutionContext.getTaskName());
taskInstance.setStartTime(taskExecutionContext.getStartTime());
taskInstance.setTaskType(taskExecutionContext.getTaskType());
taskInstance.setExecutePath(taskExecutionContext.getExecutePath());
taskInstanceCache.put(taskExecutionContext.getTaskInstanceId(), taskInstance);
}
/**
* cache taskInstance
*
* @param taskAckCommand taskAckCommand
*/
@Override
public void cacheTaskInstance(TaskExecuteAckCommand taskAckCommand) {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setState(ExecutionStatus.of(taskAckCommand.getStatus()));
taskInstance.setStartTime(taskAckCommand.getStartTime());
taskInstance.setHost(taskAckCommand.getHost());
taskInstance.setExecutePath(taskAckCommand.getExecutePath());
taskInstance.setLogPath(taskAckCommand.getLogPath());
taskInstanceCache.put(taskAckCommand.getTaskInstanceId(), taskInstance);
}
/**
* cache taskInstance
*
* @param taskExecuteResponseCommand taskExecuteResponseCommand
*/
@Override
public void cacheTaskInstance(TaskExecuteResponseCommand taskExecuteResponseCommand) {
TaskInstance taskInstance = getByTaskInstanceId(taskExecuteResponseCommand.getTaskInstanceId());
taskInstance.setState(ExecutionStatus.of(taskExecuteResponseCommand.getStatus()));
taskInstance.setEndTime(taskExecuteResponseCommand.getEndTime());
taskInstanceCache.put(taskExecuteResponseCommand.getTaskInstanceId(), taskInstance);
}
/**
* remove taskInstance by taskInstanceId
* @param taskInstanceId taskInstanceId
*/
@Override
public void removeByTaskInstanceId(Integer taskInstanceId) {
taskInstanceCache.remove(taskInstanceId);
}
class RefreshTaskInstanceTimerTask extends TimerTask {
@Override
public void run() {
for (Entry<Integer, TaskInstance> taskInstanceEntry : taskInstanceCache.entrySet()) {
TaskInstance taskInstance = processService.findTaskInstanceById(taskInstanceEntry.getKey());
if (null != taskInstance && taskInstance.getState() == ExecutionStatus.NEED_FAULT_TOLERANCE) {
taskInstanceCache.computeIfPresent(taskInstanceEntry.getKey(), (k, v) -> taskInstance);
}
}
}
}
}

10
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessor.java
View File

@ -24,8 +24,6 @@ import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.remote.utils.ChannelUtils;
import org.apache.dolphinscheduler.server.master.cache.TaskInstanceCacheManager;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThread;
@ -52,14 +50,8 @@ public class TaskAckProcessor implements NettyRequestProcessor {
*/
private final TaskResponseService taskResponseService;
/**
* taskInstance cache manager
*/
private final TaskInstanceCacheManager taskInstanceCacheManager;
public TaskAckProcessor() {
this.taskResponseService = SpringApplicationContext.getBean(TaskResponseService.class);
this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
}
public void init(ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceExecMaps) {
@ -78,8 +70,6 @@ public class TaskAckProcessor implements NettyRequestProcessor {
TaskExecuteAckCommand taskAckCommand = JSONUtils.parseObject(command.getBody(), TaskExecuteAckCommand.class);
logger.info("taskAckCommand : {}", taskAckCommand);
taskInstanceCacheManager.cacheTaskInstance(taskAckCommand);
String workerAddress = ChannelUtils.toAddress(channel).getAddress();
ExecutionStatus ackStatus = ExecutionStatus.of(taskAckCommand.getStatus());

10
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java
View File

@ -23,8 +23,6 @@ import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.server.master.cache.TaskInstanceCacheManager;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThread;
@ -51,14 +49,8 @@ public class TaskResponseProcessor implements NettyRequestProcessor {
*/
private final TaskResponseService taskResponseService;
/**
* taskInstance cache manager
*/
private final TaskInstanceCacheManager taskInstanceCacheManager;
public TaskResponseProcessor() {
this.taskResponseService = SpringApplicationContext.getBean(TaskResponseService.class);
this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
}
public void init(ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceExecMaps) {
@ -79,8 +71,6 @@ public class TaskResponseProcessor implements NettyRequestProcessor {
TaskExecuteResponseCommand responseCommand = JSONUtils.parseObject(command.getBody(), TaskExecuteResponseCommand.class);
logger.info("received command : {}", responseCommand);
taskInstanceCacheManager.cacheTaskInstance(responseCommand);
// TaskResponseEvent
TaskResponseEvent taskResponseEvent = TaskResponseEvent.newResult(ExecutionStatus.of(responseCommand.getStatus()),
responseCommand.getEndTime(),

177
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/cache/impl/TaskInstanceCacheManagerImplTest.java vendored
View File

@ -1,177 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.apache.dolphinscheduler.server.master.cache.impl;
import static org.apache.dolphinscheduler.common.Constants.CACHE_REFRESH_TIME_MILLIS;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.TaskType;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
import java.util.Calendar;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class TaskInstanceCacheManagerImplTest {
@InjectMocks
private TaskInstanceCacheManagerImpl taskInstanceCacheManager;
@Mock(name = "processService")
private ProcessService processService;
@Before
public void before() {
TaskExecuteAckCommand taskExecuteAckCommand = new TaskExecuteAckCommand();
taskExecuteAckCommand.setStatus(1);
taskExecuteAckCommand.setExecutePath("/dolphinscheduler/worker");
taskExecuteAckCommand.setHost("worker007");
taskExecuteAckCommand.setLogPath("/temp/worker.log");
taskExecuteAckCommand.setStartTime(new Date(1970, Calendar.AUGUST,7));
taskExecuteAckCommand.setTaskInstanceId(0);
taskInstanceCacheManager.cacheTaskInstance(taskExecuteAckCommand);
}
@Test
public void testInit() throws InterruptedException {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(0);
taskInstance.setState(ExecutionStatus.NEED_FAULT_TOLERANCE);
taskInstance.setExecutePath("/dolphinscheduler/worker");
taskInstance.setHost("worker007");
taskInstance.setLogPath("/temp/worker.log");
taskInstance.setProcessInstanceId(0);
Mockito.when(processService.findTaskInstanceById(0)).thenReturn(taskInstance);
taskInstanceCacheManager.init();
TimeUnit.MILLISECONDS.sleep(CACHE_REFRESH_TIME_MILLIS + 1000);
Assert.assertEquals(taskInstance.getState(), taskInstanceCacheManager.getByTaskInstanceId(0).getState());
}
@Test
public void getByTaskInstanceIdFromCache() {
TaskInstance instanceGot = taskInstanceCacheManager.getByTaskInstanceId(0);
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(0);
taskInstance.setState(ExecutionStatus.RUNNING_EXECUTION);
taskInstance.setExecutePath("/dolphinscheduler/worker");
taskInstance.setHost("worker007");
taskInstance.setLogPath("/temp/worker.log");
taskInstance.setStartTime(new Date(1970, Calendar.AUGUST,7));
Assert.assertEquals(taskInstance.toString(), instanceGot.toString());
}
@Test
public void getByTaskInstanceIdFromDatabase() {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setId(1);
taskInstance.setState(ExecutionStatus.RUNNING_EXECUTION);
taskInstance.setExecutePath("/dolphinscheduler/worker");
taskInstance.setHost("worker007");
taskInstance.setLogPath("/temp/worker.log");
taskInstance.setStartTime(new Date(1970, Calendar.AUGUST,7));
Mockito.when(processService.findTaskInstanceById(1)).thenReturn(taskInstance);
TaskInstance instanceGot = taskInstanceCacheManager.getByTaskInstanceId(1);
Assert.assertEquals(taskInstance, instanceGot);
}
@Test
public void cacheTaskInstanceByTaskExecutionContext() {
TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
taskExecutionContext.setTaskInstanceId(2);
taskExecutionContext.setTaskName("blackberrier test");
taskExecutionContext.setStartTime(new Date(1970, Calendar.AUGUST,7));
taskExecutionContext.setTaskType(TaskType.SPARK.getDesc());
taskExecutionContext.setExecutePath("/tmp");
taskInstanceCacheManager.cacheTaskInstance(taskExecutionContext);
TaskInstance taskInstance = taskInstanceCacheManager.getByTaskInstanceId(2);
Assert.assertEquals(taskInstance.getId(), 2);
Assert.assertEquals(taskInstance.getName(), "blackberrier test");
Assert.assertEquals(taskInstance.getStartTime(), new Date(1970, Calendar.AUGUST, 7));
Assert.assertEquals(taskInstance.getTaskType(), TaskType.SPARK.getDesc());
Assert.assertEquals(taskInstance.getExecutePath(), "/tmp");
}
@Test
public void testCacheTaskInstanceByTaskExecuteAckCommand() {
TaskInstance taskInstance = taskInstanceCacheManager.getByTaskInstanceId(0);
Assert.assertEquals(ExecutionStatus.RUNNING_EXECUTION, taskInstance.getState());
Assert.assertEquals(new Date(1970, Calendar.AUGUST, 7), taskInstance.getStartTime());
Assert.assertEquals("worker007", taskInstance.getHost());
Assert.assertEquals("/dolphinscheduler/worker", taskInstance.getExecutePath());
Assert.assertEquals("/temp/worker.log", taskInstance.getLogPath());
}
@Test
public void testCacheTaskInstanceByTaskExecuteResponseCommand() {
TaskExecuteResponseCommand responseCommand = new TaskExecuteResponseCommand();
responseCommand.setTaskInstanceId(0);
responseCommand.setStatus(9);
responseCommand.setEndTime(new Date(1970, Calendar.AUGUST, 8));
taskInstanceCacheManager.cacheTaskInstance(responseCommand);
TaskInstance taskInstance = taskInstanceCacheManager.getByTaskInstanceId(0);
Assert.assertEquals(new Date(1970, Calendar.AUGUST, 8), taskInstance.getEndTime());
Assert.assertEquals(ExecutionStatus.KILL, taskInstance.getState());
}
@Test
public void removeByTaskInstanceId() {
taskInstanceCacheManager.removeByTaskInstanceId(0);
Assert.assertNull(taskInstanceCacheManager.getByTaskInstanceId(0));
}
}

10
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessorTest.java
View File

@ -18,20 +18,16 @@
package org.apache.dolphinscheduler.server.master.processor;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
import java.net.InetSocketAddress;
import java.util.Date;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
@ -39,7 +35,7 @@ import org.powermock.modules.junit4.PowerMockRunner;
import io.netty.channel.Channel;
/**
* task ack processor test
* task ack processor test
*/
@RunWith(PowerMockRunner.class)
@PrepareForTest({SpringApplicationContext.class, TaskResponseEvent.class})
@ -47,7 +43,6 @@ public class TaskAckProcessorTest {
private TaskAckProcessor taskAckProcessor;
private TaskResponseService taskResponseService;
private TaskInstanceCacheManagerImpl taskInstanceCacheManager;
private ProcessService processService;
private TaskExecuteAckCommand taskExecuteAckCommand;
private TaskResponseEvent taskResponseEvent;
@ -60,9 +55,6 @@ public class TaskAckProcessorTest {
taskResponseService = PowerMockito.mock(TaskResponseService.class);
PowerMockito.when(SpringApplicationContext.getBean(TaskResponseService.class)).thenReturn(taskResponseService);
taskInstanceCacheManager = PowerMockito.mock(TaskInstanceCacheManagerImpl.class);
PowerMockito.when(SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class)).thenReturn(taskInstanceCacheManager);
processService = PowerMockito.mock(ProcessService.class);
PowerMockito.when(SpringApplicationContext.getBean(ProcessService.class)).thenReturn(processService);

6
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/registry/DependencyConfig.java
View File

@ -36,7 +36,6 @@ import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UdfFuncMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.dispatch.host.HostManager;
import org.apache.dolphinscheduler.server.master.dispatch.host.RandomHostManager;
@ -67,11 +66,6 @@ public class DependencyConfig {
return Mockito.mock(AlertMapper.class);
}
@Bean
public TaskInstanceCacheManagerImpl taskInstanceCacheManagerImpl() {
return Mockito.mock(TaskInstanceCacheManagerImpl.class);
}
@Bean
public ProcessService processService() {
return Mockito.mock(ProcessService.class);

6
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackServiceTestConfig.java
View File

@ -36,7 +36,6 @@ import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UdfFuncMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.mockito.Mockito;
@ -59,11 +58,6 @@ public class TaskCallbackServiceTestConfig {
return Mockito.mock(AlertMapper.class);
}
@Bean
public TaskInstanceCacheManagerImpl taskInstanceCacheManagerImpl() {
return Mockito.mock(TaskInstanceCacheManagerImpl.class);
}
@Bean
public ProcessService processService() {
return Mockito.mock(ProcessService.class);