apache · njnu-seafish · Jul 22, 2026 · Jul 29, 2026 · Jul 29, 2026 · Jul 29, 2026
@@ -55,4 +55,6 @@ This document records the incompatible updates between each version. You need to
   * **Removed transient fields**: `stateDescList`, `workflowDefinition`, `dagData`, `queue`, `locations`, `dependenceScheduleTimes`
   * **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))
+* 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. This constraint ensures database-level 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))
 
@@ -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.

@@ -55,4 +55,6 @@
   * **移除的非数据库字段**：`stateDescList`、`workflowDefinition`、`dagData`、`queue`、`locations`、`dependenceScheduleTimes`
   * **移除的派生属性**：`cmdTypeIfComplement`、`complementData`（补数执行相关，如需获取请使用详情接口）
   * 如需获取这些字段，请使用详情接口 `GET /projects/{projectCode}/workflow-instances/{id}`，该接口仍返回完整的 `WorkflowInstance` 对象 ([#18444](https://github.com/apache/dolphinscheduler/pull/18444))
+* 为 `t_ds_alert` 表的 `(sign, workflow_instance_id, alert_type)` 新增唯一约束 `uk_alert_dedup`。升级脚本会自动清理已存在的重复行（保留 id 最大的一条），之后添加唯一索引。该约束确保任务结果告警在并发投递时数据库层面幂等。([#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))
 
@@ -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 手动清理。

@@ -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())

@@ -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,45 @@ void testSendAlert() {
                 .thenReturn(paramsMap);
         alertSender.syncTestSend(PLUGIN_DEFINE_ID, PLUGIN_INSTANCE_PARAMS);
     }
+
+    /**
+     * Simulates a mixed-version rolling upgrade scenario: an old Alert Server that
+     * does not yet know about a new AlertType enum value (e.g. TASK_RESULT) will
+     * have MyBatis map the unknown value to null. {@link AlertSender#getAlertData}
+     * must refuse to send rather than silently relabeling the alert as a
+     * workflow-instance failure.
+     */
+    @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);
+        // Simulate old Alert Server where unknown alert_type is deserialized as null
+        alert.setAlertType(null);
+
+        Assertions.assertThrows(IllegalArgumentException.class,
+                () -> alertSender.getAlertData(alert),
+                "getAlertData should refuse to build AlertData with null alertType");
+    }
+
+    /**
+     * Ensures the normal path still works correctly when alertType is present.
+     */
+    @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());
+    }
 }
@@ -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) {

@@ -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,41 @@ 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);
+        try {
+            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;
+        } catch (DuplicateKeyException e) {
+            // Concurrent race: NOT EXISTS passed but another thread inserted first.
+            // The uk_alert_dedup unique constraint caught it — treat as a skip.
+            log.info("skip duplicate task result alert (concurrent race), sign: {}, workflowInstanceId: {}", sign,
+                    alert.getWorkflowInstanceId());
+            return 0;
+        }
+    }
+
     /**
      * update alert sending(execution) status
      *
@@ -132,6 +168,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)

@@ -24,6 +24,8 @@
 import org.apache.ibatis.mapping.DatabaseIdProvider;
 import org.apache.ibatis.mapping.VendorDatabaseIdProvider;
 
+import java.util.Properties;
+
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -63,7 +65,13 @@ public MybatisPlusInterceptor paginationInterceptor(DbType dbType) {
 
     @Bean
     public DatabaseIdProvider databaseIdProvider() {
-        return new VendorDatabaseIdProvider();
+        VendorDatabaseIdProvider provider = new VendorDatabaseIdProvider();
+        Properties properties = new Properties();
+        properties.setProperty("MySQL", "mysql");
+        properties.setProperty("PostgreSQL", "postgresql");
+        properties.setProperty("H2", "h2");
+        provider.setProperties(properties);
+        return provider;
     }
 
     @Bean

@@ -84,6 +84,14 @@ public class Alert {
     @TableField("alert_type")
     private AlertType alertType;
 
+    /**
+     * Transient field used only for sign generation in idempotent task-result alert inserts.
+     * <p>Not persisted to the database — it is incorporated into the SHA-1 sign so that
+     * alerts from different tasks with identical content are not treated as duplicates.
+     */
+    @TableField(exist = false)
+    private Integer taskInstanceId;
+
     @TableField(exist = false)
     private Map<String, Object> info = new HashMap<>();
 

@@ -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);

@@ -51,6 +51,29 @@
         having count(*) = 0
     </insert>
 
+    <!--
+      Idempotent insert for task-result alerts.
+      Uses INSERT ... SELECT ... WHERE NOT EXISTS to skip duplicates
+      without dialect-specific upsert syntax (ON DUPLICATE KEY / MERGE / ON CONFLICT).
+      The uk_alert_dedup unique constraint remains as a concurrent-safety net;
+      a DuplicateKeyException is caught by the DAO layer and treated as a skip.
+    -->
+    <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}
+        WHERE NOT EXISTS (
+            SELECT 1 FROM t_ds_alert
+            WHERE sign = #{alert.sign}
+              AND workflow_instance_id = #{alert.workflowInstanceId}
+              AND alert_type = #{alert.alertType.code}
+        )
+    </insert>
+
     <select id="listingAlertByStatus" resultType="org.apache.dolphinscheduler.dao.entity.Alert">
         select
         <include refid="baseSql"/>

diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_h2.sql b/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)
 );
 
 -- ----------------------------

diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_mysql.sql b/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;
 
 -- ----------------------------

diff --git a/dolphinscheduler-dao/src/main/resources/sql/dolphinscheduler_postgresql.sql b/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

diff --git a/...nscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql b/...nscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql
@@ -20,3 +20,15 @@ ALTER TABLE `t_ds_workflow_instance` ADD INDEX idx_project_start_time (project_c
 ALTER TABLE `t_ds_schedules`
     ADD COLUMN `missed_fire_policy` tinyint NOT NULL DEFAULT '2' COMMENT 'missed fire policy: 0 skip missed, 1 fire once now, 2 fire all missed' AFTER `crontab`;
 
+-- 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`);
+
diff --git a/...duler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql b/...duler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql
@@ -19,3 +19,15 @@ CREATE INDEX idx_project_submit_time ON t_ds_task_instance (project_code ASC, su
 CREATE INDEX idx_project_start_time ON t_ds_workflow_instance (project_code ASC, start_time DESC);
 ALTER TABLE t_ds_schedules
     ADD COLUMN missed_fire_policy smallint NOT NULL DEFAULT 2;
+
+-- 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);