This is an automated email from the ASF dual-hosted git repository.
ruanwenjun pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 7cead6c895 [Fix-17854] [Worker&SQL Task] Fix SQL task query result
alert not being sent (#18549)
7cead6c895 is described below
commit 7cead6c895dbd0bc9e04ea2300c22c6bdff4fcb9
Author: suyc <[email protected]>
AuthorDate: Tue Oct 13 15:33:12 2026 +0800
[Fix-17854] [Worker&SQL Task] Fix SQL task query result alert not being
sent (#18549)
---
deploy/kubernetes/dolphinscheduler/README.md | 3 +
deploy/kubernetes/dolphinscheduler/values.yaml | 9 +
docs/docs/en/architecture/configuration.md | 3 +
docs/docs/en/guide/upgrade/incompatible.md | 5 +
docs/docs/en/guide/upgrade/upgrade.md | 22 ++
docs/docs/zh/architecture/configuration.md | 3 +
docs/docs/zh/guide/upgrade/incompatible.md | 5 +
docs/docs/zh/guide/upgrade/upgrade.md | 22 ++
.../alert/service/AlertSender.java | 7 +
.../alert/runner/AlertSenderTest.java | 30 ++
.../common/constants/Constants.java | 17 +
.../dolphinscheduler/common/enums/AlertType.java | 3 +-
.../src/main/resources/common.properties | 8 +
.../src/test/resources/common.properties | 8 +-
.../org/apache/dolphinscheduler/dao/AlertDao.java | 25 ++
.../apache/dolphinscheduler/dao/entity/Alert.java | 6 +
.../lifecycle/event/TaskSuccessLifecycleEvent.java | 6 +
.../handler/TaskSuccessLifecycleEventHandler.java | 24 +-
.../master/rpc/TaskExecutorEventListenerImpl.java | 2 +
.../TaskSuccessLifecycleEventHandlerTest.java | 173 +++++++++
.../service/alert/WorkflowAlertManager.java | 28 ++
.../service/alert/WorkflowAlertManagerTest.java | 123 +++++++
.../events/TaskExecutorSuccessLifecycleEvent.java | 7 +
.../plugin/task/api/AbstractTask.java | 51 ++-
.../plugin/task/api/TaskExecutionContext.java | 5 +
.../plugin/task/api/model/TaskAlertInfo.java | 5 +
.../plugin/task/api/AbstractTaskTest.java | 54 +++
.../dolphinscheduler/plugin/task/sql/SqlTask.java | 149 +++++++-
.../plugin/task/sql/SqlTaskTest.java | 393 +++++++++++++++++++++
.../worker/executor/PhysicalTaskExecutor.java | 8 +
30 files changed, 1184 insertions(+), 20 deletions(-)
diff --git a/deploy/kubernetes/dolphinscheduler/README.md
b/deploy/kubernetes/dolphinscheduler/README.md
index 69b5590a0e..99de38ec81 100644
--- a/deploy/kubernetes/dolphinscheduler/README.md
+++ b/deploy/kubernetes/dolphinscheduler/README.md
@@ -203,6 +203,9 @@ The paths above match the API and Worker server classpaths.
Check the configurat
| conf.common."resource.storage.type" | string | `"S3"` | resource storage
type: HDFS, S3, OSS, GCS, ABS, NONE |
| conf.common."resource.storage.upload.base.path" | string |
`"/dolphinscheduler"` | resource store on HDFS/S3 path, resource file will
store to this base path, self configuration, please make sure the directory
exists on hdfs and have read write permissions. "/dolphinscheduler" is
recommended |
| conf.common."shell.kill.wait.timeout" | int | `10` | If the shell process is
still active after this timeout value (in seconds), then will use kill -9 to
kill it |
+| conf.common."sql.task.alert.max.content.bytes" | int | `60000` | max
serialized UTF-8 bytes of the SQL task-result alert content, kept below the
65535-byte capacity of the MySQL TEXT column |
+| conf.common."sql.task.alert.max.row.value.length" | int | `100` | max chars
of a single string value kept in the SQL task-result alert content, longer
values are truncated with an ellipsis |
+| conf.common."sql.task.alert.max.rows" | int | `30` | max rows kept in the
SQL task-result alert content, rows beyond this are replaced by an omission
marker |
| conf.common."sudo.enable" | bool | `true` | use sudo or not, if set true,
executing user is tenant user and deploy user needs sudo permissions; if set
false, executing user is the deploy user and doesn't need sudo permissions |
| conf.common."support.hive.oneSession" | bool | `false` | Whether hive SQL is
executed in the same session |
| conf.common."task.resource.limit.state" | bool | `false` | Task resource
limit state |
diff --git a/deploy/kubernetes/dolphinscheduler/values.yaml
b/deploy/kubernetes/dolphinscheduler/values.yaml
index 048da5aa7f..8f168534e4 100644
--- a/deploy/kubernetes/dolphinscheduler/values.yaml
+++ b/deploy/kubernetes/dolphinscheduler/values.yaml
@@ -393,6 +393,15 @@ conf:
# -- way to collect applicationId: log, aop
appId.collect: log
+ # -- max rows kept in the SQL task-result alert content, rows beyond this
are replaced by an omission marker
+ sql.task.alert.max.rows: 30
+
+ # -- max chars of a single string value kept in the SQL task-result alert
content, longer values are truncated with an ellipsis
+ sql.task.alert.max.row.value.length: 100
+
+ # -- max serialized UTF-8 bytes of the SQL task-result alert content, kept
below the 65535-byte capacity of the MySQL TEXT column
+ sql.task.alert.max.content.bytes: 60000
+
common:
## Configmap
configmap:
diff --git a/docs/docs/en/architecture/configuration.md
b/docs/docs/en/architecture/configuration.md
index cb37799a85..5b5c128ba4 100644
--- a/docs/docs/en/architecture/configuration.md
+++ b/docs/docs/en/architecture/configuration.md
@@ -223,6 +223,9 @@ The default configuration is as follows:
| sudo.enable | true
| whether to enable sudo
[...]
| zeppelin.rest.url | http://localhost:8080
| the RESTful API url of zeppelin
[...]
| appId.collect | log
| way to collect applicationId, if use aop, alter the
configuration from log to aop, annotation of applicationId auto collection
related configuration in `bin/env/dolphinscheduler_env.sh` should be removed.
Note: Aop way doesn't support submitting yarn job on remote host by client mode
like Beeline, and will failure if override applicationId collection-related
environment configuration in d [...]
+| sql.task.alert.max.rows | 30
| max rows kept in the SQL task-result alert content, rows
beyond this are replaced by an omission marker. Takes effect after restarting
Worker Server
[...]
+| sql.task.alert.max.row.value.length | 100
| max chars of a single string value kept in the SQL
task-result alert content, longer values are truncated with an ellipsis. Takes
effect after restarting Worker Server
[...]
+| sql.task.alert.max.content.bytes | 60000
| max serialized UTF-8 bytes of the SQL task-result alert
content, kept below the 65535-byte capacity of the MySQL TEXT column that
stores alert content. Takes effect after restarting Worker Server
[...]
### Api-server related configuration
diff --git a/docs/docs/en/guide/upgrade/incompatible.md
b/docs/docs/en/guide/upgrade/incompatible.md
index 650b52df38..2b6649dad2 100644
--- a/docs/docs/en/guide/upgrade/incompatible.md
+++ b/docs/docs/en/guide/upgrade/incompatible.md
@@ -57,3 +57,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
+
+* `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:/ [...]
+* SQL task-result alert content is size-limited: by default at most 30 rows
are kept, string field values longer than 100 characters are truncated with an
ellipsis, and the serialized content is bounded to 60000 UTF-8 bytes (below the
65535-byte capacity of the MySQL TEXT column); rows beyond the budget are
represented by an omission marker. These defaults are tunable via
`sql.task.alert.max.rows`, `sql.task.alert.max.row.value.length` and
`sql.task.alert.max.content.bytes` in `common.pr [...]
+
diff --git a/docs/docs/en/guide/upgrade/upgrade.md
b/docs/docs/en/guide/upgrade/upgrade.md
index dfb3016a07..67b631e1fe 100644
--- a/docs/docs/en/guide/upgrade/upgrade.md
+++ b/docs/docs/en/guide/upgrade/upgrade.md
@@ -75,6 +75,28 @@ 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.
+
+#### If the upgrade fails
+
+If the upgrade order was not followed (e.g. Master/Worker has already been
upgraded before the Alert Server) and new `alert_type` values have been
persisted to the `t_ds_alert` table, follow these steps to recover:
+
+1. Upgrade the Alert Server to a version that includes the new enum values and
restart it. Alerts stuck in `WAIT_EXECUTION` will be picked up and delivered
normally.
+2. If the Alert Server cannot be upgraded immediately, stop the old Alert
Server first, then back up and clean up the affected alerts so they cannot keep
blocking alert delivery:
+
+ ```sql
+ -- 1. back up the affected alerts
+ create table t_ds_alert_bak as select * from t_ds_alert where alert_type =
8;
+ -- 2. remove them from the active table
+ delete from t_ds_alert where alert_type = 8;
+ ```
+
#### 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/architecture/configuration.md
b/docs/docs/zh/architecture/configuration.md
index a2b9437955..00012cf0ea 100644
--- a/docs/docs/zh/architecture/configuration.md
+++ b/docs/docs/zh/architecture/configuration.md
@@ -223,6 +223,9 @@ common.properties配置文件目前主要是配置hadoop/s3/yarn/applicationId
| sudo.enable | true
| 是否开启sudo
|
| zeppelin.rest.url | http://localhost:8080
| zeppelin RESTful API 接口地址
|
| appId.collect | log
| 收集applicationId方式，
如果用aop方法，将配置log替换为aop，并将`bin/env/dolphinscheduler_env.sh`自动收集applicationId相关环境变量配置的注释取消掉，注意：aop不支持远程主机提交yarn作业的方式比如Beeline客户端提交，且如果用户环境覆盖了dolphinscheduler_env.sh收集applicationId相关环境变量配置，aop方法会失效
|
+| sql.task.alert.max.rows | 30
| SQL 任务结果告警内容最多保留的行数，超出的行以省略标记表示。修改后需重启 Worker Server 生效
|
+| sql.task.alert.max.row.value.length | 100
| SQL 任务结果告警内容中单个字符串字段值最多保留的字符数，超出的值以省略号截断。修改后需重启 Worker
Server 生效
|
+| sql.task.alert.max.content.bytes | 60000
| SQL 任务结果告警内容序列化后的最大 UTF-8 字节数，小于存储告警内容的 MySQL TEXT 列
65535 字节容量。修改后需重启 Worker Server 生效
|
## Api-server相关配置
diff --git a/docs/docs/zh/guide/upgrade/incompatible.md
b/docs/docs/zh/guide/upgrade/incompatible.md
index a0f9455019..7e3de19628 100644
--- a/docs/docs/zh/guide/upgrade/incompatible.md
+++ b/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
+
+* `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))
+* SQL 任务结果告警内容有大小限制：默认最多保留 30 行，超过 100 字符的字符串字段值会被截断并以省略号标记；序列化后内容总大小限制在 60000
UTF-8 字节以内（兼容 MySQL TEXT 列 65535 字节容量），超出预算的行以省略标记对象表示。以上默认值可通过
`common.properties` 中的
`sql.task.alert.max.rows`、`sql.task.alert.max.row.value.length`、`sql.task.alert.max.content.bytes`
配置调整，修改后需重启 Worker 生效。
([#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 19ccad73ca..0e5221442e 100644
--- a/docs/docs/zh/guide/upgrade/upgrade.md
+++ b/docs/docs/zh/guide/upgrade/upgrade.md
@@ -76,6 +76,28 @@ 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"的顺序升级可消除此风险。
+
+#### 如果升级失败
+
+如果未按上述顺序升级（例如在 Alert Server 之前先升级了 Master/Worker），且新的 `alert_type` 枚举值已写入
`t_ds_alert` 表，请按以下步骤恢复：
+
+1. 将 Alert Server 升级到包含新枚举值的版本并重启，卡在 `WAIT_EXECUTION` 状态的告警会被正常拉取并投递。
+2. 如果暂时无法升级 Alert Server，请先停止旧版 Alert Server，再备份并清理受影响告警，避免其持续阻塞告警投递：
+
+ ```sql
+ -- 1. 备份受影响告警
+ create table t_ds_alert_bak as select * from t_ds_alert where alert_type =
8;
+ -- 2. 从主表清理
+ delete from t_ds_alert where alert_type = 8;
+ ```
+
#### 升级后的注意事项
在历史版本中可能告警插件会有一些脏数据，升级后请参考一下 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 9c9cd034bd..2f829e5bcd 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 class AlertSender extends
AbstractEventSender<Alert> {
@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 18246f485a..2e8f5b2672 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.verify;
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 @@ class AlertSenderTest {
.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());
+ }
}
diff --git
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java
index 81783ed285..e5815701df 100644
---
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java
+++
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java
@@ -59,6 +59,23 @@ public final class Constants {
*/
public static final String SHELL_KILL_WAIT_TIMEOUT =
"shell.kill.wait.timeout";
+ /**
+ * sql.task.alert.max.rows: this property defines the max rows kept in the
task-result alert content.
+ */
+ public static final String SQL_TASK_ALERT_MAX_ROWS =
"sql.task.alert.max.rows";
+
+ /**
+ * sql.task.alert.max.row.value.length: this property defines the max
chars of a single string value kept in the
+ * task-result alert content.
+ */
+ public static final String SQL_TASK_ALERT_MAX_ROW_VALUE_LENGTH =
"sql.task.alert.max.row.value.length";
+
+ /**
+ * sql.task.alert.max.content.bytes: this property defines the max
serialized UTF-8 bytes of the task-result alert
+ * content.
+ */
+ public static final String SQL_TASK_ALERT_MAX_CONTENT_BYTES =
"sql.task.alert.max.content.bytes";
+
public static final String SUDO_ENABLE = "sudo.enable";
public static final String RESOURCE_STORAGE_TYPE = "resource.storage.type";
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 058afcb3fc..6cc850d6f1 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-common/src/main/resources/common.properties
b/dolphinscheduler-common/src/main/resources/common.properties
index 9e5adea62e..fa397d7ce9 100644
--- a/dolphinscheduler-common/src/main/resources/common.properties
+++ b/dolphinscheduler-common/src/main/resources/common.properties
@@ -108,3 +108,11 @@ shell.interceptor.type=bash
# Whether to enable remote logging
remote.logging.enable=false
+
+# SQL task-result alert content limits, take effect on worker server restart:
+# sql.task.alert.max.rows: max rows kept in the alert content, default: 30
+# sql.task.alert.max.row.value.length: max chars of a single string value
kept in the alert content, default: 100
+# sql.task.alert.max.content.bytes: max serialized UTF-8 bytes of the alert
content, default: 60000
+sql.task.alert.max.rows=30
+sql.task.alert.max.row.value.length=100
+sql.task.alert.max.content.bytes=60000
diff --git a/dolphinscheduler-common/src/test/resources/common.properties
b/dolphinscheduler-common/src/test/resources/common.properties
index f849b1c3e0..99ec046b94 100644
--- a/dolphinscheduler-common/src/test/resources/common.properties
+++ b/dolphinscheduler-common/src/test/resources/common.properties
@@ -201,4 +201,10 @@
remote.logging.google.cloud.storage.credential=/path/to/credential
# gcs bucket name, required if you set remote.logging.target=GCS
remote.logging.google.cloud.storage.bucket.name=<your-bucket>
-
+# SQL task-result alert content limits, take effect on worker server restart:
+# sql.task.alert.max.rows: max rows kept in the alert content, default: 30
+# sql.task.alert.max.row.value.length: max chars of a single string value
kept in the alert content, default: 100
+# sql.task.alert.max.content.bytes: max serialized UTF-8 bytes of the alert
content, default: 60000
+sql.task.alert.max.rows=30
+sql.task.alert.max.row.value.length=100
+sql.task.alert.max.content.bytes=60000
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 19ed2d70f2..f5a4ab3c48 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
@@ -108,6 +108,24 @@ public class AlertDao {
return count;
}
+ /**
+ * Insert a task-result alert.
+ *
+ * @return insert count
+ */
+ 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.insert(alert);
+ log.info("add task result alert to db , alert: {}", alert);
+ return count;
+ }
+
/**
* update alert sending(execution) status
*
@@ -132,6 +150,13 @@ public class AlertDao {
* @return sign's str
*/
private String generateSign(Alert alert) {
+ // Task-result alerts include the task instance ID in the sign so that
different
+ // tasks returning identical results get different signs, and repeated
success
+ // events of the same task yield the same sign. For other alert types
+ // taskInstanceId is null and the sign falls back to content-only.
+ 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/entity/Alert.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java
index 96ad2890ba..5916bedc37 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,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<>();
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
index 9c7f41e1b5..4568768bca 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
+++
b/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 class TaskSuccessLifecycleEvent extends
AbstractTaskLifecycleEvent {
"task=" + taskExecution.getName() +
", endTime=" + endTime +
", varPool='" + varPool + '\'' +
+ ", needAlert=" + needAlert +
'}';
}
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java
index 86a6ec1799..d9e9f88f5f 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java
@@ -17,6 +17,7 @@
package
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
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;
@@ -24,18 +25,26 @@ import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifec
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
@@ -44,6 +53,19 @@ public class TaskSuccessLifecycleEventHandler extends
AbstractTaskLifecycleEvent
final ITaskExecution taskExecution,
final TaskSuccessLifecycleEvent taskSuccessEvent) {
taskStateAction.onSucceedEvent(workflowExecution, taskExecution,
taskSuccessEvent);
+
+ // Only persist the task result alert when the success event has been
accepted and the
+ // task has actually transitioned to SUCCESS. State actions that
reject a late success
+ // event (e.g. a paused or killed task) only log a warning and keep
the task in its
+ // current state, so the alert must be skipped as well.
+ if (taskExecution.getTaskInstance().getState() ==
TaskExecutionStatus.SUCCESS
+ && taskSuccessEvent.isNeedAlert()) {
+ workflowAlertManager.sendTaskResultAlert(
+ taskExecution.getWorkflowInstance(),
+ taskExecution.getTaskInstance(),
+ taskSuccessEvent.getTaskAlertInfo());
+ }
+
taskExecutorClient.ackTaskExecutorLifecycleEvent(
taskExecution,
new
ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck(
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java
index 278f5160dc..0566c83985 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java
@@ -116,6 +116,8 @@ public class TaskExecutorEventListenerImpl implements
ITaskExecutorEventListener
.taskExecution(taskExecution)
.endTime(new
Date(taskExecutorSuccessLifecycleEvent.getEndTime()))
.varPool(taskExecutorSuccessLifecycleEvent.getVarPool())
+ .needAlert(taskExecutorSuccessLifecycleEvent.isNeedAlert())
+
.taskAlertInfo(taskExecutorSuccessLifecycleEvent.getTaskAlertInfo())
.build();
taskExecution.getWorkflowEventBus().publish(taskSuccessEvent);
} finally {
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandlerTest.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandlerTest.java
new file mode 100644
index 0000000000..8545becfd9
--- /dev/null
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandlerTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.
+ */
+
+package
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
+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.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 java.util.Collections;
+import java.util.Date;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class TaskSuccessLifecycleEventHandlerTest {
+
+ @Mock
+ private TaskExecutorClient taskExecutorClient;
+
+ @Mock
+ private WorkflowAlertManager workflowAlertManager;
+
+ @Mock
+ private ITaskStateAction taskStateAction;
+
+ @Mock
+ private ITaskExecution taskExecution;
+
+ @Mock
+ private IWorkflowExecution workflowExecution;
+
+ private TaskInstance taskInstance;
+
+ private TaskSuccessLifecycleEventHandler taskSuccessLifecycleEventHandler;
+
+ @BeforeEach
+ void setUp() {
+ taskSuccessLifecycleEventHandler =
+ new TaskSuccessLifecycleEventHandler(taskExecutorClient,
workflowAlertManager);
+
+ taskInstance = new TaskInstance();
+ taskInstance.setState(TaskExecutionStatus.RUNNING_EXECUTION);
+ taskInstance.setHost("localhost");
+
+ when(taskExecution.getTaskInstance()).thenReturn(taskInstance);
+ when(taskExecution.getId()).thenReturn(1);
+ }
+
+ @Test
+ void shouldSendResultAlertWhenSucceedEventTransitionsTaskToSuccess() {
+ doAnswer(invocation -> {
+ taskInstance.setState(TaskExecutionStatus.SUCCESS);
+ return null;
+ }).when(taskStateAction)
+ .onSucceedEvent(any(), any(), any());
+
+ final TaskSuccessLifecycleEvent taskSuccessEvent =
newSuccessEvent(true);
+
+ taskSuccessLifecycleEventHandler.handle(
+ taskStateAction, workflowExecution, taskExecution,
taskSuccessEvent);
+
+ verify(workflowAlertManager).sendTaskResultAlert(
+ any(), eq(taskInstance), any());
+ verify(taskExecutorClient).ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ }
+
+ @Test
+ void shouldNotSendResultAlertWhenSucceedEventArrivesOnPausedTask() {
+ taskInstance.setState(TaskExecutionStatus.PAUSE);
+
+ final TaskSuccessLifecycleEvent taskSuccessEvent =
newSuccessEvent(true);
+
+ taskSuccessLifecycleEventHandler.handle(
+ taskStateAction, workflowExecution, taskExecution,
taskSuccessEvent);
+
+ verifyNoInteractions(workflowAlertManager);
+ verify(taskExecutorClient).ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ }
+
+ @Test
+ void shouldNotSendResultAlertWhenSucceedEventArrivesOnKilledTask() {
+ taskInstance.setState(TaskExecutionStatus.KILL);
+
+ final TaskSuccessLifecycleEvent taskSuccessEvent =
newSuccessEvent(true);
+
+ taskSuccessLifecycleEventHandler.handle(
+ taskStateAction, workflowExecution, taskExecution,
taskSuccessEvent);
+
+ verifyNoInteractions(workflowAlertManager);
+ verify(taskExecutorClient).ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ }
+
+ @Test
+ void shouldStillSendResultAlertForRepeatedSucceedEvent() {
+ taskInstance.setState(TaskExecutionStatus.SUCCESS);
+
+ final TaskSuccessLifecycleEvent taskSuccessEvent =
newSuccessEvent(true);
+
+ taskSuccessLifecycleEventHandler.handle(
+ taskStateAction, workflowExecution, taskExecution,
taskSuccessEvent);
+
+ // Repeated success events are accepted idempotently: the alert is
sent again,
+ // which may insert a duplicate row; occasional duplicates are
accepted.
+ verify(workflowAlertManager).sendTaskResultAlert(
+ any(), eq(taskInstance), any());
+ verify(taskExecutorClient).ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ }
+
+ @Test
+ void shouldNotSendResultAlertWhenNeedAlertIsFalse() {
+ doAnswer(invocation -> {
+ taskInstance.setState(TaskExecutionStatus.SUCCESS);
+ return null;
+ }).when(taskStateAction)
+ .onSucceedEvent(any(), any(), any());
+
+ final TaskSuccessLifecycleEvent taskSuccessEvent =
newSuccessEvent(false);
+
+ taskSuccessLifecycleEventHandler.handle(
+ taskStateAction, workflowExecution, taskExecution,
taskSuccessEvent);
+
+ verifyNoInteractions(workflowAlertManager);
+ verify(taskExecutorClient).ackTaskExecutorLifecycleEvent(
+ eq(taskExecution),
any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class));
+ }
+
+ private TaskSuccessLifecycleEvent newSuccessEvent(final boolean needAlert)
{
+ return TaskSuccessLifecycleEvent.builder()
+ .taskExecution(taskExecution)
+ .endTime(new Date())
+ .varPool(Collections.emptyList())
+ .needAlert(needAlert)
+ .taskAlertInfo(new TaskAlertInfo())
+ .build();
+ }
+}
diff --git
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java
index 9335e9551e..cba5179b6e 100644
---
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java
+++
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java
@@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.service.alert;
+import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.Flag;
@@ -34,6 +35,7 @@ import
org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
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 +210,30 @@ public class WorkflowAlertManager {
alertDao.sendWorkflowTimeoutAlert(workflowInstance, projectUser);
}
+ public void sendTaskResultAlert(WorkflowInstance workflowInstance,
+ TaskInstance taskInstance,
+ TaskAlertInfo taskAlertInfo) {
+ if (taskAlertInfo == null || taskAlertInfo.getAlertGroupId() == null
+ || taskAlertInfo.getAlertGroupId() <= 0) {
+ return;
+ }
+
+ Alert alert = new Alert();
+ alert.setTitle(taskAlertInfo.getTitle());
+ alert.setContent(taskAlertInfo.getContent());
+ alert.setWarningType(WarningType.SUCCESS);
+ alert.setAlertStatus(AlertStatus.WAIT_EXECUTION);
+ 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);
+ alert.setTaskInstanceId(taskInstance.getId());
+ alertDao.addTaskResultAlert(alert);
+ log.info("Send task result alert for task: {} in workflow: {}",
+ taskInstance.getName(), workflowInstance.getName());
+ }
}
diff --git
a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManagerTest.java
b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManagerTest.java
new file mode 100644
index 0000000000..0fcdfe0cbc
--- /dev/null
+++
b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManagerTest.java
@@ -0,0 +1,123 @@
+/*
+ * 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.
+ */
+
+package org.apache.dolphinscheduler.service.alert;
+
+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.entity.Alert;
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+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 org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class WorkflowAlertManagerTest {
+
+ @Mock
+ private AlertDao alertDao;
+
+ @Mock
+ private WorkflowDefinitionLogDao workflowDefinitionLogDao;
+
+ @Mock
+ private UserDao userDao;
+
+ @Mock
+ private ProjectDao projectDao;
+
+ @InjectMocks
+ private WorkflowAlertManager workflowAlertManager;
+
+ @Test
+ void testSendTaskResultAlertSetsWaitExecutionStatus() {
+ WorkflowInstance workflowInstance = new WorkflowInstance();
+ workflowInstance.setId(1001);
+ workflowInstance.setName("test-workflow");
+ workflowInstance.setProjectCode(2001L);
+ workflowInstance.setWorkflowDefinitionCode(3001L);
+
+ TaskInstance taskInstance = new TaskInstance();
+ taskInstance.setId(5001);
+ taskInstance.setName("sql-query-task");
+
+ TaskAlertInfo taskAlertInfo = new TaskAlertInfo();
+ taskAlertInfo.setTitle("SQL Task Result");
+
taskAlertInfo.setContent("[{\"taskName\":\"sql-query-task\",\"result\":\"ok\"}]");
+ taskAlertInfo.setAlertGroupId(1);
+ taskAlertInfo.setAlertType(AlertType.TASK_RESULT);
+
+
Mockito.when(alertDao.addTaskResultAlert(Mockito.any(Alert.class))).thenReturn(1);
+
+ workflowAlertManager.sendTaskResultAlert(workflowInstance,
taskInstance, taskAlertInfo);
+
+ ArgumentCaptor<Alert> alertCaptor =
ArgumentCaptor.forClass(Alert.class);
+ Mockito.verify(alertDao).addTaskResultAlert(alertCaptor.capture());
+
+ Alert captured = alertCaptor.getValue();
+ Assertions.assertEquals(AlertStatus.WAIT_EXECUTION,
captured.getAlertStatus());
+ Assertions.assertEquals("SQL Task Result", captured.getTitle());
+ Assertions.assertEquals(WarningType.SUCCESS,
captured.getWarningType());
+ Assertions.assertEquals(AlertType.TASK_RESULT,
captured.getAlertType());
+ Assertions.assertEquals(1001, captured.getWorkflowInstanceId());
+ Assertions.assertEquals(2001L, captured.getProjectCode());
+ Assertions.assertEquals(3001L, captured.getWorkflowDefinitionCode());
+ Assertions.assertEquals(1, captured.getAlertGroupId());
+ Assertions.assertNotNull(captured.getCreateTime());
+ Assertions.assertEquals(5001, captured.getTaskInstanceId());
+ }
+
+ @Test
+ void testSendTaskResultAlertNullTaskAlertInfoDoesNothing() {
+ WorkflowInstance workflowInstance = new WorkflowInstance();
+ workflowInstance.setId(1001);
+ TaskInstance taskInstance = new TaskInstance();
+
+ workflowAlertManager.sendTaskResultAlert(workflowInstance,
taskInstance, null);
+
+ Mockito.verifyNoInteractions(alertDao);
+ }
+
+ @Test
+ void testSendTaskResultAlertNullAlertGroupIdDoesNothing() {
+ WorkflowInstance workflowInstance = new WorkflowInstance();
+ workflowInstance.setId(1001);
+ TaskInstance taskInstance = new TaskInstance();
+
+ TaskAlertInfo taskAlertInfo = new TaskAlertInfo();
+ taskAlertInfo.setTitle("SQL Task Result");
+ taskAlertInfo.setContent("[]");
+ taskAlertInfo.setAlertGroupId(null);
+
+ workflowAlertManager.sendTaskResultAlert(workflowInstance,
taskInstance, taskAlertInfo);
+
+ Mockito.verifyNoInteractions(alertDao);
+ }
+}
diff --git
a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
index 947d1edf36..2c55c80c36 100644
---
a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
+++
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
@@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.task.executor.events;
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 class TaskExecutorSuccessLifecycleEvent extends
AbstractTaskExecutorLifec
.varPool(taskExecutionContext.getVarPool())
.type(TaskExecutorLifecycleEventType.SUCCESS)
.endTime(taskExecutionContext.getEndTime())
+ .needAlert(taskExecutionContext.isNeedAlert())
+ .taskAlertInfo(taskExecutionContext.getTaskAlertInfo())
.build();
}
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
index 0c8f208fcb..afa990e7bb 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
@@ -56,8 +56,22 @@ public abstract class AbstractTask {
*/
protected volatile int exitStatusCode = -1;
+ /**
+ * @deprecated Use {@link TaskExecutionContext#isNeedAlert()} on the
+ * {@code taskRequest} field instead. Retained for binary
+ * compatibility with third-party plugins compiled against
+ * the previous AbstractTask API.
+ */
+ @Deprecated
protected boolean needAlert = false;
+ /**
+ * @deprecated Use {@link TaskExecutionContext#getTaskAlertInfo()} on the
+ * {@code taskRequest} field instead. Retained for binary
+ * compatibility with third-party plugins compiled against
+ * the previous AbstractTask API.
+ */
+ @Deprecated
protected TaskAlertInfo taskAlertInfo;
/**
@@ -109,20 +123,53 @@ public abstract class AbstractTask {
this.appIds = appIds;
}
+ /**
+ * @deprecated Use {@link TaskExecutionContext#isNeedAlert()} on the
+ * {@code taskRequest} field instead. This method is retained
as a
+ * compatibility bridge so that third-party plugins compiled
against
+ * the previous AbstractTask API continue to work. The value
is merged
+ * with the legacy field so that direct field writes from such
plugins
+ * are still honored.
+ */
+ @Deprecated
public boolean getNeedAlert() {
- return needAlert;
+ return needAlert || (taskRequest != null && taskRequest.isNeedAlert());
}
+ /**
+ * @deprecated Use {@link TaskExecutionContext#setNeedAlert(boolean)} on
the
+ * {@code taskRequest} field instead.
+ */
+ @Deprecated
public void setNeedAlert(boolean needAlert) {
this.needAlert = needAlert;
+ if (taskRequest != null) {
+ taskRequest.setNeedAlert(needAlert);
+ }
}
+ /**
+ * @deprecated Use {@link TaskExecutionContext#getTaskAlertInfo()} on the
+ * {@code taskRequest} field instead.
+ */
+ @Deprecated
public TaskAlertInfo getTaskAlertInfo() {
- return taskAlertInfo;
+ if (taskAlertInfo != null) {
+ return taskAlertInfo;
+ }
+ return taskRequest != null ? taskRequest.getTaskAlertInfo() : null;
}
+ /**
+ * @deprecated Use {@link
TaskExecutionContext#setTaskAlertInfo(TaskAlertInfo)}
+ * on the {@code taskRequest} field instead.
+ */
+ @Deprecated
public void setTaskAlertInfo(TaskAlertInfo taskAlertInfo) {
this.taskAlertInfo = taskAlertInfo;
+ if (taskRequest != null) {
+ taskRequest.setTaskAlertInfo(taskAlertInfo);
+ }
}
/**
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
index 85d6f68219..a072258011 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
@@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.plugin.task.api;
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;
}
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java
index e72242e934..d2c22e565a 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java
@@ -17,6 +17,8 @@
package org.apache.dolphinscheduler.plugin.task.api.model;
+import org.apache.dolphinscheduler.common.enums.AlertType;
+
import lombok.Data;
@Data
@@ -27,4 +29,7 @@ public class TaskAlertInfo {
private String content;
private Integer alertGroupId;
+
+ private AlertType alertType;
+
}
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTaskTest.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTaskTest.java
index 5421c05575..f3a8a8b534 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTaskTest.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTaskTest.java
@@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.plugin.task.api;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
import
org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
import
org.apache.dolphinscheduler.plugin.task.api.parameters.SubWorkflowParameters;
@@ -62,12 +63,65 @@ public class AbstractTaskTest {
Assertions.assertEquals(TaskExecutionStatus.FAILURE,
task.getExitStatus());
}
+ @Test
+ public void testDirectFieldAssignmentIsExposedByGetters() {
+ AbstractTask task = new TestTask();
+ // legacy plugins may assign the protected fields directly
+ task.needAlert = true;
+ TaskAlertInfo alertInfo = new TaskAlertInfo();
+ alertInfo.setTitle("legacy alert");
+ task.taskAlertInfo = alertInfo;
+
+ Assertions.assertTrue(task.getNeedAlert());
+ Assertions.assertSame(alertInfo, task.getTaskAlertInfo());
+ }
+
+ @Test
+ public void testSettersWriteBothFieldAndContext() {
+ TaskExecutionContext ctx = new TaskExecutionContext();
+ AbstractTask task = new TestTask(ctx);
+ task.setNeedAlert(true);
+ TaskAlertInfo alertInfo = new TaskAlertInfo();
+ alertInfo.setTitle("setter alert");
+ task.setTaskAlertInfo(alertInfo);
+
+ Assertions.assertTrue(ctx.isNeedAlert());
+ Assertions.assertSame(alertInfo, ctx.getTaskAlertInfo());
+ Assertions.assertTrue(task.getNeedAlert());
+ Assertions.assertSame(alertInfo, task.getTaskAlertInfo());
+ }
+
+ @Test
+ public void testGettersFallBackToContextWhenFieldNotSet() {
+ TaskExecutionContext ctx = new TaskExecutionContext();
+ AbstractTask task = new TestTask(ctx);
+ ctx.setNeedAlert(true);
+ TaskAlertInfo alertInfo = new TaskAlertInfo();
+ alertInfo.setTitle("context alert");
+ ctx.setTaskAlertInfo(alertInfo);
+
+ Assertions.assertTrue(task.getNeedAlert());
+ Assertions.assertSame(alertInfo, task.getTaskAlertInfo());
+ }
+
+ @Test
+ public void testGettersReturnDefaultsWhenNothingSet() {
+ AbstractTask task = new TestTask();
+
+ Assertions.assertFalse(task.getNeedAlert());
+ Assertions.assertNull(task.getTaskAlertInfo());
+ }
+
private static final class TestTask extends AbstractTask {
private TestTask() {
super(new TaskExecutionContext());
}
+ private TestTask(TaskExecutionContext taskExecutionContext) {
+ super(taskExecutionContext);
+ }
+
@Override
public void handle(TaskCallBack taskCallBack) {
}
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
index 113dfbcd9e..12c0cce323 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
@@ -17,8 +17,11 @@
package org.apache.dolphinscheduler.plugin.task.sql;
+import org.apache.dolphinscheduler.common.constants.Constants;
+import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import
org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClientProvider;
import
org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourcePluginManager;
import org.apache.dolphinscheduler.plugin.datasource.api.utils.DataSourceUtils;
@@ -56,6 +59,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -66,6 +70,7 @@ import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -90,6 +95,34 @@ public class SqlTask extends AbstractTask {
*/
private static final int QUERY_LIMIT = 10000;
+ /**
+ * default max rows kept in the task-result alert content; rows beyond
this are replaced by an omission marker,
+ * overridable via the sql.task.alert.max.rows property in
common.properties
+ */
+ private static final int DEFAULT_ALERT_MAX_ROWS = 30;
+
+ /**
+ * default max length of a single string field value kept in the alert
content; longer values are truncated
+ * with an ellipsis, overridable via the
sql.task.alert.max.row.value.length property in common.properties
+ */
+ private static final int DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH = 100;
+
+ /**
+ * default max serialized UTF-8 bytes of the alert content, kept below the
65535-byte capacity
+ * of the MySQL TEXT column that stores t_ds_alert.content, overridable
via the
+ * sql.task.alert.max.content.bytes property in common.properties
+ */
+ private static final int DEFAULT_ALERT_MAX_CONTENT_BYTES = 60000;
+
+ private final int alertMaxRows =
+ PropertyUtils.getInt(Constants.SQL_TASK_ALERT_MAX_ROWS,
DEFAULT_ALERT_MAX_ROWS);
+
+ private final int alertMaxRowValueLength =
+
PropertyUtils.getInt(Constants.SQL_TASK_ALERT_MAX_ROW_VALUE_LENGTH,
DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH);
+
+ private final int alertMaxContentBytes =
+ PropertyUtils.getInt(Constants.SQL_TASK_ALERT_MAX_CONTENT_BYTES,
DEFAULT_ALERT_MAX_CONTENT_BYTES);
+
private final SQLTaskExecutionContext sqlTaskExecutionContext;
private final DbType dbType;
@@ -287,13 +320,18 @@ public class SqlTask extends AbstractTask {
}
}
- String result = resultJSONArray.isEmpty() ?
JSONUtils.toJsonString(generateEmptyRow(resultSet))
- : JSONUtils.toJsonString(resultJSONArray);
+ String result;
+ ArrayNode alertArray = resultJSONArray;
+ if (resultJSONArray.isEmpty()) {
+ ArrayNode emptyRow = generateEmptyRow(resultSet);
+ result = JSONUtils.toJsonString(emptyRow);
+ alertArray = emptyRow;
+ } else {
+ result = JSONUtils.toJsonString(resultJSONArray);
+ }
if (Boolean.TRUE.equals(sqlParameters.getSendEmail())) {
- sendAttachment(sqlParameters.getGroupId(),
StringUtils.isNotEmpty(sqlParameters.getTitle())
- ? sqlParameters.getTitle()
- : taskExecutionContext.getTaskName() + " query result
sets", result);
+ prepareTaskResultAlert(alertArray);
}
log.debug("execute sql result : {}", result);
return result;
@@ -320,18 +358,99 @@ public class SqlTask extends AbstractTask {
}
/**
- * send alert as an attachment
- *
- * @param title title
- * @param content content
+ * Prepare task result alert info with the complete query result as
content.
*/
- private void sendAttachment(int groupId, String title, String content) {
- setNeedAlert(Boolean.TRUE);
+ private void prepareTaskResultAlert(ArrayNode resultJSONArray) {
TaskAlertInfo taskAlertInfo = new TaskAlertInfo();
- taskAlertInfo.setAlertGroupId(groupId);
- taskAlertInfo.setContent(content);
- taskAlertInfo.setTitle(title);
- setTaskAlertInfo(taskAlertInfo);
+ taskAlertInfo.setAlertGroupId(sqlParameters.getGroupId());
+ taskAlertInfo.setTitle(StringUtils.isNotEmpty(sqlParameters.getTitle())
+ ? sqlParameters.getTitle()
+ : taskExecutionContext.getTaskName() + " query result sets");
+ taskAlertInfo.setContent(limitAlertContent(resultJSONArray));
+ taskAlertInfo.setAlertType(AlertType.TASK_RESULT);
+
+ taskExecutionContext.setNeedAlert(true);
+ taskExecutionContext.setTaskAlertInfo(taskAlertInfo);
+ log.debug("Prepare task result alert: title={}, alertGroupId={},
alertType={}, totalRows={}",
+ taskAlertInfo.getTitle(), taskAlertInfo.getAlertGroupId(),
taskAlertInfo.getAlertType(),
+ resultJSONArray.size());
+ }
+
+ /**
+ * Build a size-limited alert content that stays a valid JSON array.
+ * At most the configured max rows are kept; string field values longer
than the configured
+ * max value length are truncated with an ellipsis. The serialized content
is bounded by
+ * the configured max UTF-8 bytes: rows are dropped from the tail until it
fits; a row that
+ * alone exceeds the budget is replaced by the omission marker.
+ */
+ private String limitAlertContent(ArrayNode resultJSONArray) {
+ int totalRows = resultJSONArray.size();
+ ArrayNode limited = JSONUtils.createArrayNode();
+ int rowsToKeep = Math.min(totalRows, alertMaxRows);
+ for (int i = 0; i < rowsToKeep; i++) {
+ limited.add(limitRowValues(resultJSONArray.get(i)));
+ }
+
+ int omittedRows = totalRows - rowsToKeep;
+ while (true) {
+ if (omittedRows > 0) {
+ limited.add(createOmissionMarker(omittedRows));
+ }
+ String content = JSONUtils.toJsonString(limited);
+ if (content.getBytes(StandardCharsets.UTF_8).length <=
alertMaxContentBytes) {
+ return content;
+ }
+ if (omittedRows > 0) {
+ limited.remove(limited.size() - 1);
+ }
+ if (limited.isEmpty()) {
+ // Even a lone marker exceeds the budget (pathological
budget); return the minimal notice.
+ return
JSONUtils.toJsonString(createOmissionMarkerArray(totalRows));
+ }
+ limited.remove(limited.size() - 1);
+ omittedRows++;
+ }
+ }
+
+ private ObjectNode createOmissionMarker(int omittedRows) {
+ ObjectNode omitted = JSONUtils.createObjectNode();
+ omitted.put("...", omittedRows + " more rows omitted");
+ return omitted;
+ }
+
+ private ArrayNode createOmissionMarkerArray(int totalRows) {
+ ArrayNode markerArray = JSONUtils.createArrayNode();
+ markerArray.add(createOmissionMarker(totalRows));
+ return markerArray;
+ }
+
+ private JsonNode limitRowValues(JsonNode row) {
+ if (!row.isObject()) {
+ return row;
+ }
+ ObjectNode limitedRow = JSONUtils.createObjectNode();
+ Iterator<Map.Entry<String, JsonNode>> fields = row.fields();
+ while (fields.hasNext()) {
+ Map.Entry<String, JsonNode> entry = fields.next();
+ JsonNode value = entry.getValue();
+ if (value.isTextual() && value.textValue().length() >
alertMaxRowValueLength) {
+ limitedRow.put(entry.getKey(),
+ truncateAtCodePointBoundary(value.textValue(),
alertMaxRowValueLength) + "...");
+ } else {
+ limitedRow.set(entry.getKey(), value);
+ }
+ }
+ return limitedRow;
+ }
+
+ /**
+ * Cut a string to at most maxChars characters without splitting a
surrogate pair.
+ */
+ private String truncateAtCodePointBoundary(String text, int maxChars) {
+ if (text.codePointCount(0, text.length()) <= maxChars) {
+ return text;
+ }
+ return text.substring(0, text.offsetByCodePoints(0, maxChars));
}
private String executeQuery(Connection connection, SqlBinds sqlBinds,
String handlerType) throws Exception {
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java
index 92f0fa0702..7dc36bc9e2 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java
@@ -20,8 +20,10 @@ package org.apache.dolphinscheduler.plugin.task.sql;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import org.apache.dolphinscheduler.common.enums.AlertType;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
import org.apache.dolphinscheduler.plugin.task.api.TaskException;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
@@ -29,6 +31,7 @@ import
org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
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.SqlParameters;
import
org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
import
org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
@@ -36,6 +39,7 @@ import
org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext;
import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
import org.apache.dolphinscheduler.spi.enums.DbType;
+import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
@@ -50,7 +54,10 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.Lists;
@@ -489,4 +496,390 @@ class SqlTaskTest {
return resourceParametersHelper;
}
+ private TaskExecutionContext createSqlTaskWithAlert(ArrayNode resultArray,
int displayRows, String title,
+ int groupId) {
+ SqlParameters sqlParameters = new SqlParameters();
+ sqlParameters.setType("HIVE");
+ sqlParameters.setDatasource(1);
+ sqlParameters.setSql("select 1");
+ sqlParameters.setSendEmail(true);
+ sqlParameters.setDisplayRows(displayRows);
+ sqlParameters.setGroupId(groupId);
+ if (title != null) {
+ sqlParameters.setTitle(title);
+ }
+
+ TaskExecutionContext ctx = new TaskExecutionContext();
+ ctx.setTaskParams(JSONUtils.toJsonString(sqlParameters));
+ ctx.setTaskName("test_sql_task");
+
ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE));
+
+ SqlTask task = new SqlTask(ctx);
+ try {
+ Method method =
SqlTask.class.getDeclaredMethod("prepareTaskResultAlert", ArrayNode.class);
+ method.setAccessible(true);
+ method.invoke(task, resultArray);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return ctx;
+ }
+
+ @Test
+ void testPrepareTaskResultAlertSetsNeedAlertAndTaskAlertInfo() {
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+
resultArray.add(JSONUtils.parseObject("{\"id\":\"1\",\"name\":\"alice\"}"));
+
resultArray.add(JSONUtils.parseObject("{\"id\":\"2\",\"name\":\"bob\"}"));
+
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "My
Alert Title", 5);
+
+ Assertions.assertTrue(ctx.isNeedAlert());
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ Assertions.assertNotNull(alertInfo);
+ Assertions.assertEquals(5, alertInfo.getAlertGroupId());
+ Assertions.assertEquals("My Alert Title", alertInfo.getTitle());
+ Assertions.assertEquals(AlertType.TASK_RESULT,
alertInfo.getAlertType());
+ Assertions.assertNotNull(alertInfo.getContent());
+
+ ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+ Assertions.assertEquals(2, contentArray.size());
+ }
+
+ @Test
+ void testPrepareTaskResultAlertKeepsFullContentRegardlessOfDisplayRows()
throws Exception {
+ int totalRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS") - 1;
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ for (int i = 0; i < totalRows; i++) {
+ resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}"));
+ }
+
+ // displayRows is only a log-display setting, the alert content must
keep all rows
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 3,
null, 1);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ Assertions.assertNotNull(alertInfo);
+
+ ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+ Assertions.assertEquals(totalRows, contentArray.size());
+ Assertions.assertEquals("0", contentArray.get(0).get("id").asText());
+ Assertions.assertEquals(String.valueOf(totalRows - 1),
contentArray.get(totalRows - 1).get("id").asText());
+ }
+
+ @Test
+ void testPrepareTaskResultAlertKeepsFullContentWhenDisplayRowsUnset()
throws Exception {
+ int totalRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS") - 1;
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ for (int i = 0; i < totalRows; i++) {
+ resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}"));
+ }
+
+ // displayRows = 0 means unset, the alert content still keeps all rows
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 0,
"default rows test", 2);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ Assertions.assertNotNull(alertInfo);
+
+ ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+ Assertions.assertEquals(totalRows, contentArray.size());
+ }
+
+ @Test
+ void testPrepareTaskResultAlertUsesCustomTitle() {
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ resultArray.add(JSONUtils.parseObject("{\"id\":\"1\"}"));
+
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10,
"Custom Title", 1);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ Assertions.assertNotNull(alertInfo);
+ Assertions.assertEquals("Custom Title", alertInfo.getTitle());
+ }
+
+ @Test
+ void testPrepareTaskResultAlertUsesDefaultTitleWhenEmpty() {
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ resultArray.add(JSONUtils.parseObject("{\"id\":\"1\"}"));
+
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10,
null, 1);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ Assertions.assertNotNull(alertInfo);
+ Assertions.assertEquals("test_sql_task query result sets",
alertInfo.getTitle());
+ }
+
+ @Test
+ void testPrepareTaskResultAlertTruncatesRowsBeyondMax() throws Exception {
+ int maxRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS");
+ int totalRows = maxRows + 10;
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ for (int i = 0; i < totalRows; i++) {
+ resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}"));
+ }
+
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10,
"truncate rows test", 1);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+ Assertions.assertEquals(maxRows + 1, contentArray.size());
+ Assertions.assertEquals("0", contentArray.get(0).get("id").asText());
+ JsonNode last = contentArray.get(contentArray.size() - 1);
+ Assertions.assertTrue(last.has("..."));
+ Assertions.assertEquals((totalRows - maxRows) + " more rows omitted",
last.get("...").asText());
+ }
+
+ @Test
+ void testPrepareTaskResultAlertTruncatesLongFieldValue() throws Exception {
+ int maxLen =
readStaticIntConstant("DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH");
+ StringBuilder longValue = new StringBuilder();
+ for (int i = 0; i < maxLen + 50; i++) {
+ longValue.append("x");
+ }
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ resultArray.add(JSONUtils.parseObject("{\"id\":\"1\",\"payload\":\"" +
longValue + "\"}"));
+
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10,
"truncate field test", 1);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+ Assertions.assertEquals(1, contentArray.size());
+ String payload = contentArray.get(0).get("payload").asText();
+ Assertions.assertEquals(maxLen + 3, payload.length());
+ Assertions.assertTrue(payload.endsWith("..."));
+ }
+
+ @Test
+ void testPrepareTaskResultAlertBoundsTotalBytesWithWideRows() throws
Exception {
+ int maxRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS");
+ int maxLen =
readStaticIntConstant("DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH");
+ int budget = readStaticIntConstant("DEFAULT_ALERT_MAX_CONTENT_BYTES");
+
+ // Wide rows: enough columns that keeping all rows overflows the byte
budget
+ // even though every single value stays under the per-value character
limit.
+ int columns = budget / (maxRows * (maxLen - 1)) + 5;
+ int totalRows = maxRows;
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ for (int i = 0; i < totalRows; i++) {
+ ObjectNode row = JSONUtils.createObjectNode();
+ for (int c = 0; c < columns; c++) {
+ row.put("col_" + c, repeatChar('v', maxLen - 1));
+ }
+ resultArray.add(row);
+ }
+
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10,
"wide rows test", 1);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ int contentBytes =
alertInfo.getContent().getBytes(StandardCharsets.UTF_8).length;
+ Assertions.assertTrue(contentBytes <= budget,
+ "serialized alert content must stay within the byte budget: "
+ contentBytes);
+
+ ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+ int keptRows = contentArray.size() - 1;
+ Assertions.assertTrue(keptRows < totalRows, "rows must be dropped from
the tail to fit the budget");
+ Assertions.assertTrue(contentArray.get(0).has("col_0"));
+ JsonNode last = contentArray.get(contentArray.size() - 1);
+ Assertions.assertTrue(last.has("..."));
+ Assertions.assertEquals((totalRows - keptRows) + " more rows omitted",
last.get("...").asText());
+ }
+
+ @Test
+ void testPrepareTaskResultAlertBoundsTotalBytesWithMultibyteText() throws
Exception {
+ int maxRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS");
+ int maxLen =
readStaticIntConstant("DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH");
+ int budget = readStaticIntConstant("DEFAULT_ALERT_MAX_CONTENT_BYTES");
+
+ // Each character is 3 UTF-8 bytes, so rows within the character
limits still
+ // overflow the byte budget and rows must be dropped from the tail.
+ char cjk = '\u4e2d';
+ int columns = budget / (maxRows * 3 * maxLen) + 5;
+ int totalRows = maxRows;
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ for (int i = 0; i < totalRows; i++) {
+ ObjectNode row = JSONUtils.createObjectNode();
+ for (int c = 0; c < columns; c++) {
+ row.put("col_" + c, repeatChar(cjk, maxLen));
+ }
+ resultArray.add(row);
+ }
+
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10,
"multibyte test", 1);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ int contentBytes =
alertInfo.getContent().getBytes(StandardCharsets.UTF_8).length;
+ Assertions.assertTrue(contentBytes <= budget,
+ "multibyte alert content must stay within the byte budget: " +
contentBytes);
+
+ ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+ int keptRows = contentArray.size() - 1;
+ Assertions.assertTrue(keptRows < totalRows, "rows must be dropped from
the tail to fit the budget");
+ // Kept values are exactly at the character limit, so they must stay
complete.
+ String firstValue = contentArray.get(0).get("col_0").asText();
+ Assertions.assertEquals(repeatChar(cjk, maxLen), firstValue);
+ JsonNode last = contentArray.get(contentArray.size() - 1);
+ Assertions.assertTrue(last.has("..."));
+ Assertions.assertEquals((totalRows - keptRows) + " more rows omitted",
last.get("...").asText());
+ }
+
+ @Test
+ void testPrepareTaskResultAlertReplacesRowExceedingByteBudgetAlone()
throws Exception {
+ int budget = readStaticIntConstant("DEFAULT_ALERT_MAX_CONTENT_BYTES");
+
+ // Nested values are not truncated per-field, so a single row can
exceed the
+ // budget alone; it must be replaced by the omission marker.
+ ObjectNode row = JSONUtils.createObjectNode();
+ row.put("id", "1");
+ ObjectNode payload = JSONUtils.createObjectNode();
+ payload.put("blob", repeatChar('n', budget * 2));
+ row.set("payload", payload);
+
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ resultArray.add(row);
+
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10,
"huge nested value test", 1);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ int contentBytes =
alertInfo.getContent().getBytes(StandardCharsets.UTF_8).length;
+ Assertions.assertTrue(contentBytes <= budget);
+
+ ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+ Assertions.assertEquals(1, contentArray.size());
+ JsonNode only = contentArray.get(0);
+ Assertions.assertTrue(only.has("..."));
+ Assertions.assertEquals("1 more rows omitted",
only.get("...").asText());
+ }
+
+ private String repeatChar(char c, int count) {
+ StringBuilder builder = new StringBuilder(count);
+ for (int i = 0; i < count; i++) {
+ builder.append(c);
+ }
+ return builder.toString();
+ }
+
+ @Test
+ void testPrepareTaskResultAlertRespectsConfiguredLimits() throws Exception
{
+ int configuredMaxRows = 3;
+ int configuredMaxValueLength = 10;
+ int configuredMaxBytes = 60000;
+ int totalRows = 10;
+
+ ArrayNode resultArray = JSONUtils.createArrayNode();
+ for (int i = 0; i < totalRows; i++) {
+ resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}"));
+ }
+
+ try (MockedStatic<PropertyUtils> propertyUtils =
Mockito.mockStatic(PropertyUtils.class)) {
+ propertyUtils.when(() ->
PropertyUtils.getInt("sql.task.alert.max.rows",
30)).thenReturn(configuredMaxRows);
+ propertyUtils.when(() ->
PropertyUtils.getInt("sql.task.alert.max.row.value.length", 100))
+ .thenReturn(configuredMaxValueLength);
+ propertyUtils.when(() ->
PropertyUtils.getInt("sql.task.alert.max.content.bytes", 60000))
+ .thenReturn(configuredMaxBytes);
+
+ TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10,
"configured limits test", 1);
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ ArrayNode contentArray =
JSONUtils.parseArray(alertInfo.getContent());
+ Assertions.assertEquals(configuredMaxRows + 1,
contentArray.size());
+ JsonNode last = contentArray.get(contentArray.size() - 1);
+ Assertions.assertTrue(last.has("..."));
+ Assertions.assertEquals((totalRows - configuredMaxRows) + " more
rows omitted",
+ last.get("...").asText());
+ }
+ }
+
+ private int readStaticIntConstant(String name) throws Exception {
+ Field field = SqlTask.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return field.getInt(null);
+ }
+
+ @Test
+ void testResultProcessSendEmailDisabledDoesNotSetNeedAlert() throws
Exception {
+ // Build a SqlTask with sendEmail = false
+ String taskParams =
"{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\",\"sendEmail\":false}";
+ TaskExecutionContext ctx = new TaskExecutionContext();
+ ctx.setTaskParams(taskParams);
+ ctx.setTaskName("no_alert_task");
+
ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE));
+
+ SqlTask task = new SqlTask(ctx);
+
+ ResultSet mockResultSet = mock(ResultSet.class);
+ ResultSetMetaData mockMetaData = mock(ResultSetMetaData.class);
+ when(mockResultSet.getMetaData()).thenReturn(mockMetaData);
+ when(mockMetaData.getColumnCount()).thenReturn(1);
+ when(mockMetaData.getColumnLabel(1)).thenReturn("id");
+ when(mockResultSet.next()).thenReturn(true, false);
+ when(mockResultSet.getObject(1)).thenReturn("1");
+
+ Method resultProcessMethod =
SqlTask.class.getDeclaredMethod("resultProcess", ResultSet.class);
+ resultProcessMethod.setAccessible(true);
+ resultProcessMethod.invoke(task, mockResultSet);
+
+ Assertions.assertFalse(ctx.isNeedAlert());
+ Assertions.assertNull(ctx.getTaskAlertInfo());
+ }
+
+ @Test
+ void testResultProcessEmptyResultSetPreparesAlertWithEmptyRow() throws
Exception {
+ String taskParams =
"{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\""
+ +
",\"sendEmail\":true,\"displayRows\":10,\"groupId\":3,\"title\":\"empty
result\"}";
+
+ TaskExecutionContext ctx = new TaskExecutionContext();
+ ctx.setTaskParams(taskParams);
+ ctx.setTaskName("empty_result_task");
+
ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE));
+
+ SqlTask task = new SqlTask(ctx);
+
+ ResultSet mockResultSet = mock(ResultSet.class);
+ ResultSetMetaData mockMetaData = mock(ResultSetMetaData.class);
+ when(mockResultSet.getMetaData()).thenReturn(mockMetaData);
+ when(mockMetaData.getColumnCount()).thenReturn(2);
+ when(mockMetaData.getColumnLabel(1)).thenReturn("id");
+ when(mockMetaData.getColumnLabel(2)).thenReturn("name");
+ when(mockResultSet.next()).thenReturn(false); // empty result set
+
+ Method resultProcessMethod =
SqlTask.class.getDeclaredMethod("resultProcess", ResultSet.class);
+ resultProcessMethod.setAccessible(true);
+ resultProcessMethod.invoke(task, mockResultSet);
+
+ Assertions.assertTrue(ctx.isNeedAlert());
+
+ TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+ Assertions.assertNotNull(alertInfo);
+ Assertions.assertEquals(3, alertInfo.getAlertGroupId());
+ Assertions.assertEquals("empty result", alertInfo.getTitle());
+ Assertions.assertEquals(AlertType.TASK_RESULT,
alertInfo.getAlertType());
+
+ ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+ Assertions.assertEquals(1, contentArray.size());
+ Assertions.assertEquals("", contentArray.get(0).get("id").asText());
+ Assertions.assertEquals("", contentArray.get(0).get("name").asText());
+ }
+
+ @Test
+ void testTaskExecutionContextCarriesAlertInfoForEventPropagation() {
+ TaskExecutionContext ctx = new TaskExecutionContext();
+ ctx.setNeedAlert(true);
+
+ TaskAlertInfo alertInfo = new TaskAlertInfo();
+ alertInfo.setTitle("test title");
+ alertInfo.setContent("[{\"id\":\"1\"}]");
+ alertInfo.setAlertGroupId(7);
+ alertInfo.setAlertType(AlertType.TASK_RESULT);
+ ctx.setTaskAlertInfo(alertInfo);
+
+ Assertions.assertTrue(ctx.isNeedAlert());
+ Assertions.assertNotNull(ctx.getTaskAlertInfo());
+ Assertions.assertEquals("test title",
ctx.getTaskAlertInfo().getTitle());
+ Assertions.assertEquals(7, ctx.getTaskAlertInfo().getAlertGroupId());
+ Assertions.assertEquals(AlertType.TASK_RESULT,
ctx.getTaskAlertInfo().getAlertType());
+
+ TaskExecutionContext ctx2 = new TaskExecutionContext();
+ Assertions.assertFalse(ctx2.isNeedAlert());
+ Assertions.assertNull(ctx2.getTaskAlertInfo());
+ }
+
}
diff --git
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java
index 7118a099d5..7d589599b2 100644
---
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java
+++
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java
@@ -23,6 +23,7 @@ import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.plugin.storage.api.StorageOperator;
import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.plugin.task.api.log.TaskLogMarkers;
import org.apache.dolphinscheduler.plugin.task.api.model.ApplicationInfo;
import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext;
@@ -91,6 +92,13 @@ public class PhysicalTaskExecutor extends
AbstractTaskExecutor {
@Override
protected TaskExecutorState doTrackTaskPluginStatus() {
+ if (physicalTask.getExitStatus() == TaskExecutionStatus.SUCCESS) {
+ // Synchronize the legacy needAlert/taskAlertInfo fields written
directly
+ // by AbstractTask subclasses into the context, so that the success
+ // lifecycle event carries the alert info of third-party plugins.
+ taskExecutionContext.setNeedAlert(physicalTask.getNeedAlert());
+
taskExecutionContext.setTaskAlertInfo(physicalTask.getTaskAlertInfo());
+ }
return
TaskExecutorStateMappings.mapState(physicalTask.getExitStatus());
}
