[Fix-17854] [Worker&SQL Task] Fix SQL task query result alert not being sent

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java

@@ -29,7 +29,7 @@ public enum AlertType {
 
     /**
      * 0 workflow instance failure, 1 workflow instance success, 2 workflow instance blocked, 3 workflow instance timeout, 4 fault tolerance warning,
-     * 5 task failure, 6 task success, 7 task timeout
+     * 5 task failure, 6 task success, 7 task timeout, 8 task result
       */
     WORKFLOW_INSTANCE_FAILURE(0, "workflow instance failure"),
     WORKFLOW_INSTANCE_SUCCESS(1, "workflow instance success"),
@@ -39,6 +39,7 @@ public enum AlertType {
     TASK_FAILURE(5, "task failure"),
     TASK_SUCCESS(6, "task success"),
     TASK_TIMEOUT(7, "task timeout"),
+    TASK_RESULT(8, "task result"),
     ;
 
     AlertType(int code, String descp) {

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

@@ -108,6 +108,33 @@ public int addAlert(Alert alert) {
         return count;
     }
 
+    /**
+     * Insert a task-result alert idempotently. If an alert with the same sign,
+     * workflow instance id and alert type already exists, the insert is skipped.
+     * <p>This guards against duplicate inserts caused by at-least-once delivery
+     * of task success lifecycle events.
+     *
+     * @param alert alert, must have sign, workflowInstanceId and alertType set
+     * @return insert count (1 if inserted, 0 if skipped)
+     */
+    public int addTaskResultAlert(Alert alert) {
+        if (null == alert.getAlertGroupId() || NumberUtils.INTEGER_ZERO.equals(alert.getAlertGroupId())) {
+            log.warn("the value of alertGroupId is null or 0 ");
+            return 0;
+        }
+
+        String sign = generateSign(alert);
+        alert.setSign(sign);
+        int count = alertMapper.insertTaskResultAlertIfAbsent(alert);
+        if (count > 0) {
+            log.info("add task result alert to db , alert: {}", alert);
+        } else {
+            log.info("skip duplicate task result alert, sign: {}, workflowInstanceId: {}", sign,
+                    alert.getWorkflowInstanceId());
+        }
+        return count;
+    }
+
     /**
      * update alert sending(execution) status
      *

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java

@@ -43,6 +43,12 @@ List<Alert> listingAlertByStatus(@Param("minAlertId") int minAlertId, @Param("al
     void insertAlertWhenServerCrash(@Param("alert") Alert alert,
                                     @Param("crashAlarmSuppressionStartTime") Date crashAlarmSuppressionStartTime);
 
+    /**
+     * Insert a task-result alert only if no alert with the same sign, workflow instance id and alert type
+     * already exists. This makes the insert idempotent against at-least-once event delivery.
+     */
+    int insertTaskResultAlertIfAbsent(@Param("alert") Alert alert);
+
     void deleteByWorkflowInstanceId(@Param("workflowInstanceId") Integer processInstanceId);
 
     List<Alert> selectByWorkflowInstanceId(@Param("workflowInstanceId") Integer processInstanceId);

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml

@@ -51,6 +51,34 @@
         having count(*) = 0
     </insert>
 
+    <!--
+        Idempotent insert for task-result alerts.  Only inserts a row when no alert
+        with the same sign, workflow_instance_id and alert_type already exists.
+        This guards against duplicate inserts caused by at-least-once event delivery.
+    -->
+    <insert id="insertTaskResultAlertIfAbsent">
+        insert into t_ds_alert(sign, title, content, alert_status, warning_type, log, alertgroup_id, create_time,
+                               update_time, project_code, workflow_definition_code, workflow_instance_id, alert_type)
+        SELECT #{alert.sign},
+               #{alert.title},
+               #{alert.content},
+               #{alert.alertStatus.code},
+               #{alert.warningType.code},
+               #{alert.log},
+               #{alert.alertGroupId},
+               #{alert.createTime},
+               #{alert.updateTime},
+               #{alert.projectCode},
+               #{alert.workflowDefinitionCode},
+               #{alert.workflowInstanceId},
+               #{alert.alertType.code}
+        from t_ds_alert
+        where sign = #{alert.sign}
+          and workflow_instance_id = #{alert.workflowInstanceId}
+          and alert_type = #{alert.alertType.code}
+        having count(*) = 0
+    </insert>
+
     <select id="listingAlertByStatus" resultType="org.apache.dolphinscheduler.dao.entity.Alert">
         select
         <include refid="baseSql"/>

dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/AlertDaoTest.java

@@ -18,6 +18,8 @@
 package org.apache.dolphinscheduler.dao.repository.impl;
 
 import org.apache.dolphinscheduler.common.enums.AlertStatus;
+import org.apache.dolphinscheduler.common.enums.AlertType;
+import org.apache.dolphinscheduler.common.enums.WarningType;
 import org.apache.dolphinscheduler.dao.AlertDao;
 import org.apache.dolphinscheduler.dao.BaseDaoTest;
 import org.apache.dolphinscheduler.dao.entity.Alert;
@@ -66,4 +68,45 @@ void testSendServerStoppedAlert() {
                 .count();
         Assertions.assertEquals(1L, count);
     }
+
+    @Test
+    void testAddTaskResultAlertIdempotent() {
+        String content = "[{\"taskName\":\"sql-task-1\",\"result\":\"ok\"}]";
+        int workflowInstanceId = 999999;
+
+        Alert alert = new Alert();
+        alert.setTitle("SQL Task Result");
+        alert.setContent(content);
+        alert.setWarningType(WarningType.SUCCESS);
+        alert.setAlertGroupId(1);
+        alert.setAlertStatus(AlertStatus.WAIT_EXECUTION);
+        alert.setWorkflowInstanceId(workflowInstanceId);
+        alert.setAlertType(AlertType.TASK_RESULT);
+        alert.setCreateTime(new java.util.Date());
+
+        // First insert should succeed
+        int firstCount = alertDao.addTaskResultAlert(alert);
+        Assertions.assertEquals(1, firstCount);
+
+        // Second insert with the same content + workflowInstanceId + alertType should be skipped
+        Alert duplicateAlert = new Alert();
+        duplicateAlert.setTitle("SQL Task Result");
+        duplicateAlert.setContent(content);
+        duplicateAlert.setWarningType(WarningType.SUCCESS);
+        duplicateAlert.setAlertGroupId(1);
+        duplicateAlert.setAlertStatus(AlertStatus.WAIT_EXECUTION);
+        duplicateAlert.setWorkflowInstanceId(workflowInstanceId);
+        duplicateAlert.setAlertType(AlertType.TASK_RESULT);
+        duplicateAlert.setCreateTime(new java.util.Date());
+
+        int secondCount = alertDao.addTaskResultAlert(duplicateAlert);
+        Assertions.assertEquals(0, secondCount);
+
+        // Verify only one alert row exists for this workflow instance
+        long count = alertDao.listAlerts(workflowInstanceId)
+                .stream()
+                .filter(a -> a.getAlertType() == AlertType.TASK_RESULT)
+                .count();
+        Assertions.assertEquals(1L, count);
+    }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event;
 
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventType;
 import org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.AbstractTaskLifecycleEvent;
@@ -41,6 +42,10 @@ public class TaskSuccessLifecycleEvent extends AbstractTaskLifecycleEvent {
 
     private final List<Property> varPool;
 
+    private final boolean needAlert;
+
+    private final TaskAlertInfo taskAlertInfo;
+
     @Override
     public ILifecycleEventType getEventType() {
         return TaskLifecycleEventType.SUCCEEDED;
@@ -52,6 +57,7 @@ public String toString() {
                 "task=" + taskExecution.getName() +
                 ", endTime=" + endTime +
                 ", varPool='" + varPool + '\'' +
+                ", needAlert=" + needAlert +
                 '}';
     }
 }

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

@@ -17,33 +17,60 @@
 
 package org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler;
 
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventType;
 import org.apache.dolphinscheduler.server.master.engine.task.client.TaskExecutorClient;
 import org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifecycleEventType;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskSuccessLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.task.statemachine.ITaskStateAction;
 import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
+import org.apache.dolphinscheduler.service.alert.WorkflowAlertManager;
 import org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorLifecycleEventReporter;
 import org.apache.dolphinscheduler.task.executor.events.TaskExecutorLifecycleEventType;
 
+import lombok.extern.slf4j.Slf4j;
+
 import org.springframework.stereotype.Component;
 
+@Slf4j
 @Component
 public class TaskSuccessLifecycleEventHandler extends AbstractTaskLifecycleEventHandler<TaskSuccessLifecycleEvent> {
 
     private final TaskExecutorClient taskExecutorClient;
 
-    public TaskSuccessLifecycleEventHandler(final TaskExecutorClient taskExecutorClient) {
+    private final WorkflowAlertManager workflowAlertManager;
+
+    public TaskSuccessLifecycleEventHandler(final TaskExecutorClient taskExecutorClient,
+                                            final WorkflowAlertManager workflowAlertManager) {
         this.taskExecutorClient = taskExecutorClient;
+        this.workflowAlertManager = workflowAlertManager;
     }
 
     @Override
     public void handle(final ITaskStateAction taskStateAction,
                        final IWorkflowExecution workflowExecution,
                        final ITaskExecution taskExecution,
                        final TaskSuccessLifecycleEvent taskSuccessEvent) {
+        // 1. State transition + DB persistence (may throw if state mismatch)
         taskStateAction.onSucceedEvent(workflowExecution, taskExecution, taskSuccessEvent);
+
+        // 2. Persist task-result alert only after the success state transition is confirmed
+        if (taskSuccessEvent.isNeedAlert()) {
+            final TaskAlertInfo taskAlertInfo = taskSuccessEvent.getTaskAlertInfo();
+            if (taskAlertInfo != null && taskAlertInfo.getAlertGroupId() != null
+                    && taskAlertInfo.getAlertGroupId() > 0) {
+                workflowAlertManager.sendTaskResultAlert(
+                        taskExecution.getWorkflowInstance(),
+                        taskExecution.getTaskInstance(),
+                        taskAlertInfo);
+            } else {
+                log.warn("Task: {} need alert but alertGroupId is invalid, skip sending alert",
+                        taskExecution.getName());
+            }
+        }
+
+        // 3. ACK the worker — only after state transition and alert persistence are done
         taskExecutorClient.ackTaskExecutorLifecycleEvent(
                 taskExecution,
                 new ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck(

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java

@@ -116,6 +116,8 @@ public void onTaskExecutorSuccess(final TaskExecutorSuccessLifecycleEvent taskEx
                     .taskExecution(taskExecution)
                     .endTime(new Date(taskExecutorSuccessLifecycleEvent.getEndTime()))
                     .varPool(taskExecutorSuccessLifecycleEvent.getVarPool())
+                    .needAlert(taskExecutorSuccessLifecycleEvent.isNeedAlert())
+                    .taskAlertInfo(taskExecutorSuccessLifecycleEvent.getTaskAlertInfo())
                     .build();
             taskExecution.getWorkflowEventBus().publish(taskSuccessEvent);
         } finally {

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

@@ -34,6 +34,7 @@
 import org.apache.dolphinscheduler.dao.repository.ProjectDao;
 import org.apache.dolphinscheduler.dao.repository.UserDao;
 import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionLogDao;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 
 import java.util.ArrayList;
 import java.util.Date;
@@ -208,4 +209,34 @@ public void sendWorkflowTimeoutAlert(WorkflowInstance workflowInstance) {
         alertDao.sendWorkflowTimeoutAlert(workflowInstance, projectUser);
     }
 
+    /**
+     * send task result alert
+     *
+     * @param workflowInstance workflow instance
+     * @param taskInstance     task instance
+     * @param taskAlertInfo    task alert info
+     */
+    public void sendTaskResultAlert(WorkflowInstance workflowInstance,
+                                    TaskInstance taskInstance,
+                                    TaskAlertInfo taskAlertInfo) {
+        if (taskAlertInfo == null || taskAlertInfo.getAlertGroupId() == null) {
+            return;
+        }
+
+        Alert alert = new Alert();
+        alert.setTitle(taskAlertInfo.getTitle());
+        alert.setContent(taskAlertInfo.getContent());
+        alert.setWarningType(WarningType.SUCCESS);
+        alert.setCreateTime(new Date());
+        alert.setAlertGroupId(taskAlertInfo.getAlertGroupId());
+        alert.setProjectCode(workflowInstance.getProjectCode());
+        alert.setWorkflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode());
+        alert.setWorkflowInstanceId(workflowInstance.getId());
+        alert.setAlertType(taskAlertInfo.getAlertType() != null
+                ? taskAlertInfo.getAlertType()
+                : AlertType.TASK_RESULT);
+        alertDao.addTaskResultAlert(alert);
+        log.info("Send task result alert for task: {} in workflow: {}",
+                taskInstance.getName(), workflowInstance.getName());
+    }
 }

dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java

@@ -19,6 +19,7 @@
 
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.task.executor.ITaskExecutor;
 
 import java.util.List;
@@ -48,6 +49,10 @@ public class TaskExecutorSuccessLifecycleEvent extends AbstractTaskExecutorLifec
 
     private Long latestReportTime;
 
+    private boolean needAlert;
+
+    private TaskAlertInfo taskAlertInfo;
+
     public static TaskExecutorSuccessLifecycleEvent of(final ITaskExecutor taskExecutor) {
         final TaskExecutionContext taskExecutionContext = taskExecutor.getTaskExecutionContext();
         return TaskExecutorSuccessLifecycleEvent.builder()
@@ -57,6 +62,8 @@ public static TaskExecutorSuccessLifecycleEvent of(final ITaskExecutor taskExecu
                 .varPool(taskExecutionContext.getVarPool())
                 .type(TaskExecutorLifecycleEventType.SUCCESS)
                 .endTime(taskExecutionContext.getEndTime())
+                .needAlert(taskExecutionContext.isNeedAlert())
+                .taskAlertInfo(taskExecutionContext.getTaskAlertInfo())
                 .build();
     }
 

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java

@@ -19,7 +19,6 @@
 
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
-import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
 
 import java.util.Map;
@@ -56,10 +55,6 @@ public abstract class AbstractTask {
      */
     protected volatile int exitStatusCode = -1;
 
-    protected boolean needAlert = false;
-
-    protected TaskAlertInfo taskAlertInfo;
-
     /**
      * constructor
      *
@@ -109,22 +104,6 @@ public void setAppIds(String appIds) {
         this.appIds = appIds;
     }
 
-    public boolean getNeedAlert() {
-        return needAlert;
-    }
-
-    public void setNeedAlert(boolean needAlert) {
-        this.needAlert = needAlert;
-    }
-
-    public TaskAlertInfo getTaskAlertInfo() {
-        return taskAlertInfo;
-    }
-
-    public void setTaskAlertInfo(TaskAlertInfo taskAlertInfo) {
-        this.taskAlertInfo = taskAlertInfo;
-    }
-
     /**
      * get task parameters
      *

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java

@@ -19,6 +19,7 @@
 
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
 import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext;
 
@@ -130,6 +131,10 @@ public class TaskExecutionContext implements Serializable {
 
     private final long firstDispatchTime = System.currentTimeMillis();
 
+    private boolean needAlert;
+
+    private TaskAlertInfo taskAlertInfo;
+
     public int increaseDispatchFailTimes() {
         return ++dispatchFailTimes;
     }