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

Add workflowInstanceHost and taskInstanceHost in ITaskInstanceExecutionEvent (#14827)

This commit is contained in:
Wenjun Ruan 2023-08-29 13:32:17 +08:00 committed by GitHub
parent 9e60632201
commit 742f0c3e60
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 35 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/ITaskInstanceExecutionEvent.java
View File

@ -27,7 +27,13 @@ public interface ITaskInstanceExecutionEvent {
void setEventSendTime(long eventSendTime);
void setHost(String host);
void setWorkflowInstanceHost(String host);
String getWorkflowInstanceHost();
void setTaskInstanceHost(String host);
String getTaskInstanceHost();
TaskInstanceExecutionEventType getEventType();

4
dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/TaskInstanceExecutionFinishEvent.java
View File

@ -34,7 +34,9 @@ public class TaskInstanceExecutionFinishEvent implements ITaskInstanceExecutionE
private long startTime;
private String host;
private String taskInstanceHost;
private String workflowInstanceHost;
private String logPath;

4
dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/TaskInstanceExecutionInfoEvent.java
View File

@ -32,7 +32,9 @@ public class TaskInstanceExecutionInfoEvent implements ITaskInstanceExecutionEve
private long startTime;
private String host;
private String workflowInstanceHost;
private String taskInstanceHost;
private String logPath;

4
dolphinscheduler-extract/dolphinscheduler-extract-master/src/main/java/org/apache/dolphinscheduler/extract/master/transportor/TaskInstanceExecutionRunningEvent.java
View File

@ -34,7 +34,9 @@ public class TaskInstanceExecutionRunningEvent implements ITaskInstanceExecution
private long startTime;
private String host;
private String taskInstanceHost;
private String workflowInstanceHost;
private TaskExecutionStatus status;

6
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskEvent.java
View File

@ -114,7 +114,7 @@ public class TaskEvent {
event.setExecutePath(command.getExecutePath());
event.setLogPath(command.getLogPath());
event.setAppIds(command.getAppIds());
event.setWorkerAddress(command.getHost());
event.setWorkerAddress(command.getTaskInstanceHost());
event.setEvent(TaskEventType.RUNNING);
return event;
}
@ -131,7 +131,7 @@ public class TaskEvent {
event.setProcessId(command.getProcessId());
event.setAppIds(command.getAppIds());
event.setVarPool(command.getVarPool());
event.setWorkerAddress(command.getHost());
event.setWorkerAddress(command.getTaskInstanceHost());
event.setEvent(TaskEventType.RESULT);
return event;
}
@ -151,7 +151,7 @@ public class TaskEvent {
event.setTaskInstanceId(command.getTaskInstanceId());
event.setStartTime(DateUtils.timeStampToDate(command.getStartTime()));
event.setLogPath(command.getLogPath());
event.setWorkerAddress(command.getHost());
event.setWorkerAddress(command.getTaskInstanceHost());
event.setEvent(TaskEventType.UPDATE_PID);
return event;
}

5
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecuteRunningEventSender.java
View File

@ -35,7 +35,7 @@ public class LogicTaskInstanceExecuteRunningEventSender
public void sendMessage(TaskInstanceExecutionRunningEvent taskInstanceExecutionRunningEvent) {
ITaskInstanceExecutionEventListener iTaskInstanceExecutionEventListener =
SingletonJdkDynamicRpcClientProxyFactory
.getProxyClient(taskInstanceExecutionRunningEvent.getHost(),
.getProxyClient(taskInstanceExecutionRunningEvent.getWorkflowInstanceHost(),
ITaskInstanceExecutionEventListener.class);
iTaskInstanceExecutionEventListener.onTaskInstanceExecutionRunning(taskInstanceExecutionRunningEvent);
}
@ -47,7 +47,8 @@ public class LogicTaskInstanceExecuteRunningEventSender
taskExecuteRunningMessage.setProcessInstanceId(taskExecutionContext.getProcessInstanceId());
taskExecuteRunningMessage.setStatus(taskExecutionContext.getCurrentExecutionStatus());
taskExecuteRunningMessage.setLogPath(taskExecutionContext.getLogPath());
taskExecuteRunningMessage.setHost(taskExecutionContext.getHost());
taskExecuteRunningMessage.setWorkflowInstanceHost(taskExecutionContext.getWorkflowInstanceHost());
taskExecuteRunningMessage.setTaskInstanceHost(taskExecutionContext.getHost());
taskExecuteRunningMessage.setStartTime(taskExecutionContext.getStartTime());
taskExecuteRunningMessage.setExecutePath(taskExecutionContext.getExecutePath());
taskExecuteRunningMessage.setAppIds(taskExecutionContext.getAppIds());

5
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/message/LogicTaskInstanceExecutionFinishEventSender.java
View File

@ -33,7 +33,7 @@ public class LogicTaskInstanceExecutionFinishEventSender
public void sendMessage(TaskInstanceExecutionFinishEvent message) {
ITaskInstanceExecutionEventListener iTaskInstanceExecutionEventListener =
SingletonJdkDynamicRpcClientProxyFactory
.getProxyClient(message.getHost(), ITaskInstanceExecutionEventListener.class);
.getProxyClient(message.getWorkflowInstanceHost(), ITaskInstanceExecutionEventListener.class);
iTaskInstanceExecutionEventListener.onTaskInstanceExecutionFinish(message);
}
@ -47,7 +47,8 @@ public class LogicTaskInstanceExecutionFinishEventSender
taskExecuteResultMessage.setExecutePath(taskExecutionContext.getExecutePath());
taskExecuteResultMessage.setAppIds(taskExecutionContext.getAppIds());
taskExecuteResultMessage.setProcessId(taskExecutionContext.getProcessId());
taskExecuteResultMessage.setHost(taskExecutionContext.getHost());
taskExecuteResultMessage.setWorkflowInstanceHost(taskExecutionContext.getWorkflowInstanceHost());
taskExecuteResultMessage.setTaskInstanceHost(taskExecutionContext.getHost());
taskExecuteResultMessage.setStartTime(taskExecutionContext.getStartTime());
taskExecuteResultMessage.setEndTime(taskExecutionContext.getEndTime());
taskExecuteResultMessage.setVarPool(taskExecutionContext.getVarPool());

2
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageRetryRunner.java
View File

@ -96,7 +96,7 @@ public class MessageRetryRunner extends BaseDaemonThread {
List<TaskInstanceMessage> taskInstanceMessages = this.needToRetryMessages.get(taskInstanceId);
if (taskInstanceMessages != null) {
taskInstanceMessages.forEach(taskInstanceMessage -> {
taskInstanceMessage.getEvent().setHost(messageReceiverHost);
taskInstanceMessage.getEvent().setWorkflowInstanceHost(messageReceiverHost);
});
}
}

5
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskInstanceExecutionFinishEventSender.java
View File

@ -34,7 +34,7 @@ public class TaskInstanceExecutionFinishEventSender
public void sendEvent(TaskInstanceExecutionFinishEvent taskInstanceExecutionFinishEvent) {
ITaskInstanceExecutionEventListener iTaskInstanceExecutionEventListener =
SingletonJdkDynamicRpcClientProxyFactory
.getProxyClient(taskInstanceExecutionFinishEvent.getHost(),
.getProxyClient(taskInstanceExecutionFinishEvent.getWorkflowInstanceHost(),
ITaskInstanceExecutionEventListener.class);
iTaskInstanceExecutionEventListener.onTaskInstanceExecutionFinish(taskInstanceExecutionFinishEvent);
}
@ -49,7 +49,8 @@ public class TaskInstanceExecutionFinishEventSender
taskExecuteResultMessage.setExecutePath(taskExecutionContext.getExecutePath());
taskExecuteResultMessage.setAppIds(taskExecutionContext.getAppIds());
taskExecuteResultMessage.setProcessId(taskExecutionContext.getProcessId());
taskExecuteResultMessage.setHost(taskExecutionContext.getHost());
taskExecuteResultMessage.setWorkflowInstanceHost(taskExecutionContext.getWorkflowInstanceHost());
taskExecuteResultMessage.setTaskInstanceHost(taskExecutionContext.getHost());
taskExecuteResultMessage.setStartTime(taskExecutionContext.getStartTime());
taskExecuteResultMessage.setEndTime(taskExecutionContext.getEndTime());
taskExecuteResultMessage.setVarPool(taskExecutionContext.getVarPool());

5
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskInstanceExecutionInfoUpdateEventSender.java
View File

@ -36,7 +36,7 @@ public class TaskInstanceExecutionInfoUpdateEventSender
public void sendEvent(TaskInstanceExecutionInfoEvent taskInstanceExecutionInfoEvent) {
ITaskInstanceExecutionEventListener iTaskInstanceExecutionEventListener =
SingletonJdkDynamicRpcClientProxyFactory
.getProxyClient(taskInstanceExecutionInfoEvent.getHost(),
.getProxyClient(taskInstanceExecutionInfoEvent.getWorkflowInstanceHost(),
ITaskInstanceExecutionEventListener.class);
iTaskInstanceExecutionEventListener.onTaskInstanceExecutionInfoUpdate(taskInstanceExecutionInfoEvent);
}
@ -46,7 +46,8 @@ public class TaskInstanceExecutionInfoUpdateEventSender
TaskInstanceExecutionInfoEvent taskUpdatePidRequest = new TaskInstanceExecutionInfoEvent();
taskUpdatePidRequest.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());
taskUpdatePidRequest.setProcessInstanceId(taskExecutionContext.getProcessInstanceId());
taskUpdatePidRequest.setHost(taskExecutionContext.getHost());
taskUpdatePidRequest.setWorkflowInstanceHost(taskExecutionContext.getWorkflowInstanceHost());
taskUpdatePidRequest.setTaskInstanceHost(taskExecutionContext.getHost());
taskUpdatePidRequest.setStartTime(taskExecutionContext.getStartTime());
return taskUpdatePidRequest;
}

5
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/TaskInstanceExecutionRunningEventSender.java
View File

@ -36,7 +36,7 @@ public class TaskInstanceExecutionRunningEventSender
public void sendEvent(TaskInstanceExecutionRunningEvent taskInstanceExecutionRunningEvent) {
ITaskInstanceExecutionEventListener iTaskInstanceExecutionEventListener =
SingletonJdkDynamicRpcClientProxyFactory
.getProxyClient(taskInstanceExecutionRunningEvent.getHost(),
.getProxyClient(taskInstanceExecutionRunningEvent.getWorkflowInstanceHost(),
ITaskInstanceExecutionEventListener.class);
iTaskInstanceExecutionEventListener.onTaskInstanceExecutionRunning(taskInstanceExecutionRunningEvent);
}
@ -48,7 +48,8 @@ public class TaskInstanceExecutionRunningEventSender
taskExecuteRunningMessage.setProcessInstanceId(taskExecutionContext.getProcessInstanceId());
taskExecuteRunningMessage.setStatus(taskExecutionContext.getCurrentExecutionStatus());
taskExecuteRunningMessage.setLogPath(taskExecutionContext.getLogPath());
taskExecuteRunningMessage.setHost(taskExecutionContext.getHost());
taskExecuteRunningMessage.setWorkflowInstanceHost(taskExecutionContext.getWorkflowInstanceHost());
taskExecuteRunningMessage.setTaskInstanceHost(taskExecutionContext.getHost());
taskExecuteRunningMessage.setStartTime(taskExecutionContext.getStartTime());
taskExecuteRunningMessage.setExecutePath(taskExecutionContext.getExecutePath());
taskExecuteRunningMessage.setAppIds(taskExecutionContext.getAppIds());