[Improvement-6540][server] Optimize the code related to the processInstanceExecMaps object

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/MasterServer.java

@@ -30,12 +30,10 @@
 import org.apache.dolphinscheduler.server.master.processor.TaskResponseProcessor;
 import org.apache.dolphinscheduler.server.master.registry.MasterRegistryClient;
 import org.apache.dolphinscheduler.server.master.runner.EventExecuteService;
-import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThread;
 import org.apache.dolphinscheduler.server.master.runner.MasterSchedulerService;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 import org.apache.dolphinscheduler.service.quartz.QuartzExecutors;
 
-import java.util.concurrent.ConcurrentHashMap;
 
 import javax.annotation.PostConstruct;
 
@@ -100,7 +98,6 @@ public class MasterServer implements IStoppable {
  @Autowired
  private EventExecuteService eventExecuteService;
 
- private ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceExecMaps = new ConcurrentHashMap<>();
 
  /**
  * master server startup, not use web service
@@ -121,28 +118,18 @@ public void run() {
  NettyServerConfig serverConfig = new NettyServerConfig();
  serverConfig.setListenPort(masterConfig.getListenPort());
  this.nettyRemotingServer = new NettyRemotingServer(serverConfig);
- TaskAckProcessor ackProcessor = new TaskAckProcessor();
- ackProcessor.init(processInstanceExecMaps);
- TaskResponseProcessor taskResponseProcessor = new TaskResponseProcessor();
- taskResponseProcessor.init(processInstanceExecMaps);
- StateEventProcessor stateEventProcessor = new StateEventProcessor();
- stateEventProcessor.init(processInstanceExecMaps);
- this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RESPONSE, taskResponseProcessor);
- this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_ACK, ackProcessor);
+ this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RESPONSE, new TaskResponseProcessor());
+ this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_ACK, new TaskAckProcessor());
  this.nettyRemotingServer.registerProcessor(CommandType.TASK_KILL_RESPONSE, new TaskKillResponseProcessor());
- this.nettyRemotingServer.registerProcessor(CommandType.STATE_EVENT_REQUEST, stateEventProcessor);
+ this.nettyRemotingServer.registerProcessor(CommandType.STATE_EVENT_REQUEST, new StateEventProcessor());
  this.nettyRemotingServer.start();
 
  // self tolerant
- this.masterRegistryClient.init(this.processInstanceExecMaps);
  this.masterRegistryClient.start();
  this.masterRegistryClient.setRegistryStoppable(this);
 
- this.eventExecuteService.init(this.processInstanceExecMaps);
  this.eventExecuteService.start();
  // scheduler start
- this.masterSchedulerService.init(this.processInstanceExecMaps);
-
  this.masterSchedulerService.start();
 
  // start QuartzExecutors

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/cache/ProcessInstanceExecCacheManager.java

@@ -0,0 +1,66 @@
+/*
+ * 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.server.master.runner.WorkflowExecuteThread;
+
+import java.util.Collection;
+
+/**
+ * cache of process instance id and WorkflowExecuteThread
+ */
+public interface ProcessInstanceExecCacheManager {
+
+ /**
+ * get WorkflowExecuteThread by process instance id
+ *
+ * @param processInstanceId processInstanceId
+ * @return WorkflowExecuteThread
+ */
+ WorkflowExecuteThread getByProcessInstanceId(int processInstanceId);
+
+ /**
+ * judge the process instance does it exist
+ *
+ * @param processInstanceId processInstanceId
+ * @return true - if process instance id exists in cache
+ */
+ boolean contains(int processInstanceId);
+
+ /**
+ * remove cache by process instance id
+ *
+ * @param processInstanceId processInstanceId
+ */
+ void removeByProcessInstanceId(int processInstanceId);
+
+ /**
+ * cache
+ *
+ * @param processInstanceId processInstanceId
+ * @param workflowExecuteThread notnull
+ */
+ void cache(int processInstanceId, WorkflowExecuteThread workflowExecuteThread);
+
+ /**
+ * get all WorkflowExecuteThread from cache
+ *
+ * @return all WorkflowExecuteThread in cache
+ */
+ Collection<WorkflowExecuteThread> getAll();
+}

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/cache/impl/ProcessInstanceExecCacheManagerImpl.java

@@ -0,0 +1,65 @@
+/*
+ * 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 org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager;
+import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThread;
+import org.springframework.stereotype.Component;
+
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * cache of process instance id and WorkflowExecuteThread
+ */
+@Component
+public class ProcessInstanceExecCacheManagerImpl implements ProcessInstanceExecCacheManager {
+
+ /**
+ * use a concurrent map cache
+ */
+ private final ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceExecMaps = new ConcurrentHashMap<>();
+
+ @Override
+ public WorkflowExecuteThread getByProcessInstanceId(int processInstanceId) {
+ return processInstanceExecMaps.get(processInstanceId);
+ }
+
+ @Override
+ public boolean contains(int processInstanceId) {
+ return processInstanceExecMaps.containsKey(processInstanceId);
+ }
+
+ @Override
+ public void removeByProcessInstanceId(int processInstanceId) {
+ processInstanceExecMaps.remove(processInstanceId);
+ }
+
+ @Override
+ public void cache(int processInstanceId, WorkflowExecuteThread workflowExecuteThread) {
+ if (workflowExecuteThread == null) {
+ return;
+ }
+ processInstanceExecMaps.put(processInstanceId, workflowExecuteThread);
+ }
+
+ @Override
+ public Collection<WorkflowExecuteThread> getAll() {
+ return processInstanceExecMaps.values();
+ }
+}

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/StateEventProcessor.java

@@ -26,10 +26,8 @@
 import org.apache.dolphinscheduler.remote.command.StateEventChangeCommand;
 import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
 import org.apache.dolphinscheduler.server.master.processor.queue.StateEventResponseService;
-import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThread;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
-import java.util.concurrent.ConcurrentHashMap;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -51,9 +49,6 @@ public StateEventProcessor() {
  stateEventResponseService = SpringApplicationContext.getBean(StateEventResponseService.class);
  }
 
- public void init(ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceExecMaps) {
- this.stateEventResponseService.init(processInstanceExecMaps);
- }
 
  @Override
  public void process(Channel channel, Command command) {

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

@@ -28,10 +28,8 @@
 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;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
-import java.util.concurrent.ConcurrentHashMap;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -62,9 +60,6 @@ public TaskAckProcessor() {
  this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
  }
 
- public void init(ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceExecMaps) {
- this.taskResponseService.init(processInstanceExecMaps);
- }
 
  /**
  * task ack process

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

@@ -27,10 +27,8 @@
 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;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
-import java.util.concurrent.ConcurrentHashMap;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -61,9 +59,6 @@ public TaskResponseProcessor() {
  this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
  }
 
- public void init(ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceExecMaps) {
- this.taskResponseService.init(processInstanceExecMaps);
- }
 
  /**
  * task final result response

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/StateEventResponseService.java

@@ -21,19 +21,20 @@
 import org.apache.dolphinscheduler.common.enums.StateEvent;
 import org.apache.dolphinscheduler.common.thread.Stopper;
 import org.apache.dolphinscheduler.remote.command.StateEventResponseCommand;
+import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager;
 import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThread;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import io.netty.channel.Channel;
@@ -59,13 +60,12 @@ public class StateEventResponseService {
  */
  private Thread responseWorker;
 
- private ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceMapper;
+ /**
+ * cache of process instance id and WorkflowExecuteThread
+ */
+ @Autowired
+ private ProcessInstanceExecCacheManager processInstanceExecCacheManager;
 
- public void init(ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceMapper) {
- if (this.processInstanceMapper == null) {
- this.processInstanceMapper = processInstanceMapper;
- }
- }
 
  @PostConstruct
  public void start() {
@@ -131,12 +131,12 @@ private void writeResponse(StateEvent stateEvent, ExecutionStatus status) {
 
  private void persist(StateEvent stateEvent) {
  try {
- if (!this.processInstanceMapper.containsKey(stateEvent.getProcessInstanceId())) {
+ if (!this.processInstanceExecCacheManager.contains(stateEvent.getProcessInstanceId())) {
  writeResponse(stateEvent, ExecutionStatus.FAILURE);
  return;
  }
 
- WorkflowExecuteThread workflowExecuteThread = this.processInstanceMapper.get(stateEvent.getProcessInstanceId());
+ WorkflowExecuteThread workflowExecuteThread = this.processInstanceExecCacheManager.getByProcessInstanceId(stateEvent.getProcessInstanceId());
  workflowExecuteThread.addStateEvent(stateEvent);
  writeResponse(stateEvent, ExecutionStatus.SUCCESS);
  } catch (Exception e) {

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseService.java

@@ -25,13 +25,13 @@
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.remote.command.DBTaskAckCommand;
 import org.apache.dolphinscheduler.remote.command.DBTaskResponseCommand;
+import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager;
 import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteThread;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
 
 import javax.annotation.PostConstruct;
@@ -66,18 +66,18 @@ public class TaskResponseService {
  @Autowired
  private ProcessService processService;
 
+ /**
+ * cache of process instance id and WorkflowExecuteThread
+ */
+ @Autowired
+ private ProcessInstanceExecCacheManager processInstanceExecCacheManager;
+
  /**
  * task response worker
  */
  private Thread taskResponseWorker;
 
- private ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceMapper;
 
- public void init(ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceMapper) {
- if (this.processInstanceMapper == null) {
- this.processInstanceMapper = processInstanceMapper;
- }
- }
 
  @PostConstruct
  public void start() {
@@ -194,7 +194,7 @@ private void persist(TaskResponseEvent taskResponseEvent) {
  default:
  throw new IllegalArgumentException("invalid event type : " + event);
  }
- WorkflowExecuteThread workflowExecuteThread = this.processInstanceMapper.get(taskResponseEvent.getProcessInstanceId());
+ WorkflowExecuteThread workflowExecuteThread = this.processInstanceExecCacheManager.getByProcessInstanceId(taskResponseEvent.getProcessInstanceId());
  if (workflowExecuteThread != null) {
  StateEvent stateEvent = new StateEvent();
  stateEvent.setProcessInstanceId(taskResponseEvent.getProcessInstanceId());