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

docs/docs/en/guide/upgrade/incompatible.md

@@ -56,3 +56,8 @@ This document records the incompatible updates between each version. You need to
   * **Removed derived properties**: `cmdTypeIfComplement`, `complementData` (related to complement-data executions; use the detail API to obtain them)
   * To obtain any of these fields, use the detail API `GET /projects/{projectCode}/workflow-instances/{id}` instead, which continues to return the full `WorkflowInstance` object. ([#18444](https://github.com/apache/dolphinscheduler/pull/18444))
 
+## 3.5.0
+
+* Add a unique constraint `uk_alert_dedup` on `(sign, workflow_instance_id, alert_type)` to the `t_ds_alert` table. The upgrade script automatically cleans up existing duplicate rows (keeping the one with the largest id) before adding the unique index. Task-result alerts are inserted by application code; the constraint rejects duplicates and the DAO layer treats the resulting `DuplicateKeyException` as a skip, ensuring idempotency for task-result alerts under concurrent delivery.([#18549](https://github.com/apache/dolphinscheduler/pull/18549))
+* `AlertSender#getAlertData` now explicitly rejects a null `alertType` by throwing `IllegalArgumentException` with an error log, instead of throwing a raw `NullPointerException`. During a rolling upgrade, **the Alert Server must be upgraded before Master/Worker**, otherwise an old Alert Server cannot recognize newly introduced `alert_type` enum values (e.g. `TASK_RESULT`) and alert delivery will fail. See the [upgrade guide](upgrade.md#rolling-upgrade-order) for details.([#18549](https://github.com/apache/dolphinscheduler/pull/18549))
+

docs/docs/en/guide/upgrade/upgrade.md

@@ -75,6 +75,14 @@ Execution result:
 - After version 3.3.X and later, we only support upgrading from 3.0.0. For versions lower than this, please download the historical version and upgrade to 3.0.0.
 - After version 3.3.X and later, binary packages no longer provide plugins dependencies by default, so when you use them for the first time, you need to download and install them yourself. For more information, please refer to [Pseudo-Cluster](../installation/pseudo-cluster.md).
 
+#### Rolling Upgrade Order
+
+When performing a rolling upgrade, **the Alert Server must be upgraded before Master/Worker**.
+
+Starting from 3.5.0, the Master may persist new `alert_type` enum values (e.g. `TASK_RESULT`) into the `t_ds_alert` table. If the Alert Server has not yet been upgraded to a version that includes the new enum value, MyBatis will deserialize the unknown value as `null`, causing `AlertSender` to throw a `NullPointerException` while building the alert data. The alert will remain stuck in `WAIT_EXECUTION` and never be delivered.
+
+Upgrading in the order "Alert Server → Master → Worker" eliminates this risk.
+
 #### Precautions after the upgrade
 
 The alert plugin may have some dirty data. After the upgrade, clear it manually by referring to SQL.

docs/docs/zh/guide/upgrade/incompatible.md

@@ -56,3 +56,8 @@
   * **移除的派生属性**：`cmdTypeIfComplement`、`complementData`（补数执行相关，如需获取请使用详情接口）
   * 如需获取这些字段，请使用详情接口 `GET /projects/{projectCode}/workflow-instances/{id}`，该接口仍返回完整的 `WorkflowInstance` 对象 ([#18444](https://github.com/apache/dolphinscheduler/pull/18444))
 
+## 3.5.0
+
+* 为 `t_ds_alert` 表的 `(sign, workflow_instance_id, alert_type)` 新增唯一约束 `uk_alert_dedup`。升级脚本会自动清理已存在的重复行（保留 id 最大的一条），之后添加唯一索引。任务结果告警由应用代码插入，重复插入由该约束拒绝，DAO 层将产生的 `DuplicateKeyException` 视为跳过，从而确保任务结果告警在并发投递时幂等。([#18549](https://github.com/apache/dolphinscheduler/pull/18549))
+* `AlertSender#getAlertData` 对 null `alertType` 增加了显式拒绝（抛 `IllegalArgumentException` 并记录错误日志），而非直接抛 `NullPointerException`。滚动升级时**必须先升级 Alert Server，再升级 Master/Worker**，否则旧 Alert Server 无法识别新增的 `alert_type` 枚举值（如 `TASK_RESULT`），会导致告警投递失败。详见[升级文档](upgrade.md#滚动升降级顺序)。([#18549](https://github.com/apache/dolphinscheduler/pull/18549))
+

docs/docs/zh/guide/upgrade/upgrade.md

@@ -76,6 +76,14 @@ jar 包 并添加到 `./tools/libs` 目录下，设置以下环境变量
 - 在 3.3.X 以及之后的版本，我们仅支持从 3.0.0 开始进行升级，低于此版本的请下载历史版本升级至 3.0.0。
 - 在 3.3.X 以及之后的版本，二进制包不再默认提供插件依赖，因此第一次使用时，需要自行下载安装。具体请参考请参照[伪集群部署(Pseudo-Cluster)](../installation/pseudo-cluster.md)
 
+#### 滚动升降级顺序
+
+在滚动升降级场景下，**必须先升级 Alert Server，再升级 Master/Worker**。
+
+从 3.5.0 起，Master 可能向 `t_ds_alert` 表写入新的 `alert_type` 枚举值（如 `TASK_RESULT`）。如果 Alert Server 尚未升级到包含该枚举值的版本，MyBatis 会将未知的枚举值反序列化为 `null`，导致 `AlertSender` 在构建告警数据时抛出 `NullPointerException`，告警将卡在 `WAIT_EXECUTION` 状态无法投递。
+
+按照"Alert Server → Master → Worker"的顺序升级可消除此风险。
+
 #### 升级后的注意事项
 
 在历史版本中可能告警插件会有一些脏数据，升级后请参考一下 SQL 手动清理。

dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/AlertSender.java

@@ -100,6 +100,13 @@ public List<AlertPluginInstance> getAlertPluginInstanceList(Alert event) {
 
     @Override
     public AlertData getAlertData(Alert event) {
+        if (event.getAlertType() == null) {
+            log.error("Alert {} has null alertType, refusing to send. This indicates the Alert Server version "
+                    + "is older than the Master that created the alert. Please upgrade the Alert Server first "
+                    + "(see upgrade guide: Alert Server -> Master -> Worker).", event.getId());
+            throw new IllegalArgumentException(
+                    "Cannot build AlertData with null alertType for alert " + event.getId());
+        }
         return AlertData.builder()
                 .id(event.getId())
                 .content(event.getContent())

dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java

@@ -25,6 +25,7 @@
 import static org.mockito.Mockito.when;
 
 import org.apache.dolphinscheduler.alert.api.AlertChannel;
+import org.apache.dolphinscheduler.alert.api.AlertData;
 import org.apache.dolphinscheduler.alert.api.AlertResult;
 import org.apache.dolphinscheduler.alert.config.AlertConfig;
 import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager;
@@ -210,4 +211,33 @@ void testSendAlert() {
                 .thenReturn(paramsMap);
         alertSender.syncTestSend(PLUGIN_DEFINE_ID, PLUGIN_INSTANCE_PARAMS);
     }
+
+    @Test
+    void testGetAlertDataWithNullAlertType() {
+        Alert alert = new Alert();
+        alert.setId(1);
+        alert.setTitle(TITLE);
+        alert.setContent(CONTENT);
+        alert.setAlertGroupId(ALERT_GROUP_ID);
+        alert.setWarningType(WarningType.FAILURE);
+        alert.setAlertType(null);
+
+        Assertions.assertThrows(IllegalArgumentException.class,
+                () -> alertSender.getAlertData(alert));
+    }
+
+    @Test
+    void testGetAlertDataWithValidAlertType() {
+        Alert alert = new Alert();
+        alert.setId(2);
+        alert.setTitle(TITLE);
+        alert.setContent(CONTENT);
+        alert.setAlertGroupId(ALERT_GROUP_ID);
+        alert.setWarningType(WarningType.FAILURE);
+        alert.setAlertType(AlertType.TASK_RESULT);
+
+        AlertData alertData = alertSender.getAlertData(alert);
+        Assertions.assertNotNull(alertData);
+        Assertions.assertEquals(AlertType.TASK_RESULT.getCode(), alertData.getAlertType());
+    }
 }

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

@@ -55,6 +55,7 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Component;
 
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
@@ -108,6 +109,33 @@ public int addAlert(Alert alert) {
         return count;
     }
 
+    /**
+     * Insert a task-result alert idempotently. The uk_alert_dedup unique constraint on
+     * (sign, workflow_instance_id, alert_type) rejects duplicates; a DuplicateKeyException
+     * is caught here and treated as a skip.
+     *
+     * @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);
+        try {
+            int count = alertMapper.insertTaskResultAlert(alert);
+            log.info("add task result alert to db , alert: {}", alert);
+            return count;
+        } catch (DuplicateKeyException e) {
+            // The uk_alert_dedup unique constraint caught a duplicate — treat as a skip.
+            log.info("skip duplicate task result alert, sign: {}, workflowInstanceId: {}", sign,
+                    alert.getWorkflowInstanceId());
+            return 0;
+        }
+    }
+
     /**
      * update alert sending(execution) status
      *
@@ -132,6 +160,13 @@ public int updateAlert(AlertStatus alertStatus, String log, int id) {
      * @return sign's str
      */
     private String generateSign(Alert alert) {
+        // Task-result alerts include the task instance ID in the sign so that two
+        // different tasks returning identical results are not treated as duplicates.
+        // For other alert types taskInstanceId is null and the sign falls back to
+        // content-only, preserving the original behaviour.
+        if (alert.getTaskInstanceId() != null) {
+            return DigestUtils.sha1Hex(alert.getTaskInstanceId() + "|" + alert.getContent()).toLowerCase();
+        }
         return Optional.of(alert)
                 .map(Alert::getContent)
                 .map(DigestUtils::sha1Hex)

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java

@@ -84,6 +84,12 @@ public class Alert {
     @TableField("alert_type")
     private AlertType alertType;
 
+    /**
+     * Transient — used only for sign generation in idempotent task-result alert inserts.
+     */
+    @TableField(exist = false)
+    private Integer taskInstanceId;
+
     @TableField(exist = false)
     private Map<String, Object> info = new HashMap<>();
 

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

@@ -43,6 +43,13 @@ List<Alert> listingAlertByStatus(@Param("minAlertId") int minAlertId, @Param("al
     void insertAlertWhenServerCrash(@Param("alert") Alert alert,
                                     @Param("crashAlarmSuppressionStartTime") Date crashAlarmSuppressionStartTime);
 
+    /**
+     * Insert a task-result alert. Idempotency is enforced by the uk_alert_dedup unique constraint
+     * on (sign, workflow_instance_id, alert_type); the DAO layer catches DuplicateKeyException
+     * and treats it as a skip.
+     */
+    int insertTaskResultAlert(@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,21 @@
         having count(*) = 0
     </insert>
 
+    <!--
+      Task-result alerts rely on the uk_alert_dedup unique constraint
+      on (sign, workflow_instance_id, alert_type) for idempotency.
+      The DAO layer catches DuplicateKeyException and treats it as a skip.
+    -->
+    <insert id="insertTaskResultAlert">
+        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)
+        VALUES (#{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})
+    </insert>
+
     <select id="listingAlertByStatus" resultType="org.apache.dolphinscheduler.dao.entity.Alert">
         select
         <include refid="baseSql"/>

dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql

@@ -285,7 +285,8 @@ CREATE TABLE t_ds_alert
     workflow_instance_id     int(11) DEFAULT NULL,
     alert_type     int(11) DEFAULT NULL,
     PRIMARY KEY (id),
-    KEY            idx_sign (sign)
+    KEY            idx_sign (sign),
+    UNIQUE KEY     uk_alert_dedup (sign, workflow_instance_id, alert_type)
 );
 
 -- ----------------------------

dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql

@@ -291,7 +291,8 @@ CREATE TABLE `t_ds_alert` (
   `alert_type` int(11) DEFAULT NULL COMMENT 'alert_type',
   PRIMARY KEY (`id`),
   KEY `idx_status` (`alert_status`) USING BTREE,
-  KEY `idx_sign` (`sign`) USING BTREE
+  KEY `idx_sign` (`sign`) USING BTREE,
+  UNIQUE KEY `uk_alert_dedup` (`sign`, `workflow_instance_id`, `alert_type`)
 ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COLLATE = utf8_bin;
 
 -- ----------------------------

dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql

@@ -226,6 +226,7 @@ comment on column t_ds_alert.sign is 'sign=sha1(content)';
 
 create index idx_status on t_ds_alert (alert_status);
 create index idx_sign on t_ds_alert (sign);
+create unique index uk_alert_dedup on t_ds_alert (sign, workflow_instance_id, alert_type);
 
 --
 -- Table structure for table t_ds_alertgroup

dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql

@@ -0,0 +1,29 @@
+/*
+ * 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.
+*/
+
+-- Enforce idempotent task-result alerts at the database level.
+-- Allows INSERT IGNORE (MySQL) / ON CONFLICT DO NOTHING (PostgreSQL) to atomically
+-- prevent duplicates without check-then-insert race conditions.
+-- Clean up any existing duplicate rows before adding the unique constraint.
+DELETE t1 FROM t_ds_alert t1
+INNER JOIN t_ds_alert t2
+WHERE t1.id < t2.id
+  AND t1.sign = t2.sign
+  AND t1.workflow_instance_id = t2.workflow_instance_id
+  AND t1.alert_type = t2.alert_type;
+ALTER TABLE `t_ds_alert` ADD UNIQUE INDEX `uk_alert_dedup` (`sign`, `workflow_instance_id`, `alert_type`);
+

dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql

@@ -0,0 +1,16 @@
+/*
+ * 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.
+*/

dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql

@@ -0,0 +1,28 @@
+/*
+ * 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.
+*/
+
+-- Enforce idempotent task-result alerts at the database level.
+-- Allows INSERT IGNORE (MySQL) / ON CONFLICT DO NOTHING (PostgreSQL) to atomically
+-- prevent duplicates without check-then-insert race conditions.
+-- Clean up any existing duplicate rows before adding the unique constraint.
+DELETE FROM t_ds_alert a
+USING t_ds_alert b
+WHERE a.id < b.id
+  AND a.sign = b.sign
+  AND a.workflow_instance_id = b.workflow_instance_id
+  AND a.alert_type = b.alert_type;
+CREATE UNIQUE INDEX IF NOT EXISTS uk_alert_dedup ON t_ds_alert (sign, workflow_instance_id, alert_type);

dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql

@@ -0,0 +1,16 @@
+/*
+ * 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.
+*/