[Fix-17820][Master] Fix task timeout alerts failed (#17818)

Author: njnu-seafish

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java

@@ -244,6 +244,13 @@ private void saveTaskTimeoutAlert(Alert alert, String content, int alertGroupId)
     public void sendTaskTimeoutAlert(WorkflowInstance workflowInstance,
                                      TaskInstance taskInstance,
                                      ProjectUser projectUser) {
+        if (projectUser == null) {
+            throw new IllegalArgumentException("projectUser must not be null");
+        }
+        if (workflowInstance.getWarningGroupId() == null) {
+            throw new IllegalArgumentException("warningGroupId of the workflow instance must not be null");
+        }
+
         Alert alert = new Alert();
         List<WorkflowAlertContent> workflowAlertContentList = new ArrayList<>(1);
         WorkflowAlertContent workflowAlertContent = WorkflowAlertContent.builder()

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProjectDao.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.dao.repository;
 
 import org.apache.dolphinscheduler.dao.entity.Project;
+import org.apache.dolphinscheduler.dao.entity.ProjectUser;
 
 import java.util.Collection;
 import java.util.List;
@@ -28,4 +29,5 @@ public interface ProjectDao extends IDao<Project> {
 
     Project queryByCode(Long projectCode);
 
+    ProjectUser queryProjectWithUserByWorkflowInstanceId(int workflowInstanceId);
 }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProjectDaoImpl.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.dao.repository.impl;
 
 import org.apache.dolphinscheduler.dao.entity.Project;
+import org.apache.dolphinscheduler.dao.entity.ProjectUser;
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
 import org.apache.dolphinscheduler.dao.repository.BaseDao;
 import org.apache.dolphinscheduler.dao.repository.ProjectDao;
@@ -45,4 +46,9 @@ public List<Project> queryByCodes(Collection<Long> projectCodes) {
     public Project queryByCode(Long projectCode) {
         return mybatisMapper.queryByCode(projectCode);
     }
+
+    @Override
+    public ProjectUser queryProjectWithUserByWorkflowInstanceId(int workflowInstanceId) {
+        return mybatisMapper.queryProjectWithUserByWorkflowInstanceId(workflowInstanceId);
+    }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskTimeoutLifecycleEventHandler.java

@@ -62,10 +62,18 @@ public void handle(final ITaskStateAction taskStateAction,
             log.info("The task {} TimeoutStrategy is null.", taskName);
             return;
         }
+
+        final WorkflowInstance workflowInstance = workflowExecutionRunnable.getWorkflowInstance();
+        final boolean shouldSendAlert = workflowInstance.getWarningGroupId() != null;
+
         switch (timeoutNotifyStrategy) {
             case WARN:
                 log.info("The task {} TimeoutStrategy is WARN, try to send a timeout alert.", taskName);
-                doTaskTimeoutAlert(taskExecutionRunnable);
+                if (shouldSendAlert) {
+                    doTaskTimeoutAlert(taskExecutionRunnable);
+                } else {
+                    log.info("Skipped sending timeout alert for task {} because warningGroupId is null.", taskName);
+                }
                 break;
             case FAILED:
                 log.info("The task {} TimeoutStrategy is FAILED, try to publish a kill event.", taskName);
@@ -76,7 +84,11 @@ public void handle(final ITaskStateAction taskStateAction,
                         "The task {} TimeoutStrategy is WARNFAILED, try to publish a kill event and send a timeout alert.",
                         taskName);
                 doTaskTimeoutKill(taskExecutionRunnable);
-                doTaskTimeoutAlert(taskExecutionRunnable);
+                if (shouldSendAlert) {
+                    doTaskTimeoutAlert(taskExecutionRunnable);
+                } else {
+                    log.info("Skipped sending timeout alert for task {} because warningGroupId is null.", taskName);
+                }
             default:
                 log.warn("The task {} TimeoutStrategy is invalided.", taskName);
                 break;
@@ -90,8 +102,7 @@ private void doTaskTimeoutKill(final ITaskExecutionRunnable taskExecutionRunnabl
     private void doTaskTimeoutAlert(final ITaskExecutionRunnable taskExecutionRunnable) {
         final WorkflowInstance workflowInstance = taskExecutionRunnable.getWorkflowInstance();
         final TaskInstance taskInstance = taskExecutionRunnable.getTaskInstance();
-        // todo: inject the projectUser
-        workflowAlertManager.sendTaskTimeoutAlert(workflowInstance, taskInstance, null);
+        workflowAlertManager.sendTaskTimeoutAlert(workflowInstance, taskInstance);
     }
 
     @Override

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java

@@ -260,8 +260,8 @@ public boolean isNeedToSendWarning(WorkflowInstance workflowInstance) {
     }
 
     public void sendTaskTimeoutAlert(WorkflowInstance workflowInstance,
-                                     TaskInstance taskInstance,
-                                     ProjectUser projectUser) {
+                                     TaskInstance taskInstance) {
+        ProjectUser projectUser = projectDao.queryProjectWithUserByWorkflowInstanceId(workflowInstance.getId());
         alertDao.sendTaskTimeoutAlert(workflowInstance, taskInstance, projectUser);
     }
 }