diff --git a/docs/docs/en/guide/upgrade/incompatible.md b/docs/docs/en/guide/upgrade/incompatible.md index 9c3325809134..756b498271ec 100644 --- a/docs/docs/en/guide/upgrade/incompatible.md +++ b/docs/docs/en/guide/upgrade/incompatible.md @@ -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)) diff --git a/docs/docs/en/guide/upgrade/upgrade.md b/docs/docs/en/guide/upgrade/upgrade.md index dfb3016a07d6..a30648dbf35c 100644 --- a/docs/docs/en/guide/upgrade/upgrade.md +++ b/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. diff --git a/docs/docs/zh/guide/upgrade/incompatible.md b/docs/docs/zh/guide/upgrade/incompatible.md index 54c0c8433fb6..a4a1c4f57282 100644 --- a/docs/docs/zh/guide/upgrade/incompatible.md +++ b/docs/docs/zh/guide/upgrade/incompatible.md @@ -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)) diff --git a/docs/docs/zh/guide/upgrade/upgrade.md b/docs/docs/zh/guide/upgrade/upgrade.md index 19ccad73cab2..d296cea68887 100644 --- a/docs/docs/zh/guide/upgrade/upgrade.md +++ b/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 手动清理。 diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/AlertSender.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/AlertSender.java index 9c9cd034bdb6..2f829e5bcd92 100644 --- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/AlertSender.java +++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/AlertSender.java @@ -100,6 +100,13 @@ public List 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()) diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java index 18246f485ab2..47bb0c506673 100644 --- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java +++ b/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,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()); + } } diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java index 058afcb3fc71..6cc850d6f141 100644 --- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java +++ b/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) { diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java index 19ed2d70f2df..cc9576c1719c 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java +++ b/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,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. + *

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) diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/DaoConfiguration.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/DaoConfiguration.java index 1b897e806de4..249a8cd7f7e6 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/DaoConfiguration.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/DaoConfiguration.java @@ -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 diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java index 96ad2890ba54..23b7f7b38970 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java @@ -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. + *

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 info = new HashMap<>(); diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java index 0d60891e55f5..dcfdf1ec0e86 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java @@ -43,6 +43,12 @@ List 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 selectByWorkflowInstanceId(@Param("workflowInstanceId") Integer processInstanceId); diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml index 7891dd91376d..9b6b1bf1b7ac 100644 --- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml +++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml @@ -51,6 +51,29 @@ having count(*) = 0 + + + 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} + ) + +