From 8fdb5e9ad0bb1e1803caeaa883347d06fc9b6f3e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Wed, 22 Jul 2026 14:00:47 +0800 Subject: [PATCH 01/13] Implement SQL task query result alert via Master-side event handling --- docs/docs/en/guide/upgrade/incompatible.md | 4 ++ docs/docs/zh/guide/upgrade/incompatible.md | 5 ++- .../alert/rpc/AlertOperatorImpl.java | 3 +- .../alert/service/AlertSender.java | 5 ++- .../alert/runner/AlertSenderTest.java | 11 +++--- .../common/enums/AlertType.java | 3 +- .../mysql/dolphinscheduler_ddl.sql | 17 ++++++++ .../mysql/dolphinscheduler_dml.sql | 39 +++++++++++++++++++ .../postgresql/dolphinscheduler_ddl.sql | 18 +++++++++ .../postgresql/dolphinscheduler_dml.sql | 31 +++++++++++++++ .../dao/mapper/TaskDefinitionMapperTest.java | 4 +- .../alert/request/AlertSendRequest.java | 4 +- .../rpc/TaskExecutorEventListenerImpl.java | 13 +++++++ .../service/alert/WorkflowAlertManager.java | 25 ++++++++++++ .../TaskExecutorSuccessLifecycleEvent.java | 7 ++++ .../plugin/task/api/TaskExecutionContext.java | 5 +++ .../plugin/task/api/model/TaskAlertInfo.java | 4 ++ .../task/api/parameters/SqlParameters.java | 17 ++++++-- .../api/parameters/SqlParametersTest.java | 6 +-- .../plugin/task/sql/SqlTask.java | 38 +++++++++++++----- .../components/node/fields/use-sql-type.ts | 14 +++---- .../task/components/node/format-data.ts | 4 +- .../projects/task/components/node/types.ts | 2 +- 23 files changed, 241 insertions(+), 38 deletions(-) create mode 100644 dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql create mode 100644 dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql create mode 100644 dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql create mode 100644 dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql diff --git a/docs/docs/en/guide/upgrade/incompatible.md b/docs/docs/en/guide/upgrade/incompatible.md index 0f60fcb9986a..661e049f7871 100644 --- a/docs/docs/en/guide/upgrade/incompatible.md +++ b/docs/docs/en/guide/upgrade/incompatible.md @@ -44,3 +44,7 @@ This document records the incompatible updates between each version. You need to * Remove import and export of workflow definition. ([#17940])(https://github.com/apache/dolphinscheduler/issues/17940) +## 3.5.0 + +* Rename the `sendEmail` field in SQL task parameters to `sendAlert`. The data migration script will automatically migrate `sendEmail` to `sendAlert` in the `t_ds_task_definition` and `t_ds_task_definition_log` tables.([#17854])(https://github.com/apache/dolphinscheduler/issues/17854) + diff --git a/docs/docs/zh/guide/upgrade/incompatible.md b/docs/docs/zh/guide/upgrade/incompatible.md index 2f0c4c044ff8..601f1c6dc7ac 100644 --- a/docs/docs/zh/guide/upgrade/incompatible.md +++ b/docs/docs/zh/guide/upgrade/incompatible.md @@ -4,7 +4,6 @@ ## dev -* 将mysql驱动版本从8.0.16升级至8.0.33 ([#14684](https://github.com/apache/dolphinscheduler/pull/14684)) * 更改了环境变量名称，将 `PYTHON_HOME` 改为 `PYTHON_LAUNCHER`， 将 `DATAX_HOME` 改为 `DATAX_LAUNCHER` ([#14523](https://github.com/apache/dolphinscheduler/pull/14523)) * 更新了SQL任务中用于匹配变量的正则表达式 ([#13378](https://github.com/apache/dolphinscheduler/pull/13378)) * Remove the spark version of spark task ([#11860](https://github.com/apache/dolphinscheduler/pull/11860)). @@ -48,3 +47,7 @@ * 移除导入导出工作流([#17940])(https://github.com/apache/dolphinscheduler/issues/17940) +## 3.5.0 + +* 将SQL任务参数中的 `sendEmail` 字段重命名为 `sendAlert`，数据迁移脚本会自动将 `t_ds_task_definition` 和 `t_ds_task_definition_log` 表中的 `sendEmail` 迁移为 `sendAlert`。([#17854])(https://github.com/apache/dolphinscheduler/issues/17854) + diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/rpc/AlertOperatorImpl.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/rpc/AlertOperatorImpl.java index 95b3809dd20b..2e8788ab49f5 100644 --- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/rpc/AlertOperatorImpl.java +++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/rpc/AlertOperatorImpl.java @@ -41,7 +41,8 @@ public AlertSendResponse sendAlert(AlertSendRequest alertSendRequest) { AlertSendResponse alertSendResponse = alertSender.syncHandler( alertSendRequest.getGroupId(), alertSendRequest.getTitle(), - alertSendRequest.getContent()); + alertSendRequest.getContent(), + alertSendRequest.getAlertType()); log.info("Handle AlertSendRequest finish: {}", alertSendResponse); return alertSendResponse; } 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..3ef14c9414a7 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 @@ -55,13 +55,16 @@ public AlertSender(AlertDao alertDao, * @param alertGroupId alertGroupId * @param title title * @param content content + * @param alertType alertType * @return AlertSendResponseCommand */ - public AlertSendResponse syncHandler(int alertGroupId, String title, String content) { + public AlertSendResponse syncHandler(int alertGroupId, String title, String content, + org.apache.dolphinscheduler.common.enums.AlertType alertType) { List alertInstanceList = alertDao.listInstanceByAlertGroupId(alertGroupId); AlertData alertData = AlertData.builder() .content(content) .title(title) + .alertType(alertType.getCode()) .build(); boolean sendResponseStatus = true; 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..5ba3c6df25d9 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 @@ -89,7 +89,8 @@ void testSyncHandler() { // 1.alert instance does not exist when(alertDao.listInstanceByAlertGroupId(ALERT_GROUP_ID)).thenReturn(null); - AlertSendResponse alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); + AlertSendResponse alertSendResponse = + alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); Assertions.assertFalse(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); @@ -104,7 +105,7 @@ void testSyncHandler() { alertInstanceList.add(alertPluginInstance); when(alertDao.listInstanceByAlertGroupId(1)).thenReturn(alertInstanceList); - alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); + alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); Assertions.assertFalse(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); @@ -114,7 +115,7 @@ void testSyncHandler() { when(alertChannelMock.process(Mockito.any())).thenReturn(null); when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock)); - alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); + alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); Assertions.assertFalse(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); @@ -126,7 +127,7 @@ void testSyncHandler() { when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult); when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock)); - alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); + alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); Assertions.assertFalse(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); @@ -138,7 +139,7 @@ void testSyncHandler() { when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult); when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock)); - alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); + alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); Assertions.assertTrue(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); 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/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql new file mode 100644 index 000000000000..5f26e3515d67 --- /dev/null +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql @@ -0,0 +1,17 @@ +/* + * 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. +*/ + diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql new file mode 100644 index 000000000000..96180cc1e501 --- /dev/null +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql @@ -0,0 +1,39 @@ +/* + * 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. +*/ + +UPDATE t_ds_task_definition +SET task_params = JSON_REMOVE( + JSON_INSERT( + task_params, + '$.sendAlert', + JSON_EXTRACT(task_params, '$.sendEmail') + ), + '$.sendEmail' +) +WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; + +UPDATE t_ds_task_definition_log +SET task_params = JSON_REMOVE( + JSON_INSERT( + task_params, + '$.sendAlert', + JSON_EXTRACT(task_params, '$.sendEmail') + ), + '$.sendEmail' +) +WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; + diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql new file mode 100644 index 000000000000..61d15aaf4dfe --- /dev/null +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql @@ -0,0 +1,18 @@ +/* + * 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. +*/ + + diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql new file mode 100644 index 000000000000..970cf33b5243 --- /dev/null +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql @@ -0,0 +1,31 @@ +/* + * 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. +*/ + +UPDATE t_ds_task_definition +SET task_params = ( + (task_params::jsonb - 'sendEmail') + || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') +)::text +WHERE jsonb_path_exists(task_params::jsonb, '$.sendEmail'); + +UPDATE t_ds_task_definition_log +SET task_params = ( + (task_params::jsonb - 'sendEmail') + || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') +)::text +WHERE jsonb_path_exists(task_params::jsonb, '$.sendEmail'); + diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapperTest.java index da3cc1d27481..83fc14e5fec7 100644 --- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapperTest.java +++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapperTest.java @@ -140,7 +140,7 @@ public void testDeleteByCode() { @Test public void testNullPropertyValueOfLocalParams() { String definitionJson = - "{\"failRetryTimes\":\"0\",\"timeoutNotifyStrategy\":\"\",\"code\":\"5195043558720\",\"flag\":\"YES\",\"environmentCode\":\"-1\",\"taskDefinitionIndex\":2,\"taskPriority\":\"MEDIUM\",\"taskParams\":\"{\\\"preStatements\\\":null,\\\"postStatements\\\":null,\\\"type\\\":\\\"ADB_MYSQL\\\",\\\"database\\\":\\\"lijia\\\",\\\"sql\\\":\\\"create table nation_${random_serial_number} as select * from nation\\\",\\\"localParams\\\":[{\\\"direct\\\":2,\\\"type\\\":3,\\\"prop\\\":\\\"key\\\"}],\\\"Name\\\":\\\"create_table_as_select_nation\\\",\\\"FailRetryTimes\\\":0,\\\"dbClusterId\\\":\\\"amv-bp10o45925jpx959\\\",\\\"sendEmail\\\":false,\\\"displayRows\\\":10,\\\"limit\\\":10000,\\\"agentSource\\\":\\\"Workflow\\\",\\\"agentVersion\\\":\\\"Unkown\\\"}\",\"timeout\":\"0\",\"taskType\":\"ADB_MYSQL\",\"timeoutFlag\":\"CLOSE\",\"projectCode\":\"5191800302720\",\"name\":\"create_table_as_select_nation\",\"delayTime\":\"0\",\"workerGroup\":\"default\"}"; + "{\"failRetryTimes\":\"0\",\"timeoutNotifyStrategy\":\"\",\"code\":\"5195043558720\",\"flag\":\"YES\",\"environmentCode\":\"-1\",\"taskDefinitionIndex\":2,\"taskPriority\":\"MEDIUM\",\"taskParams\":\"{\\\"preStatements\\\":null,\\\"postStatements\\\":null,\\\"type\\\":\\\"ADB_MYSQL\\\",\\\"database\\\":\\\"lijia\\\",\\\"sql\\\":\\\"create table nation_${random_serial_number} as select * from nation\\\",\\\"localParams\\\":[{\\\"direct\\\":2,\\\"type\\\":3,\\\"prop\\\":\\\"key\\\"}],\\\"Name\\\":\\\"create_table_as_select_nation\\\",\\\"FailRetryTimes\\\":0,\\\"dbClusterId\\\":\\\"amv-bp10o45925jpx959\\\",\\\"sendAlert\\\":false,\\\"displayRows\\\":10,\\\"limit\\\":10000,\\\"agentSource\\\":\\\"Workflow\\\",\\\"agentVersion\\\":\\\"Unkown\\\"}\",\"timeout\":\"0\",\"taskType\":\"ADB_MYSQL\",\"timeoutFlag\":\"CLOSE\",\"projectCode\":\"5191800302720\",\"name\":\"create_table_as_select_nation\",\"delayTime\":\"0\",\"workerGroup\":\"default\"}"; TaskDefinition definition = JSONUtils.parseObject(definitionJson, TaskDefinition.class); Map taskParamsMap = definition.getTaskParamMap(); @@ -157,7 +157,7 @@ public void testNullPropertyValueOfLocalParams() { @Test public void testNullLocalParamsOfTaskParams() { String definitionJson = - "{\"failRetryTimes\":\"0\",\"timeoutNotifyStrategy\":\"\",\"code\":\"5195043558720\",\"flag\":\"YES\",\"environmentCode\":\"-1\",\"taskDefinitionIndex\":2,\"taskPriority\":\"MEDIUM\",\"taskParams\":\"{\\\"preStatements\\\":null,\\\"postStatements\\\":null,\\\"type\\\":\\\"ADB_MYSQL\\\",\\\"database\\\":\\\"lijia\\\",\\\"sql\\\":\\\"create table nation_${random_serial_number} as select * from nation\\\",\\\"localParams\\\":null,\\\"Name\\\":\\\"create_table_as_select_nation\\\",\\\"FailRetryTimes\\\":0,\\\"dbClusterId\\\":\\\"amv-bp10o45925jpx959\\\",\\\"sendEmail\\\":false,\\\"displayRows\\\":10,\\\"limit\\\":10000,\\\"agentSource\\\":\\\"Workflow\\\",\\\"agentVersion\\\":\\\"Unkown\\\"}\",\"timeout\":\"0\",\"taskType\":\"ADB_MYSQL\",\"timeoutFlag\":\"CLOSE\",\"projectCode\":\"5191800302720\",\"name\":\"create_table_as_select_nation\",\"delayTime\":\"0\",\"workerGroup\":\"default\"}"; + "{\"failRetryTimes\":\"0\",\"timeoutNotifyStrategy\":\"\",\"code\":\"5195043558720\",\"flag\":\"YES\",\"environmentCode\":\"-1\",\"taskDefinitionIndex\":2,\"taskPriority\":\"MEDIUM\",\"taskParams\":\"{\\\"preStatements\\\":null,\\\"postStatements\\\":null,\\\"type\\\":\\\"ADB_MYSQL\\\",\\\"database\\\":\\\"lijia\\\",\\\"sql\\\":\\\"create table nation_${random_serial_number} as select * from nation\\\",\\\"localParams\\\":null,\\\"Name\\\":\\\"create_table_as_select_nation\\\",\\\"FailRetryTimes\\\":0,\\\"dbClusterId\\\":\\\"amv-bp10o45925jpx959\\\",\\\"sendAlert\\\":false,\\\"displayRows\\\":10,\\\"limit\\\":10000,\\\"agentSource\\\":\\\"Workflow\\\",\\\"agentVersion\\\":\\\"Unkown\\\"}\",\"timeout\":\"0\",\"taskType\":\"ADB_MYSQL\",\"timeoutFlag\":\"CLOSE\",\"projectCode\":\"5191800302720\",\"name\":\"create_table_as_select_nation\",\"delayTime\":\"0\",\"workerGroup\":\"default\"}"; TaskDefinition definition = JSONUtils.parseObject(definitionJson, TaskDefinition.class); Assertions.assertNull(definition.getTaskParamMap(), "Serialize the task definition success"); diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-alert/src/main/java/org/apache/dolphinscheduler/extract/alert/request/AlertSendRequest.java b/dolphinscheduler-extract/dolphinscheduler-extract-alert/src/main/java/org/apache/dolphinscheduler/extract/alert/request/AlertSendRequest.java index dfa6515ddea2..82996bfc6c2f 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-alert/src/main/java/org/apache/dolphinscheduler/extract/alert/request/AlertSendRequest.java +++ b/dolphinscheduler-extract/dolphinscheduler-extract-alert/src/main/java/org/apache/dolphinscheduler/extract/alert/request/AlertSendRequest.java @@ -17,6 +17,8 @@ package org.apache.dolphinscheduler.extract.alert.request; +import org.apache.dolphinscheduler.common.enums.AlertType; + import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @@ -32,6 +34,6 @@ public class AlertSendRequest { private String content; - private int warnType; + private AlertType alertType; } 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 278f5160dc80..9421d293ba41 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 @@ -18,6 +18,7 @@ package org.apache.dolphinscheduler.server.master.rpc; import org.apache.dolphinscheduler.extract.master.ITaskExecutorEventListener; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils; import org.apache.dolphinscheduler.server.master.engine.IWorkflowRepository; import org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution; @@ -29,6 +30,7 @@ import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskRuntimeContextChangedEvent; import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskSuccessLifecycleEvent; import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution; +import org.apache.dolphinscheduler.service.alert.WorkflowAlertManager; import org.apache.dolphinscheduler.task.executor.events.IReportableTaskExecutorLifecycleEvent; import org.apache.dolphinscheduler.task.executor.events.TaskExecutorDispatchedLifecycleEvent; import org.apache.dolphinscheduler.task.executor.events.TaskExecutorFailedLifecycleEvent; @@ -52,6 +54,9 @@ public class TaskExecutorEventListenerImpl implements ITaskExecutorEventListener @Autowired private IWorkflowRepository workflowRepository; + @Autowired + private WorkflowAlertManager workflowAlertManager; + @Override public void onTaskExecutorDispatched(final TaskExecutorDispatchedLifecycleEvent taskExecutorDispatchedLifecycleEvent) { LogUtils.setWorkflowInstanceIdMDC(taskExecutorDispatchedLifecycleEvent.getWorkflowInstanceId()); @@ -118,6 +123,14 @@ public void onTaskExecutorSuccess(final TaskExecutorSuccessLifecycleEvent taskEx .varPool(taskExecutorSuccessLifecycleEvent.getVarPool()) .build(); taskExecution.getWorkflowEventBus().publish(taskSuccessEvent); + + if (taskExecutorSuccessLifecycleEvent.isNeedAlert()) { + TaskAlertInfo taskAlertInfo = taskExecutorSuccessLifecycleEvent.getTaskAlertInfo(); + workflowAlertManager.sendTaskResultAlert( + taskAlertInfo, + taskExecution.getTaskExecutionContext().getProjectCode(), + taskExecution.getTaskExecutionContext().getWorkflowInstanceId()); + } } finally { LogUtils.removeWorkflowInstanceIdMDC(); } 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 9335e9551eb7..576ac19377a8 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 @@ -34,6 +34,7 @@ import org.apache.dolphinscheduler.dao.repository.ProjectDao; import org.apache.dolphinscheduler.dao.repository.UserDao; import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionLogDao; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import java.util.ArrayList; import java.util.Date; @@ -208,4 +209,28 @@ public void sendWorkflowTimeoutAlert(WorkflowInstance workflowInstance) { alertDao.sendWorkflowTimeoutAlert(workflowInstance, projectUser); } + /** + * Send task result alert + * + * @param taskAlertInfo task alert info + * @param projectCode project code + * @param workflowInstanceId workflow instance id + */ + public void sendTaskResultAlert(TaskAlertInfo taskAlertInfo, Long projectCode, Integer workflowInstanceId) { + if (taskAlertInfo == null || taskAlertInfo.getAlertGroupId() == null) { + return; + } + Alert alert = new Alert(); + alert.setTitle(taskAlertInfo.getTitle()); + alert.setContent(taskAlertInfo.getContent()); + alert.setAlertGroupId(taskAlertInfo.getAlertGroupId()); + alert.setAlertType(taskAlertInfo.getAlertType()); + alert.setWarningType(WarningType.SUCCESS); + alert.setCreateTime(new Date()); + alert.setProjectCode(projectCode); + alert.setWorkflowInstanceId(workflowInstanceId); + alertDao.addAlert(alert); + log.info("Added task result alert for workflow instance: {}", workflowInstanceId); + } + } 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 947d1edf3633..2c55c80c365c 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 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import org.apache.dolphinscheduler.task.executor.ITaskExecutor; import java.util.List; @@ -48,6 +49,10 @@ public class TaskExecutorSuccessLifecycleEvent extends AbstractTaskExecutorLifec private Long latestReportTime; + private boolean needAlert; + + private TaskAlertInfo taskAlertInfo; + public static TaskExecutorSuccessLifecycleEvent of(final ITaskExecutor taskExecutor) { final TaskExecutionContext taskExecutionContext = taskExecutor.getTaskExecutionContext(); return TaskExecutorSuccessLifecycleEvent.builder() @@ -57,6 +62,8 @@ public static TaskExecutorSuccessLifecycleEvent of(final ITaskExecutor taskExecu .varPool(taskExecutionContext.getVarPool()) .type(TaskExecutorLifecycleEventType.SUCCESS) .endTime(taskExecutionContext.getEndTime()) + .needAlert(taskExecutionContext.isNeedAlert()) + .taskAlertInfo(taskExecutionContext.getTaskAlertInfo()) .build(); } 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 85d6f682196b..a072258011df 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 @@ 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 e72242e93473..d6771f6b20ab 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,6 @@ public class TaskAlertInfo { private String content; private Integer alertGroupId; + + private AlertType alertType; } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java index 32b110255cf9..7da491ea9871 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java @@ -73,8 +73,6 @@ public class SqlParameters extends AbstractParameters { */ private int sqlType; - private Boolean sendEmail; - private int displayRows; /** @@ -85,6 +83,7 @@ public class SqlParameters extends AbstractParameters { * 3 TABLE+attachment */ private String showType; + /** * SQL connection parameters */ @@ -92,7 +91,19 @@ public class SqlParameters extends AbstractParameters { private List preStatements; private List postStatements; + /** + * Whether to send alert for SQL query result + */ + private Boolean sendAlert; + + /** + * Alert group id + */ private int groupId; + + /** + * Alert title + */ private String title; private int limit; @@ -173,7 +184,7 @@ public String toString() { + ", sqlSource='" + sqlSource + '\'' + ", sqlResource='" + sqlResource + '\'' + ", sqlType=" + sqlType - + ", sendEmail=" + sendEmail + + ", sendAlert=" + sendAlert + ", displayRows=" + displayRows + ", limit=" + limit + ", showType='" + showType + '\'' diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java index b9388b3ba54f..39c17615f384 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java @@ -40,7 +40,7 @@ public class SqlParametersTest { private final String sql = "select * from t_ds_user"; private final int datasource = 1; private final int sqlType = 0; - private final Boolean sendEmail = true; + private final Boolean sendAlert = true; private final int displayRows = 10; private final String showType = "TABLE"; private final String title = "sql test"; @@ -63,7 +63,7 @@ public void testSqlParameters() { sqlParameters.setSql(sql); sqlParameters.setDatasource(datasource); sqlParameters.setSqlType(sqlType); - sqlParameters.setSendEmail(sendEmail); + sqlParameters.setSendAlert(sendAlert); sqlParameters.setDisplayRows(displayRows); sqlParameters.setShowType(showType); sqlParameters.setTitle(title); @@ -73,7 +73,7 @@ public void testSqlParameters() { Assertions.assertEquals(sql, sqlParameters.getSql()); Assertions.assertEquals(datasource, sqlParameters.getDatasource()); Assertions.assertEquals(sqlType, sqlParameters.getSqlType()); - Assertions.assertEquals(sendEmail, sqlParameters.getSendEmail()); + Assertions.assertEquals(sendAlert, sqlParameters.getSendAlert()); Assertions.assertEquals(displayRows, sqlParameters.getDisplayRows()); Assertions.assertEquals(showType, sqlParameters.getShowType()); Assertions.assertEquals(title, sqlParameters.getTitle()); 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 113dfbcd9eb8..851efa0e3161 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,6 +17,7 @@ package org.apache.dolphinscheduler.plugin.task.sql; +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.plugin.datasource.api.plugin.DataSourceClientProvider; @@ -290,10 +291,25 @@ private String resultProcess(ResultSet resultSet) throws Exception { String result = resultJSONArray.isEmpty() ? JSONUtils.toJsonString(generateEmptyRow(resultSet)) : JSONUtils.toJsonString(resultJSONArray); - if (Boolean.TRUE.equals(sqlParameters.getSendEmail())) { - sendAttachment(sqlParameters.getGroupId(), StringUtils.isNotEmpty(sqlParameters.getTitle()) + if (Boolean.TRUE.equals(sqlParameters.getSendAlert())) { + // Truncate alert content to avoid oversized payload when query result is large + int displayRows = sqlParameters.getDisplayRows() > 0 ? sqlParameters.getDisplayRows() + : TaskConstants.DEFAULT_DISPLAY_ROWS; + String alertContent; + if (resultJSONArray.size() > displayRows) { + ArrayNode truncatedArray = JSONUtils.createArrayNode(); + for (int i = 0; i < Math.min(displayRows, resultJSONArray.size()); i++) { + truncatedArray.add(resultJSONArray.get(i)); + } + alertContent = JSONUtils.toJsonString(truncatedArray); + log.debug("Alert content truncated to {} rows", displayRows); + } else { + alertContent = result; + } + + prepareTaskResultAlert(sqlParameters.getGroupId(), StringUtils.isNotEmpty(sqlParameters.getTitle()) ? sqlParameters.getTitle() - : taskExecutionContext.getTaskName() + " query result sets", result); + : taskExecutionContext.getTaskName() + " query result sets", alertContent); } log.debug("execute sql result : {}", result); return result; @@ -320,18 +336,20 @@ private ArrayNode generateEmptyRow(ResultSet resultSet) throws SQLException { } /** - * send alert as an attachment + * Prepare task result alert info * - * @param title title - * @param content content + * @param alertGroupId alert group id + * @param title alert title + * @param content alert content */ - private void sendAttachment(int groupId, String title, String content) { - setNeedAlert(Boolean.TRUE); + private void prepareTaskResultAlert(int alertGroupId, String title, String content) { TaskAlertInfo taskAlertInfo = new TaskAlertInfo(); - taskAlertInfo.setAlertGroupId(groupId); + taskAlertInfo.setAlertGroupId(alertGroupId); taskAlertInfo.setContent(content); taskAlertInfo.setTitle(title); - setTaskAlertInfo(taskAlertInfo); + taskAlertInfo.setAlertType(AlertType.TASK_RESULT); + taskExecutionContext.setNeedAlert(true); + taskExecutionContext.setTaskAlertInfo(taskAlertInfo); } private String executeQuery(Connection connection, SqlBinds sqlBinds, String handlerType) throws Exception { diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sql-type.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sql-type.ts index d30ef9ef3221..b9ac314ff3d4 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sql-type.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sql-type.ts @@ -24,8 +24,8 @@ import type { IJsonItem } from '../types' export function useSqlType(model: { [field: string]: any }): IJsonItem[] { const { t } = useI18n() const querySpan = computed(() => (model.sqlType === '0' ? 6 : 0)) - const emailSpan = computed(() => - model.sqlType === '0' && model.sendEmail ? 24 : 0 + const alertSpan = computed(() => + model.sqlType === '0' && model.sendAlert ? 24 : 0 ) const groups = ref([]) const groupsLoading = ref(false) @@ -69,7 +69,7 @@ export function useSqlType(model: { [field: string]: any }): IJsonItem[] { }, { type: 'switch', - field: 'sendEmail', + field: 'sendAlert', span: querySpan, name: t('project.node.send_alarm') }, @@ -109,12 +109,12 @@ export function useSqlType(model: { [field: string]: any }): IJsonItem[] { props: { placeholder: t('project.node.title_tips') }, - span: emailSpan, + span: alertSpan, validate: { trigger: ['input', 'blur'], required: true, validator(unuse, value) { - if (model.sendEmail && !value) + if (model.sendAlert && !value) return new Error(t('project.node.title_tips')) } } @@ -124,7 +124,7 @@ export function useSqlType(model: { [field: string]: any }): IJsonItem[] { field: 'groupId', name: t('project.node.alarm_group'), options: groups, - span: emailSpan, + span: alertSpan, props: { loading: groupsLoading, placeholder: t('project.node.alarm_group_tips') @@ -133,7 +133,7 @@ export function useSqlType(model: { [field: string]: any }): IJsonItem[] { trigger: ['input', 'blur'], required: true, validator(unuse, value) { - if (model.sendEmail && !value) + if (model.sendAlert && !value) return new Error(t('project.node.alarm_group_tips')) } } diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts index a7754cb24d13..b29db829f33f 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts @@ -208,9 +208,9 @@ export function formatParams(data: INodeData): { taskParams.sqlType = data.sqlType taskParams.preStatements = data.preStatements taskParams.postStatements = data.postStatements - taskParams.sendEmail = data.sendEmail + taskParams.sendAlert = data.sendAlert taskParams.displayRows = data.displayRows - if (data.sqlType === '0' && data.sendEmail) { + if (data.sqlType === '0' && data.sendAlert) { taskParams.title = data.title taskParams.groupId = data.groupId } diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts index 218f318db0d0..479f9260cc4d 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts @@ -305,7 +305,7 @@ interface ITaskParams { datasource?: string sql?: string sqlType?: string - sendEmail?: boolean + sendAlert?: boolean displayRows?: number title?: string groupId?: string From f4ad79cb55e8b59a906ec1c065959c5ed5019675 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Wed, 29 Jul 2026 11:44:14 +0800 Subject: [PATCH 02/13] restore 3.5.0_schema --- .../mysql/dolphinscheduler_ddl.sql | 17 -------- .../mysql/dolphinscheduler_dml.sql | 39 ------------------- .../postgresql/dolphinscheduler_ddl.sql | 18 --------- .../postgresql/dolphinscheduler_dml.sql | 31 --------------- 4 files changed, 105 deletions(-) delete mode 100644 dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql delete mode 100644 dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql delete mode 100644 dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql delete mode 100644 dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql deleted file mode 100644 index 5f26e3515d67..000000000000 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_ddl.sql +++ /dev/null @@ -1,17 +0,0 @@ -/* - * 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. -*/ - diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql deleted file mode 100644 index 96180cc1e501..000000000000 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql +++ /dev/null @@ -1,39 +0,0 @@ -/* - * 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. -*/ - -UPDATE t_ds_task_definition -SET task_params = JSON_REMOVE( - JSON_INSERT( - task_params, - '$.sendAlert', - JSON_EXTRACT(task_params, '$.sendEmail') - ), - '$.sendEmail' -) -WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; - -UPDATE t_ds_task_definition_log -SET task_params = JSON_REMOVE( - JSON_INSERT( - task_params, - '$.sendAlert', - JSON_EXTRACT(task_params, '$.sendEmail') - ), - '$.sendEmail' -) -WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; - diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql deleted file mode 100644 index 61d15aaf4dfe..000000000000 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_ddl.sql +++ /dev/null @@ -1,18 +0,0 @@ -/* - * 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. -*/ - - diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql deleted file mode 100644 index 970cf33b5243..000000000000 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql +++ /dev/null @@ -1,31 +0,0 @@ -/* - * 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. -*/ - -UPDATE t_ds_task_definition -SET task_params = ( - (task_params::jsonb - 'sendEmail') - || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') -)::text -WHERE jsonb_path_exists(task_params::jsonb, '$.sendEmail'); - -UPDATE t_ds_task_definition_log -SET task_params = ( - (task_params::jsonb - 'sendEmail') - || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') -)::text -WHERE jsonb_path_exists(task_params::jsonb, '$.sendEmail'); - From 5b25a776b51b1b0fa0f544dab2af14cd33d803ba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Wed, 29 Jul 2026 15:22:29 +0800 Subject: [PATCH 03/13] Migrate sendEmail field to sendAlert in historical data --- .../mysql/dolphinscheduler_dml.sql | 22 +++++++++++++++++++ .../postgresql/dolphinscheduler_dml.sql | 14 ++++++++++++ .../rpc/TaskExecutorEventListenerImpl.java | 10 +++++---- .../plugin/task/api/model/TaskAlertInfo.java | 1 + 4 files changed, 43 insertions(+), 4 deletions(-) diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql index 4a14f326b985..fd841f6a2301 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql @@ -14,3 +14,25 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + +UPDATE t_ds_task_definition +SET task_params = JSON_REMOVE( + JSON_INSERT( + task_params, + '$.sendAlert', + JSON_EXTRACT(task_params, '$.sendEmail') + ), + '$.sendEmail' +) +WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; + +UPDATE t_ds_task_definition_log +SET task_params = JSON_REMOVE( + JSON_INSERT( + task_params, + '$.sendAlert', + JSON_EXTRACT(task_params, '$.sendEmail') + ), + '$.sendEmail' +) +WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql index 4a14f326b985..86ff0204f1df 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql @@ -14,3 +14,17 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + +UPDATE t_ds_task_definition +SET task_params = ( + (task_params::jsonb - 'sendEmail') + || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') +)::text +WHERE jsonb_path_exists(task_params::jsonb, '$.sendEmail'); + +UPDATE t_ds_task_definition_log +SET task_params = ( + (task_params::jsonb - 'sendEmail') + || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') +)::text +WHERE jsonb_path_exists(task_params::jsonb, '$.sendEmail'); 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 9421d293ba41..c73035e21487 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 @@ -126,10 +126,12 @@ public void onTaskExecutorSuccess(final TaskExecutorSuccessLifecycleEvent taskEx if (taskExecutorSuccessLifecycleEvent.isNeedAlert()) { TaskAlertInfo taskAlertInfo = taskExecutorSuccessLifecycleEvent.getTaskAlertInfo(); - workflowAlertManager.sendTaskResultAlert( - taskAlertInfo, - taskExecution.getTaskExecutionContext().getProjectCode(), - taskExecution.getTaskExecutionContext().getWorkflowInstanceId()); + if (taskAlertInfo != null) { + workflowAlertManager.sendTaskResultAlert( + taskAlertInfo, + taskExecution.getTaskExecutionContext().getProjectCode(), + taskExecution.getTaskExecutionContext().getWorkflowInstanceId()); + } } } finally { LogUtils.removeWorkflowInstanceIdMDC(); 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 d6771f6b20ab..d2c22e565a79 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 @@ -31,4 +31,5 @@ public class TaskAlertInfo { private Integer alertGroupId; private AlertType alertType; + } From 78f0940795b66f55d0fe1996fa41e20310ae0964 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Thu, 30 Jul 2026 11:50:58 +0800 Subject: [PATCH 04/13] refactor prepareTaskResultAlert --- .../plugin/task/api/AbstractTask.java | 21 -------- .../task/api/parameters/SqlParameters.java | 12 +---- .../api/parameters/SqlParametersTest.java | 28 +++++++++-- .../plugin/task/sql/SqlTask.java | 48 ++++++++----------- 4 files changed, 47 insertions(+), 62 deletions(-) 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 0c8f208fcbae..530205c4556d 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 @@ -19,7 +19,6 @@ import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; import org.apache.dolphinscheduler.plugin.task.api.model.Property; -import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters; import java.util.Map; @@ -56,10 +55,6 @@ public abstract class AbstractTask { */ protected volatile int exitStatusCode = -1; - protected boolean needAlert = false; - - protected TaskAlertInfo taskAlertInfo; - /** * constructor * @@ -109,22 +104,6 @@ public void setAppIds(String appIds) { this.appIds = appIds; } - public boolean getNeedAlert() { - return needAlert; - } - - public void setNeedAlert(boolean needAlert) { - this.needAlert = needAlert; - } - - public TaskAlertInfo getTaskAlertInfo() { - return taskAlertInfo; - } - - public void setTaskAlertInfo(TaskAlertInfo taskAlertInfo) { - this.taskAlertInfo = taskAlertInfo; - } - /** * get task parameters * diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java index 7da491ea9871..84f699a9bb40 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java @@ -39,6 +39,7 @@ import lombok.Data; +import com.fasterxml.jackson.annotation.JsonAlias; import com.google.common.collect.Lists; /** @@ -75,15 +76,6 @@ public class SqlParameters extends AbstractParameters { private int displayRows; - /** - * show type - * 0 TABLE - * 1 TEXT - * 2 attachment - * 3 TABLE+attachment - */ - private String showType; - /** * SQL connection parameters */ @@ -94,6 +86,7 @@ public class SqlParameters extends AbstractParameters { /** * Whether to send alert for SQL query result */ + @JsonAlias("sendEmail") private Boolean sendAlert; /** @@ -187,7 +180,6 @@ public String toString() { + ", sendAlert=" + sendAlert + ", displayRows=" + displayRows + ", limit=" + limit - + ", showType='" + showType + '\'' + ", connParams='" + connParams + '\'' + ", groupId='" + groupId + '\'' + ", title='" + title + '\'' diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java index 39c17615f384..f4cb6f1665e4 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java @@ -17,6 +17,7 @@ package org.apache.dolphinscheduler.plugin.task.api.parameters; +import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.plugin.task.api.SQLTaskExecutionContext; import org.apache.dolphinscheduler.plugin.task.api.enums.DataType; import org.apache.dolphinscheduler.plugin.task.api.enums.Direct; @@ -42,7 +43,6 @@ public class SqlParametersTest { private final int sqlType = 0; private final Boolean sendAlert = true; private final int displayRows = 10; - private final String showType = "TABLE"; private final String title = "sql test"; private final int groupId = 0; @@ -65,7 +65,6 @@ public void testSqlParameters() { sqlParameters.setSqlType(sqlType); sqlParameters.setSendAlert(sendAlert); sqlParameters.setDisplayRows(displayRows); - sqlParameters.setShowType(showType); sqlParameters.setTitle(title); sqlParameters.setGroupId(groupId); @@ -75,7 +74,6 @@ public void testSqlParameters() { Assertions.assertEquals(sqlType, sqlParameters.getSqlType()); Assertions.assertEquals(sendAlert, sqlParameters.getSendAlert()); Assertions.assertEquals(displayRows, sqlParameters.getDisplayRows()); - Assertions.assertEquals(showType, sqlParameters.getShowType()); Assertions.assertEquals(title, sqlParameters.getTitle()); Assertions.assertEquals(groupId, sqlParameters.getGroupId()); @@ -145,4 +143,28 @@ public void testGenerateExtendedContext_setsConnectionParams() { Assertions.assertNotNull(ctx); Assertions.assertEquals("conn_params", ctx.getConnectionParams()); } + + @Test + public void testJsonAlias_sendEmail_backwardCompatibility() { + String jsonWithSendEmail = "{\"type\":\"MYSQL\",\"datasource\":1,\"sql\":\"select 1\",\"sqlType\":0," + + "\"sendEmail\":true,\"displayRows\":10,\"groupId\":2,\"title\":\"test alert\"}"; + + SqlParameters params = JSONUtils.parseObject(jsonWithSendEmail, SqlParameters.class); + Assertions.assertNotNull(params); + Assertions.assertEquals(Boolean.TRUE, params.getSendAlert()); + Assertions.assertEquals(2, params.getGroupId()); + Assertions.assertEquals("test alert", params.getTitle()); + } + + @Test + public void testJson_newFieldName_sendAlert() { + String jsonWithSendAlert = "{\"type\":\"MYSQL\",\"datasource\":1,\"sql\":\"select 1\",\"sqlType\":0," + + "\"sendAlert\":true,\"displayRows\":10,\"groupId\":3,\"title\":\"new alert\"}"; + + SqlParameters params = JSONUtils.parseObject(jsonWithSendAlert, SqlParameters.class); + Assertions.assertNotNull(params); + Assertions.assertEquals(Boolean.TRUE, params.getSendAlert()); + Assertions.assertEquals(3, params.getGroupId()); + Assertions.assertEquals("new alert", params.getTitle()); + } } 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 851efa0e3161..336af6478157 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 @@ -124,12 +124,11 @@ public AbstractParameters getParameters() { public void handle(TaskCallBack taskCallBack) throws TaskException { log.info("Full sql parameters: {}", sqlParameters); log.info( - "sql type : {}, datasource : {}, sql : {} , localParams : {},showType : {},connParams : {},varPool : {} ,query max result limit {}", + "sql type : {}, datasource : {}, sql : {} , localParams : {},connParams : {},varPool : {} ,query max result limit {}", sqlParameters.getType(), sqlParameters.getDatasource(), sqlParameters.getSql(), sqlParameters.getLocalParams(), - sqlParameters.getShowType(), sqlParameters.getConnParams(), sqlParameters.getVarPool(), sqlParameters.getLimit()); @@ -292,24 +291,7 @@ private String resultProcess(ResultSet resultSet) throws Exception { : JSONUtils.toJsonString(resultJSONArray); if (Boolean.TRUE.equals(sqlParameters.getSendAlert())) { - // Truncate alert content to avoid oversized payload when query result is large - int displayRows = sqlParameters.getDisplayRows() > 0 ? sqlParameters.getDisplayRows() - : TaskConstants.DEFAULT_DISPLAY_ROWS; - String alertContent; - if (resultJSONArray.size() > displayRows) { - ArrayNode truncatedArray = JSONUtils.createArrayNode(); - for (int i = 0; i < Math.min(displayRows, resultJSONArray.size()); i++) { - truncatedArray.add(resultJSONArray.get(i)); - } - alertContent = JSONUtils.toJsonString(truncatedArray); - log.debug("Alert content truncated to {} rows", displayRows); - } else { - alertContent = result; - } - - prepareTaskResultAlert(sqlParameters.getGroupId(), StringUtils.isNotEmpty(sqlParameters.getTitle()) - ? sqlParameters.getTitle() - : taskExecutionContext.getTaskName() + " query result sets", alertContent); + prepareTaskResultAlert(resultJSONArray); } log.debug("execute sql result : {}", result); return result; @@ -336,18 +318,28 @@ private ArrayNode generateEmptyRow(ResultSet resultSet) throws SQLException { } /** - * Prepare task result alert info + * Prepare task result alert info. + * Truncate the alert content to displayRows to avoid oversized RPC payload. * - * @param alertGroupId alert group id - * @param title alert title - * @param content alert content + * @param resultJSONArray the full query result JSON array */ - private void prepareTaskResultAlert(int alertGroupId, String title, String content) { + private void prepareTaskResultAlert(ArrayNode resultJSONArray) { TaskAlertInfo taskAlertInfo = new TaskAlertInfo(); - taskAlertInfo.setAlertGroupId(alertGroupId); - taskAlertInfo.setContent(content); - taskAlertInfo.setTitle(title); + taskAlertInfo.setAlertGroupId(sqlParameters.getGroupId()); + taskAlertInfo.setTitle(StringUtils.isNotEmpty(sqlParameters.getTitle()) + ? sqlParameters.getTitle() + : taskExecutionContext.getTaskName() + " query result sets"); + // Truncate content to displayRows to avoid oversized RPC payload + int alertRows = sqlParameters.getDisplayRows() > 0 ? sqlParameters.getDisplayRows() + : TaskConstants.DEFAULT_DISPLAY_ROWS; + alertRows = Math.min(alertRows, resultJSONArray.size()); + ArrayNode alertContent = JSONUtils.createArrayNode(); + for (int i = 0; i < alertRows; i++) { + alertContent.add(resultJSONArray.get(i)); + } + taskAlertInfo.setContent(JSONUtils.toJsonString(alertContent)); taskAlertInfo.setAlertType(AlertType.TASK_RESULT); + taskExecutionContext.setNeedAlert(true); taskExecutionContext.setTaskAlertInfo(taskAlertInfo); } From 2b76ad1530add40e0b845b961e19229e396a7631 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Thu, 30 Jul 2026 15:54:29 +0800 Subject: [PATCH 05/13] add sql result log --- .../apache/dolphinscheduler/plugin/task/sql/SqlTask.java | 6 ++++++ 1 file changed, 6 insertions(+) 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 336af6478157..99f9a5f44ae6 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 @@ -291,7 +291,10 @@ private String resultProcess(ResultSet resultSet) throws Exception { : JSONUtils.toJsonString(resultJSONArray); if (Boolean.TRUE.equals(sqlParameters.getSendAlert())) { + log.info("SendAlert is enabled, preparing task result alert"); prepareTaskResultAlert(resultJSONArray); + } else { + log.info("SendAlert is not enabled, skip task result alert"); } log.debug("execute sql result : {}", result); return result; @@ -342,6 +345,9 @@ private void prepareTaskResultAlert(ArrayNode resultJSONArray) { taskExecutionContext.setNeedAlert(true); taskExecutionContext.setTaskAlertInfo(taskAlertInfo); + log.info("Prepare task result alert: title={}, alertGroupId={}, alertType={}, totalRows={}, alertRows={}", + taskAlertInfo.getTitle(), taskAlertInfo.getAlertGroupId(), taskAlertInfo.getAlertType(), + resultJSONArray.size(), alertRows); } private String executeQuery(Connection connection, SqlBinds sqlBinds, String handlerType) throws Exception { From 793d28a770bd22fad306597a6d0d3f6e9b7c6011 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Tue, 11 Aug 2026 14:34:53 +0800 Subject: [PATCH 06/13] update sendTaskResultAlert --- .../alert/service/AlertSender.java | 4 +-- .../mysql/dolphinscheduler_dml.sql | 24 ++++++++-------- .../postgresql/dolphinscheduler_dml.sql | 4 +-- .../rpc/TaskExecutorEventListenerImpl.java | 13 ++++++--- .../service/alert/WorkflowAlertManager.java | 28 +++++++++++-------- .../task/api/parameters/SqlParameters.java | 4 +++ .../plugin/task/sql/SqlTask.java | 13 +++++++-- 7 files changed, 56 insertions(+), 34 deletions(-) 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 3ef14c9414a7..7846edf6d983 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 @@ -22,6 +22,7 @@ import org.apache.dolphinscheduler.alert.config.AlertConfig; import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager; import org.apache.dolphinscheduler.common.enums.AlertStatus; +import org.apache.dolphinscheduler.common.enums.AlertType; import org.apache.dolphinscheduler.dao.AlertDao; import org.apache.dolphinscheduler.dao.entity.Alert; import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance; @@ -58,8 +59,7 @@ public AlertSender(AlertDao alertDao, * @param alertType alertType * @return AlertSendResponseCommand */ - public AlertSendResponse syncHandler(int alertGroupId, String title, String content, - org.apache.dolphinscheduler.common.enums.AlertType alertType) { + public AlertSendResponse syncHandler(int alertGroupId, String title, String content, AlertType alertType) { List alertInstanceList = alertDao.listInstanceByAlertGroupId(alertGroupId); AlertData alertData = AlertData.builder() .content(content) diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql index fd841f6a2301..8ed1912c2338 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql @@ -17,22 +17,22 @@ UPDATE t_ds_task_definition SET task_params = JSON_REMOVE( - JSON_INSERT( - task_params, - '$.sendAlert', - JSON_EXTRACT(task_params, '$.sendEmail') - ), - '$.sendEmail' + JSON_SET( + task_params, + '$.sendAlert', + JSON_EXTRACT(task_params, '$.sendEmail') + ), + '$.sendEmail' ) WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; UPDATE t_ds_task_definition_log SET task_params = JSON_REMOVE( - JSON_INSERT( - task_params, - '$.sendAlert', - JSON_EXTRACT(task_params, '$.sendEmail') - ), - '$.sendEmail' + JSON_SET( + task_params, + '$.sendAlert', + JSON_EXTRACT(task_params, '$.sendEmail') + ), + '$.sendEmail' ) WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql index 86ff0204f1df..ef3ee9b2f4dd 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql @@ -20,11 +20,11 @@ SET task_params = ( (task_params::jsonb - 'sendEmail') || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') )::text -WHERE jsonb_path_exists(task_params::jsonb, '$.sendEmail'); +WHERE task_params IS NOT NULL AND jsonb_path_exists(task_params::jsonb, '$.sendEmail'); UPDATE t_ds_task_definition_log SET task_params = ( (task_params::jsonb - 'sendEmail') || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') )::text -WHERE jsonb_path_exists(task_params::jsonb, '$.sendEmail'); +WHERE task_params IS NOT NULL AND jsonb_path_exists(task_params::jsonb, '$.sendEmail'); 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 c73035e21487..ebca57911d47 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 @@ -124,13 +124,18 @@ public void onTaskExecutorSuccess(final TaskExecutorSuccessLifecycleEvent taskEx .build(); taskExecution.getWorkflowEventBus().publish(taskSuccessEvent); + // Handle task result alert if (taskExecutorSuccessLifecycleEvent.isNeedAlert()) { TaskAlertInfo taskAlertInfo = taskExecutorSuccessLifecycleEvent.getTaskAlertInfo(); - if (taskAlertInfo != null) { + if (taskAlertInfo != null && taskAlertInfo.getAlertGroupId() != null + && taskAlertInfo.getAlertGroupId() > 0) { workflowAlertManager.sendTaskResultAlert( - taskAlertInfo, - taskExecution.getTaskExecutionContext().getProjectCode(), - taskExecution.getTaskExecutionContext().getWorkflowInstanceId()); + taskExecution.getWorkflowInstance(), + taskExecution.getTaskInstance(), + taskAlertInfo); + } else { + log.warn("Task: {} need alert but alertGroupId is invalid, skip sending alert", + taskExecution.getName()); } } } finally { 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 576ac19377a8..ce6e12be0f20 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 @@ -210,27 +210,33 @@ public void sendWorkflowTimeoutAlert(WorkflowInstance workflowInstance) { } /** - * Send task result alert + * send task result alert * - * @param taskAlertInfo task alert info - * @param projectCode project code - * @param workflowInstanceId workflow instance id + * @param workflowInstance workflow instance + * @param taskInstance task instance + * @param taskAlertInfo task alert info */ - public void sendTaskResultAlert(TaskAlertInfo taskAlertInfo, Long projectCode, Integer workflowInstanceId) { + public void sendTaskResultAlert(WorkflowInstance workflowInstance, + TaskInstance taskInstance, + TaskAlertInfo taskAlertInfo) { if (taskAlertInfo == null || taskAlertInfo.getAlertGroupId() == null) { return; } + Alert alert = new Alert(); alert.setTitle(taskAlertInfo.getTitle()); alert.setContent(taskAlertInfo.getContent()); - alert.setAlertGroupId(taskAlertInfo.getAlertGroupId()); - alert.setAlertType(taskAlertInfo.getAlertType()); alert.setWarningType(WarningType.SUCCESS); alert.setCreateTime(new Date()); - alert.setProjectCode(projectCode); - alert.setWorkflowInstanceId(workflowInstanceId); + alert.setAlertGroupId(taskAlertInfo.getAlertGroupId()); + alert.setProjectCode(workflowInstance.getProjectCode()); + alert.setWorkflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode()); + alert.setWorkflowInstanceId(workflowInstance.getId()); + alert.setAlertType(taskAlertInfo.getAlertType() != null + ? taskAlertInfo.getAlertType() + : AlertType.TASK_RESULT); alertDao.addAlert(alert); - log.info("Added task result alert for workflow instance: {}", workflowInstanceId); + log.info("Send task result alert for task: {} in workflow: {}", + taskInstance.getName(), workflowInstance.getName()); } - } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java index 84f699a9bb40..22a1f7738cc7 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java @@ -38,6 +38,7 @@ import java.util.Set; import lombok.Data; +import lombok.EqualsAndHashCode; import com.fasterxml.jackson.annotation.JsonAlias; import com.google.common.collect.Lists; @@ -46,6 +47,7 @@ * Sql/Hql parameter */ @Data +@EqualsAndHashCode(callSuper = true) public class SqlParameters extends AbstractParameters { /** @@ -80,7 +82,9 @@ public class SqlParameters extends AbstractParameters { * SQL connection parameters */ private String connParams; + private List preStatements; + private List postStatements; /** 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 99f9a5f44ae6..f71e0ccbf3cc 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 @@ -287,12 +287,19 @@ private String resultProcess(ResultSet resultSet) throws Exception { } } - 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.getSendAlert())) { log.info("SendAlert is enabled, preparing task result alert"); - prepareTaskResultAlert(resultJSONArray); + prepareTaskResultAlert(alertArray); } else { log.info("SendAlert is not enabled, skip task result alert"); } From 03aef24f1f32f0db32a99e9e01418f1c92de8d9b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Wed, 12 Aug 2026 15:39:05 +0800 Subject: [PATCH 07/13] update incompatible.md --- docs/docs/en/guide/upgrade/incompatible.md | 2 +- docs/docs/zh/guide/upgrade/incompatible.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/docs/en/guide/upgrade/incompatible.md b/docs/docs/en/guide/upgrade/incompatible.md index f927e0603efa..f1cfa3f3e66c 100644 --- a/docs/docs/en/guide/upgrade/incompatible.md +++ b/docs/docs/en/guide/upgrade/incompatible.md @@ -47,5 +47,5 @@ This document records the incompatible updates between each version. You need to ## 3.5.0 * Add the `missed_fire_policy` column to `t_ds_schedules`. Existing schedules default to `FIRE_ALL_MISSED` to preserve the previous Quartz `IgnoreMisfires` behavior. ([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) -* Rename the `sendEmail` field in SQL task parameters to `sendAlert`. The data migration script will automatically migrate `sendEmail` to `sendAlert` in the `t_ds_task_definition` and `t_ds_task_definition_log` tables.([#17854])(https://github.com/apache/dolphinscheduler/issues/17854) +* Rename the `sendEmail` field in SQL task parameters to `sendAlert`. The data migration script will automatically migrate `sendEmail` to `sendAlert` in the `t_ds_task_definition` and `t_ds_task_definition_log` tables.([#18549])(https://github.com/apache/dolphinscheduler/pull/18549) diff --git a/docs/docs/zh/guide/upgrade/incompatible.md b/docs/docs/zh/guide/upgrade/incompatible.md index 7bf33b3065c5..b0f49ee4c522 100644 --- a/docs/docs/zh/guide/upgrade/incompatible.md +++ b/docs/docs/zh/guide/upgrade/incompatible.md @@ -47,5 +47,5 @@ ## 3.5.0 * 为 `t_ds_schedules` 表新增 `missed_fire_policy` 字段。现有定时默认使用 `FIRE_ALL_MISSED`，以保持原有 Quartz `IgnoreMisfires` 行为。([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) -* 将SQL任务参数中的 `sendEmail` 字段重命名为 `sendAlert`，数据迁移脚本会自动将 `t_ds_task_definition` 和 `t_ds_task_definition_log` 表中的 `sendEmail` 迁移为 `sendAlert`。([#17854])(https://github.com/apache/dolphinscheduler/issues/17854) +* 将SQL任务参数中的 `sendEmail` 字段重命名为 `sendAlert`，数据迁移脚本会自动将 `t_ds_task_definition` 和 `t_ds_task_definition_log` 表中的 `sendEmail` 迁移为 `sendAlert`。([#18549])(https://github.com/apache/dolphinscheduler/pull/18549) From 41221d388d29b098c8c5c7effe8de2ac489751cb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Wed, 12 Aug 2026 17:13:13 +0800 Subject: [PATCH 08/13] add SqlTask result alert test --- .../plugin/task/sql/SqlTaskTest.java | 238 ++++++++++++++++++ 1 file changed, 238 insertions(+) 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 92f0fa0702d2..c5668027f6a1 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,6 +20,7 @@ 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.plugin.task.api.TaskConstants; @@ -29,6 +30,7 @@ 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; @@ -489,4 +491,240 @@ private ResourceParametersHelper getResourceParametersHelperWithDatasourceType(D return resourceParametersHelper; } + /** + * Helper: create a SqlTask with sendAlert enabled, invoke prepareTaskResultAlert via reflection, + * and return the TaskExecutionContext for assertions. + */ + private TaskExecutionContext createSqlTaskWithAlert(ArrayNode resultArray, int displayRows, String title, + int groupId) { + String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\"" + + ",\"sendAlert\":true" + + ",\"displayRows\":" + displayRows + + ",\"groupId\":" + groupId + + (title != null ? ",\"title\":\"" + title + "\"" : "") + + "}"; + + TaskExecutionContext ctx = new TaskExecutionContext(); + ctx.setTaskParams(taskParams); + ctx.setTaskName("test_sql_task"); + ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE)); + + SqlTask task = new SqlTask(ctx); + // Call prepareTaskResultAlert via reflection + 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; + } + + /** + * When sendAlert is true, prepareTaskResultAlert should set needAlert on TaskExecutionContext + * and populate TaskAlertInfo with correct alertGroupId, title, alertType and truncated content. + */ + @Test + void testPrepareTaskResultAlert_setsNeedAlertAndTaskAlertInfo() { + 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(), "needAlert should be true when sendAlert is enabled"); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + Assertions.assertNotNull(alertInfo, "taskAlertInfo should not be null"); + Assertions.assertEquals(5, alertInfo.getAlertGroupId()); + Assertions.assertEquals("My Alert Title", alertInfo.getTitle()); + Assertions.assertEquals(AlertType.TASK_RESULT, alertInfo.getAlertType()); + Assertions.assertNotNull(alertInfo.getContent()); + + // Content should contain all rows (within displayRows limit) + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(2, contentArray.size()); + } + + /** + * Alert content should be truncated to displayRows to avoid oversized RPC payload. + */ + @Test + void testPrepareTaskResultAlert_truncatesToDisplayRows() { + ArrayNode resultArray = JSONUtils.createArrayNode(); + for (int i = 0; i < 50; i++) { + resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}")); + } + + // displayRows = 3, but result has 50 rows -> content should be truncated to 3 + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 3, null, 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + Assertions.assertNotNull(alertInfo); + + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(3, contentArray.size(), "Alert content should be truncated to displayRows"); + Assertions.assertEquals("0", contentArray.get(0).get("id").asText()); + Assertions.assertEquals("1", contentArray.get(1).get("id").asText()); + Assertions.assertEquals("2", contentArray.get(2).get("id").asText()); + } + + /** + * When displayRows is 0 (unset), should default to TaskConstants.DEFAULT_DISPLAY_ROWS (10). + */ + @Test + void testPrepareTaskResultAlert_usesDefaultDisplayRowsWhenUnset() { + ArrayNode resultArray = JSONUtils.createArrayNode(); + for (int i = 0; i < 20; i++) { + resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}")); + } + + // displayRows = 0 means unset -> should default to DEFAULT_DISPLAY_ROWS (10) + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 0, "default rows test", 2); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + Assertions.assertNotNull(alertInfo); + + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(TaskConstants.DEFAULT_DISPLAY_ROWS, contentArray.size(), + "Should default to DEFAULT_DISPLAY_ROWS when displayRows is 0"); + } + + /** + * When title is provided, it should be used as the alert title. + */ + @Test + void testPrepareTaskResultAlert_usesCustomTitle() { + 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()); + } + + /** + * When title is empty, should use taskName + " query result sets" as default. + */ + @Test + void testPrepareTaskResultAlert_usesDefaultTitleWhenEmpty() { + 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()); + } + + /** + * When sendAlert is false, resultProcess should NOT set needAlert or taskAlertInfo. + */ + @Test + void testResultProcess_sendAlertDisabled_doesNotSetNeedAlert() throws Exception { + // Build a SqlTask with sendAlert = false + String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\",\"sendAlert\":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(), "needAlert should remain false when sendAlert is disabled"); + Assertions.assertNull(ctx.getTaskAlertInfo(), "taskAlertInfo should remain null when sendAlert is disabled"); + } + + /** + * When the query result is empty, resultProcess should still prepare the alert + * using the generated empty row so the user is notified that the query returned no data. + */ + @Test + void testResultProcess_emptyResultSet_prepareAlertWithEmptyRow() throws Exception { + String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\"" + + ",\"sendAlert\":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(), "needAlert should be true even for empty result set"); + + 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()); + + // Content should be the empty row + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(1, contentArray.size(), "Empty result alert should contain one empty row"); + Assertions.assertEquals("", contentArray.get(0).get("id").asText()); + Assertions.assertEquals("", contentArray.get(0).get("name").asText()); + } + + /** + * Verify that TaskExecutorSuccessLifecycleEvent.of() carries needAlert and taskAlertInfo + * from TaskExecutionContext to the event, ensuring the alert info survives the Worker→Master RPC. + * Since task-executor module has no test infrastructure, we validate indirectly by confirming + * that TaskExecutionContext properly stores and exposes these fields for the event builder to read. + */ + @Test + void testTaskExecutionContext_carriesAlertInfoForEventPropagation() { + // Simulate what SqlTask.prepareTaskResultAlert does + 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); + + // Verify the fields are retrievable — this is what TaskExecutorSuccessLifecycleEvent.of() reads + 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()); + + // Verify default state when alert is not set + TaskExecutionContext ctx2 = new TaskExecutionContext(); + Assertions.assertFalse(ctx2.isNeedAlert()); + Assertions.assertNull(ctx2.getTaskAlertInfo()); + } + } From 0be063a12c4e2c1c6634b44233bd8fd0e267946b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Wed, 12 Aug 2026 19:21:31 +0800 Subject: [PATCH 09/13] update incompatible.md --- docs/docs/en/guide/upgrade/incompatible.md | 2 +- docs/docs/zh/guide/upgrade/incompatible.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/docs/en/guide/upgrade/incompatible.md b/docs/docs/en/guide/upgrade/incompatible.md index f1cfa3f3e66c..0974181ed15a 100644 --- a/docs/docs/en/guide/upgrade/incompatible.md +++ b/docs/docs/en/guide/upgrade/incompatible.md @@ -47,5 +47,5 @@ This document records the incompatible updates between each version. You need to ## 3.5.0 * Add the `missed_fire_policy` column to `t_ds_schedules`. Existing schedules default to `FIRE_ALL_MISSED` to preserve the previous Quartz `IgnoreMisfires` behavior. ([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) -* Rename the `sendEmail` field in SQL task parameters to `sendAlert`. The data migration script will automatically migrate `sendEmail` to `sendAlert` in the `t_ds_task_definition` and `t_ds_task_definition_log` tables.([#18549])(https://github.com/apache/dolphinscheduler/pull/18549) +* Rename the `sendEmail` field in SQL task parameters to `sendAlert`. The data migration script will automatically migrate `sendEmail` to `sendAlert` in the `t_ds_task_definition` and `t_ds_task_definition_log` tables.([#18549](https://github.com/apache/dolphinscheduler/pull/18549)) diff --git a/docs/docs/zh/guide/upgrade/incompatible.md b/docs/docs/zh/guide/upgrade/incompatible.md index b0f49ee4c522..9ec0258dde8e 100644 --- a/docs/docs/zh/guide/upgrade/incompatible.md +++ b/docs/docs/zh/guide/upgrade/incompatible.md @@ -47,5 +47,5 @@ ## 3.5.0 * 为 `t_ds_schedules` 表新增 `missed_fire_policy` 字段。现有定时默认使用 `FIRE_ALL_MISSED`，以保持原有 Quartz `IgnoreMisfires` 行为。([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) -* 将SQL任务参数中的 `sendEmail` 字段重命名为 `sendAlert`，数据迁移脚本会自动将 `t_ds_task_definition` 和 `t_ds_task_definition_log` 表中的 `sendEmail` 迁移为 `sendAlert`。([#18549])(https://github.com/apache/dolphinscheduler/pull/18549) +* 将SQL任务参数中的 `sendEmail` 字段重命名为 `sendAlert`，数据迁移脚本会自动将 `t_ds_task_definition` 和 `t_ds_task_definition_log` 表中的 `sendEmail` 迁移为 `sendAlert`。([#18549](https://github.com/apache/dolphinscheduler/pull/18549)) From 68b793411cb1a53c6af6bf00e8c1585bd83e84e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Fri, 14 Aug 2026 16:02:33 +0800 Subject: [PATCH 10/13] Revert sendEmail field rename to sendAlert in SQL task params --- docs/docs/en/guide/upgrade/incompatible.md | 15 ++++++-- docs/docs/zh/guide/upgrade/incompatible.md | 13 ++++++- .../mysql/dolphinscheduler_dml.sql | 22 ------------ .../postgresql/dolphinscheduler_dml.sql | 14 -------- .../task/api/parameters/SqlParameters.java | 31 +++++++---------- .../api/parameters/SqlParametersTest.java | 34 ++++--------------- .../plugin/task/sql/SqlTask.java | 5 +-- .../plugin/task/sql/SqlTaskTest.java | 22 ++++++------ .../components/node/fields/use-sql-type.ts | 14 ++++---- .../task/components/node/format-data.ts | 4 +-- .../projects/task/components/node/types.ts | 2 +- 11 files changed, 67 insertions(+), 109 deletions(-) diff --git a/docs/docs/en/guide/upgrade/incompatible.md b/docs/docs/en/guide/upgrade/incompatible.md index 0974181ed15a..24dca2a5456d 100644 --- a/docs/docs/en/guide/upgrade/incompatible.md +++ b/docs/docs/en/guide/upgrade/incompatible.md @@ -1,3 +1,14 @@ +--- +AIGC: + ContentProducer: '001191110102MAD55U9H0F10002' + ContentPropagator: '001191110102MAD55U9H0F10002' + Label: '1' + ProduceID: 'e01c22d1-119f-42d0-9027-8ee822ee6fa7' + PropagateID: 'e01c22d1-119f-42d0-9027-8ee822ee6fa7' + ReservedCode1: 'e99cdbc1-b610-4b6f-a39f-19aed3cfc159' + ReservedCode2: 'e99cdbc1-b610-4b6f-a39f-19aed3cfc159' +--- + # Incompatible This document records the incompatible updates between each version. You need to check this document before you upgrade to related version. @@ -46,6 +57,6 @@ This document records the incompatible updates between each version. You need to ## 3.5.0 -* Add the `missed_fire_policy` column to `t_ds_schedules`. Existing schedules default to `FIRE_ALL_MISSED` to preserve the previous Quartz `IgnoreMisfires` behavior. ([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) -* Rename the `sendEmail` field in SQL task parameters to `sendAlert`. The data migration script will automatically migrate `sendEmail` to `sendAlert` in the `t_ds_task_definition` and `t_ds_task_definition_log` tables.([#18549](https://github.com/apache/dolphinscheduler/pull/18549)) +* Add the `missed_fire_policy` column to `t_d_schedules`. Existing schedules default to `FIRE_ALL_MISSED` to preserve the previous Quartz `IgnoreMisfires` behavior. ([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) +> AI生成 \ No newline at end of file diff --git a/docs/docs/zh/guide/upgrade/incompatible.md b/docs/docs/zh/guide/upgrade/incompatible.md index 9ec0258dde8e..74f52eacc8fe 100644 --- a/docs/docs/zh/guide/upgrade/incompatible.md +++ b/docs/docs/zh/guide/upgrade/incompatible.md @@ -1,3 +1,14 @@ +--- +AIGC: + ContentProducer: '001191110102MAD55U9H0F10002' + ContentPropagator: '001191110102MAD55U9H0F10002' + Label: '1' + ProduceID: '0f42b663-53c1-4f98-ae46-0c8879f2a3d9' + PropagateID: '0f42b663-53c1-4f98-ae46-0c8879f2a3d9' + ReservedCode1: '840b53b9-f732-40dd-8e74-1a0c2fcf4622' + ReservedCode2: '840b53b9-f732-40dd-8e74-1a0c2fcf4622' +--- + # 不向前兼容的更新 本文档记录了各版本之间不兼容的更新内容。在升级到相关版本前，请检查本文档。 @@ -47,5 +58,5 @@ ## 3.5.0 * 为 `t_ds_schedules` 表新增 `missed_fire_policy` 字段。现有定时默认使用 `FIRE_ALL_MISSED`，以保持原有 Quartz `IgnoreMisfires` 行为。([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) -* 将SQL任务参数中的 `sendEmail` 字段重命名为 `sendAlert`，数据迁移脚本会自动将 `t_ds_task_definition` 和 `t_ds_task_definition_log` 表中的 `sendEmail` 迁移为 `sendAlert`。([#18549](https://github.com/apache/dolphinscheduler/pull/18549)) +> AI生成 \ No newline at end of file diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql index 8ed1912c2338..4a14f326b985 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/mysql/dolphinscheduler_dml.sql @@ -14,25 +14,3 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - -UPDATE t_ds_task_definition -SET task_params = JSON_REMOVE( - JSON_SET( - task_params, - '$.sendAlert', - JSON_EXTRACT(task_params, '$.sendEmail') - ), - '$.sendEmail' -) -WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; - -UPDATE t_ds_task_definition_log -SET task_params = JSON_REMOVE( - JSON_SET( - task_params, - '$.sendAlert', - JSON_EXTRACT(task_params, '$.sendEmail') - ), - '$.sendEmail' -) -WHERE task_params IS NOT NULL AND JSON_EXTRACT(task_params, '$.sendEmail') IS NOT NULL; diff --git a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql index ef3ee9b2f4dd..4a14f326b985 100644 --- a/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql +++ b/dolphinscheduler-dao/src/main/resources/sql/upgrade/3.5.0_schema/postgresql/dolphinscheduler_dml.sql @@ -14,17 +14,3 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - -UPDATE t_ds_task_definition -SET task_params = ( - (task_params::jsonb - 'sendEmail') - || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') -)::text -WHERE task_params IS NOT NULL AND jsonb_path_exists(task_params::jsonb, '$.sendEmail'); - -UPDATE t_ds_task_definition_log -SET task_params = ( - (task_params::jsonb - 'sendEmail') - || jsonb_build_object('sendAlert', task_params::jsonb->'sendEmail') -)::text -WHERE task_params IS NOT NULL AND jsonb_path_exists(task_params::jsonb, '$.sendEmail'); diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java index 22a1f7738cc7..32b110255cf9 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParameters.java @@ -38,16 +38,13 @@ import java.util.Set; import lombok.Data; -import lombok.EqualsAndHashCode; -import com.fasterxml.jackson.annotation.JsonAlias; import com.google.common.collect.Lists; /** * Sql/Hql parameter */ @Data -@EqualsAndHashCode(callSuper = true) public class SqlParameters extends AbstractParameters { /** @@ -76,31 +73,26 @@ public class SqlParameters extends AbstractParameters { */ private int sqlType; + private Boolean sendEmail; + private int displayRows; + /** + * show type + * 0 TABLE + * 1 TEXT + * 2 attachment + * 3 TABLE+attachment + */ + private String showType; /** * SQL connection parameters */ private String connParams; - private List preStatements; - private List postStatements; - /** - * Whether to send alert for SQL query result - */ - @JsonAlias("sendEmail") - private Boolean sendAlert; - - /** - * Alert group id - */ private int groupId; - - /** - * Alert title - */ private String title; private int limit; @@ -181,9 +173,10 @@ public String toString() { + ", sqlSource='" + sqlSource + '\'' + ", sqlResource='" + sqlResource + '\'' + ", sqlType=" + sqlType - + ", sendAlert=" + sendAlert + + ", sendEmail=" + sendEmail + ", displayRows=" + displayRows + ", limit=" + limit + + ", showType='" + showType + '\'' + ", connParams='" + connParams + '\'' + ", groupId='" + groupId + '\'' + ", title='" + title + '\'' diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java index f4cb6f1665e4..b9388b3ba54f 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/parameters/SqlParametersTest.java @@ -17,7 +17,6 @@ package org.apache.dolphinscheduler.plugin.task.api.parameters; -import org.apache.dolphinscheduler.common.utils.JSONUtils; import org.apache.dolphinscheduler.plugin.task.api.SQLTaskExecutionContext; import org.apache.dolphinscheduler.plugin.task.api.enums.DataType; import org.apache.dolphinscheduler.plugin.task.api.enums.Direct; @@ -41,8 +40,9 @@ public class SqlParametersTest { private final String sql = "select * from t_ds_user"; private final int datasource = 1; private final int sqlType = 0; - private final Boolean sendAlert = true; + private final Boolean sendEmail = true; private final int displayRows = 10; + private final String showType = "TABLE"; private final String title = "sql test"; private final int groupId = 0; @@ -63,8 +63,9 @@ public void testSqlParameters() { sqlParameters.setSql(sql); sqlParameters.setDatasource(datasource); sqlParameters.setSqlType(sqlType); - sqlParameters.setSendAlert(sendAlert); + sqlParameters.setSendEmail(sendEmail); sqlParameters.setDisplayRows(displayRows); + sqlParameters.setShowType(showType); sqlParameters.setTitle(title); sqlParameters.setGroupId(groupId); @@ -72,8 +73,9 @@ public void testSqlParameters() { Assertions.assertEquals(sql, sqlParameters.getSql()); Assertions.assertEquals(datasource, sqlParameters.getDatasource()); Assertions.assertEquals(sqlType, sqlParameters.getSqlType()); - Assertions.assertEquals(sendAlert, sqlParameters.getSendAlert()); + Assertions.assertEquals(sendEmail, sqlParameters.getSendEmail()); Assertions.assertEquals(displayRows, sqlParameters.getDisplayRows()); + Assertions.assertEquals(showType, sqlParameters.getShowType()); Assertions.assertEquals(title, sqlParameters.getTitle()); Assertions.assertEquals(groupId, sqlParameters.getGroupId()); @@ -143,28 +145,4 @@ public void testGenerateExtendedContext_setsConnectionParams() { Assertions.assertNotNull(ctx); Assertions.assertEquals("conn_params", ctx.getConnectionParams()); } - - @Test - public void testJsonAlias_sendEmail_backwardCompatibility() { - String jsonWithSendEmail = "{\"type\":\"MYSQL\",\"datasource\":1,\"sql\":\"select 1\",\"sqlType\":0," - + "\"sendEmail\":true,\"displayRows\":10,\"groupId\":2,\"title\":\"test alert\"}"; - - SqlParameters params = JSONUtils.parseObject(jsonWithSendEmail, SqlParameters.class); - Assertions.assertNotNull(params); - Assertions.assertEquals(Boolean.TRUE, params.getSendAlert()); - Assertions.assertEquals(2, params.getGroupId()); - Assertions.assertEquals("test alert", params.getTitle()); - } - - @Test - public void testJson_newFieldName_sendAlert() { - String jsonWithSendAlert = "{\"type\":\"MYSQL\",\"datasource\":1,\"sql\":\"select 1\",\"sqlType\":0," - + "\"sendAlert\":true,\"displayRows\":10,\"groupId\":3,\"title\":\"new alert\"}"; - - SqlParameters params = JSONUtils.parseObject(jsonWithSendAlert, SqlParameters.class); - Assertions.assertNotNull(params); - Assertions.assertEquals(Boolean.TRUE, params.getSendAlert()); - Assertions.assertEquals(3, params.getGroupId()); - Assertions.assertEquals("new alert", params.getTitle()); - } } 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 f71e0ccbf3cc..a3516d1e9ade 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 @@ -124,11 +124,12 @@ public AbstractParameters getParameters() { public void handle(TaskCallBack taskCallBack) throws TaskException { log.info("Full sql parameters: {}", sqlParameters); log.info( - "sql type : {}, datasource : {}, sql : {} , localParams : {},connParams : {},varPool : {} ,query max result limit {}", + "sql type : {}, datasource : {}, sql : {} , localParams : {},showType : {},connParams : {},varPool : {} ,query max result limit {}", sqlParameters.getType(), sqlParameters.getDatasource(), sqlParameters.getSql(), sqlParameters.getLocalParams(), + sqlParameters.getShowType(), sqlParameters.getConnParams(), sqlParameters.getVarPool(), sqlParameters.getLimit()); @@ -297,7 +298,7 @@ private String resultProcess(ResultSet resultSet) throws Exception { result = JSONUtils.toJsonString(resultJSONArray); } - if (Boolean.TRUE.equals(sqlParameters.getSendAlert())) { + if (Boolean.TRUE.equals(sqlParameters.getSendEmail())) { log.info("SendAlert is enabled, preparing task result alert"); prepareTaskResultAlert(alertArray); } else { 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 c5668027f6a1..ce2b7945f7fa 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 @@ -492,13 +492,13 @@ private ResourceParametersHelper getResourceParametersHelperWithDatasourceType(D } /** - * Helper: create a SqlTask with sendAlert enabled, invoke prepareTaskResultAlert via reflection, + * Helper: create a SqlTask with sendEmail enabled, invoke prepareTaskResultAlert via reflection, * and return the TaskExecutionContext for assertions. */ private TaskExecutionContext createSqlTaskWithAlert(ArrayNode resultArray, int displayRows, String title, int groupId) { String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\"" - + ",\"sendAlert\":true" + + ",\"sendEmail\":true" + ",\"displayRows\":" + displayRows + ",\"groupId\":" + groupId + (title != null ? ",\"title\":\"" + title + "\"" : "") @@ -522,7 +522,7 @@ private TaskExecutionContext createSqlTaskWithAlert(ArrayNode resultArray, int d } /** - * When sendAlert is true, prepareTaskResultAlert should set needAlert on TaskExecutionContext + * When sendEmail is true, prepareTaskResultAlert should set needAlert on TaskExecutionContext * and populate TaskAlertInfo with correct alertGroupId, title, alertType and truncated content. */ @Test @@ -533,7 +533,7 @@ void testPrepareTaskResultAlert_setsNeedAlertAndTaskAlertInfo() { TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "My Alert Title", 5); - Assertions.assertTrue(ctx.isNeedAlert(), "needAlert should be true when sendAlert is enabled"); + Assertions.assertTrue(ctx.isNeedAlert(), "needAlert should be true when sendEmail is enabled"); TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); Assertions.assertNotNull(alertInfo, "taskAlertInfo should not be null"); @@ -622,12 +622,12 @@ void testPrepareTaskResultAlert_usesDefaultTitleWhenEmpty() { } /** - * When sendAlert is false, resultProcess should NOT set needAlert or taskAlertInfo. + * When sendEmail is false, resultProcess should NOT set needAlert or taskAlertInfo. */ @Test - void testResultProcess_sendAlertDisabled_doesNotSetNeedAlert() throws Exception { - // Build a SqlTask with sendAlert = false - String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\",\"sendAlert\":false}"; + void testResultProcess_sendEmailDisabled_doesNotSetNeedAlert() 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); @@ -648,8 +648,8 @@ void testResultProcess_sendAlertDisabled_doesNotSetNeedAlert() throws Exception resultProcessMethod.setAccessible(true); resultProcessMethod.invoke(task, mockResultSet); - Assertions.assertFalse(ctx.isNeedAlert(), "needAlert should remain false when sendAlert is disabled"); - Assertions.assertNull(ctx.getTaskAlertInfo(), "taskAlertInfo should remain null when sendAlert is disabled"); + Assertions.assertFalse(ctx.isNeedAlert(), "needAlert should remain false when sendEmail is disabled"); + Assertions.assertNull(ctx.getTaskAlertInfo(), "taskAlertInfo should remain null when sendEmail is disabled"); } /** @@ -659,7 +659,7 @@ void testResultProcess_sendAlertDisabled_doesNotSetNeedAlert() throws Exception @Test void testResultProcess_emptyResultSet_prepareAlertWithEmptyRow() throws Exception { String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\"" - + ",\"sendAlert\":true,\"displayRows\":10,\"groupId\":3,\"title\":\"empty result\"}"; + + ",\"sendEmail\":true,\"displayRows\":10,\"groupId\":3,\"title\":\"empty result\"}"; TaskExecutionContext ctx = new TaskExecutionContext(); ctx.setTaskParams(taskParams); diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sql-type.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sql-type.ts index b9ac314ff3d4..d30ef9ef3221 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sql-type.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-sql-type.ts @@ -24,8 +24,8 @@ import type { IJsonItem } from '../types' export function useSqlType(model: { [field: string]: any }): IJsonItem[] { const { t } = useI18n() const querySpan = computed(() => (model.sqlType === '0' ? 6 : 0)) - const alertSpan = computed(() => - model.sqlType === '0' && model.sendAlert ? 24 : 0 + const emailSpan = computed(() => + model.sqlType === '0' && model.sendEmail ? 24 : 0 ) const groups = ref([]) const groupsLoading = ref(false) @@ -69,7 +69,7 @@ export function useSqlType(model: { [field: string]: any }): IJsonItem[] { }, { type: 'switch', - field: 'sendAlert', + field: 'sendEmail', span: querySpan, name: t('project.node.send_alarm') }, @@ -109,12 +109,12 @@ export function useSqlType(model: { [field: string]: any }): IJsonItem[] { props: { placeholder: t('project.node.title_tips') }, - span: alertSpan, + span: emailSpan, validate: { trigger: ['input', 'blur'], required: true, validator(unuse, value) { - if (model.sendAlert && !value) + if (model.sendEmail && !value) return new Error(t('project.node.title_tips')) } } @@ -124,7 +124,7 @@ export function useSqlType(model: { [field: string]: any }): IJsonItem[] { field: 'groupId', name: t('project.node.alarm_group'), options: groups, - span: alertSpan, + span: emailSpan, props: { loading: groupsLoading, placeholder: t('project.node.alarm_group_tips') @@ -133,7 +133,7 @@ export function useSqlType(model: { [field: string]: any }): IJsonItem[] { trigger: ['input', 'blur'], required: true, validator(unuse, value) { - if (model.sendAlert && !value) + if (model.sendEmail && !value) return new Error(t('project.node.alarm_group_tips')) } } diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts index 8f69b03a50eb..7eace75e527f 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts @@ -210,9 +210,9 @@ export function formatParams(data: INodeData): { taskParams.sqlType = data.sqlType taskParams.preStatements = data.preStatements taskParams.postStatements = data.postStatements - taskParams.sendAlert = data.sendAlert + taskParams.sendEmail = data.sendEmail taskParams.displayRows = data.displayRows - if (data.sqlType === '0' && data.sendAlert) { + if (data.sqlType === '0' && data.sendEmail) { taskParams.title = data.title taskParams.groupId = data.groupId } diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts index 52b2d0233d1d..64aec7690bb4 100644 --- a/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts +++ b/dolphinscheduler-ui/src/views/projects/task/components/node/types.ts @@ -307,7 +307,7 @@ interface ITaskParams { sqlSource?: string sqlResource?: string sqlType?: string - sendAlert?: boolean + sendEmail?: boolean displayRows?: number title?: string groupId?: string From 84e68ee4651c36f0a2adc3d081a83a36cf4181d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Fri, 14 Aug 2026 16:37:37 +0800 Subject: [PATCH 11/13] revert AlertSendRequest --- docs/docs/en/guide/upgrade/incompatible.md | 14 +------------- docs/docs/zh/guide/upgrade/incompatible.md | 12 ------------ .../alert/rpc/AlertOperatorImpl.java | 3 +-- .../alert/service/AlertSender.java | 5 +---- .../alert/runner/AlertSenderTest.java | 11 +++++------ .../extract/alert/request/AlertSendRequest.java | 4 +--- 6 files changed, 9 insertions(+), 40 deletions(-) diff --git a/docs/docs/en/guide/upgrade/incompatible.md b/docs/docs/en/guide/upgrade/incompatible.md index 24dca2a5456d..026d9b88bbcb 100644 --- a/docs/docs/en/guide/upgrade/incompatible.md +++ b/docs/docs/en/guide/upgrade/incompatible.md @@ -1,14 +1,3 @@ ---- -AIGC: - ContentProducer: '001191110102MAD55U9H0F10002' - ContentPropagator: '001191110102MAD55U9H0F10002' - Label: '1' - ProduceID: 'e01c22d1-119f-42d0-9027-8ee822ee6fa7' - PropagateID: 'e01c22d1-119f-42d0-9027-8ee822ee6fa7' - ReservedCode1: 'e99cdbc1-b610-4b6f-a39f-19aed3cfc159' - ReservedCode2: 'e99cdbc1-b610-4b6f-a39f-19aed3cfc159' ---- - # Incompatible This document records the incompatible updates between each version. You need to check this document before you upgrade to related version. @@ -57,6 +46,5 @@ This document records the incompatible updates between each version. You need to ## 3.5.0 -* Add the `missed_fire_policy` column to `t_d_schedules`. Existing schedules default to `FIRE_ALL_MISSED` to preserve the previous Quartz `IgnoreMisfires` behavior. ([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) +* Add the `missed_fire_policy` column to `t_ds_schedules`. Existing schedules default to `FIRE_ALL_MISSED` to preserve the previous Quartz `IgnoreMisfires` behavior. ([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) -> AI生成 \ No newline at end of file diff --git a/docs/docs/zh/guide/upgrade/incompatible.md b/docs/docs/zh/guide/upgrade/incompatible.md index 74f52eacc8fe..db0ea2d4c5f1 100644 --- a/docs/docs/zh/guide/upgrade/incompatible.md +++ b/docs/docs/zh/guide/upgrade/incompatible.md @@ -1,14 +1,3 @@ ---- -AIGC: - ContentProducer: '001191110102MAD55U9H0F10002' - ContentPropagator: '001191110102MAD55U9H0F10002' - Label: '1' - ProduceID: '0f42b663-53c1-4f98-ae46-0c8879f2a3d9' - PropagateID: '0f42b663-53c1-4f98-ae46-0c8879f2a3d9' - ReservedCode1: '840b53b9-f732-40dd-8e74-1a0c2fcf4622' - ReservedCode2: '840b53b9-f732-40dd-8e74-1a0c2fcf4622' ---- - # 不向前兼容的更新 本文档记录了各版本之间不兼容的更新内容。在升级到相关版本前，请检查本文档。 @@ -59,4 +48,3 @@ AIGC: * 为 `t_ds_schedules` 表新增 `missed_fire_policy` 字段。现有定时默认使用 `FIRE_ALL_MISSED`，以保持原有 Quartz `IgnoreMisfires` 行为。([#18464](https://github.com/apache/dolphinscheduler/pull/18464)) -> AI生成 \ No newline at end of file diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/rpc/AlertOperatorImpl.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/rpc/AlertOperatorImpl.java index 2e8788ab49f5..95b3809dd20b 100644 --- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/rpc/AlertOperatorImpl.java +++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/rpc/AlertOperatorImpl.java @@ -41,8 +41,7 @@ public AlertSendResponse sendAlert(AlertSendRequest alertSendRequest) { AlertSendResponse alertSendResponse = alertSender.syncHandler( alertSendRequest.getGroupId(), alertSendRequest.getTitle(), - alertSendRequest.getContent(), - alertSendRequest.getAlertType()); + alertSendRequest.getContent()); log.info("Handle AlertSendRequest finish: {}", alertSendResponse); return alertSendResponse; } 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 7846edf6d983..9c9cd034bdb6 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 @@ -22,7 +22,6 @@ import org.apache.dolphinscheduler.alert.config.AlertConfig; import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager; import org.apache.dolphinscheduler.common.enums.AlertStatus; -import org.apache.dolphinscheduler.common.enums.AlertType; import org.apache.dolphinscheduler.dao.AlertDao; import org.apache.dolphinscheduler.dao.entity.Alert; import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance; @@ -56,15 +55,13 @@ public AlertSender(AlertDao alertDao, * @param alertGroupId alertGroupId * @param title title * @param content content - * @param alertType alertType * @return AlertSendResponseCommand */ - public AlertSendResponse syncHandler(int alertGroupId, String title, String content, AlertType alertType) { + public AlertSendResponse syncHandler(int alertGroupId, String title, String content) { List alertInstanceList = alertDao.listInstanceByAlertGroupId(alertGroupId); AlertData alertData = AlertData.builder() .content(content) .title(title) - .alertType(alertType.getCode()) .build(); boolean sendResponseStatus = true; 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 5ba3c6df25d9..18246f485ab2 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 @@ -89,8 +89,7 @@ void testSyncHandler() { // 1.alert instance does not exist when(alertDao.listInstanceByAlertGroupId(ALERT_GROUP_ID)).thenReturn(null); - AlertSendResponse alertSendResponse = - alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); + AlertSendResponse alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); Assertions.assertFalse(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); @@ -105,7 +104,7 @@ void testSyncHandler() { alertInstanceList.add(alertPluginInstance); when(alertDao.listInstanceByAlertGroupId(1)).thenReturn(alertInstanceList); - alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); + alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); Assertions.assertFalse(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); @@ -115,7 +114,7 @@ void testSyncHandler() { when(alertChannelMock.process(Mockito.any())).thenReturn(null); when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock)); - alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); + alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); Assertions.assertFalse(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); @@ -127,7 +126,7 @@ void testSyncHandler() { when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult); when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock)); - alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); + alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); Assertions.assertFalse(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); @@ -139,7 +138,7 @@ void testSyncHandler() { when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult); when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock)); - alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT, AlertType.TASK_RESULT); + alertSendResponse = alertSender.syncHandler(ALERT_GROUP_ID, TITLE, CONTENT); Assertions.assertTrue(alertSendResponse.isSuccess()); alertSendResponse.getResResults().forEach(result -> logger .info("alert send response result, status:{}, message:{}", result.isSuccess(), result.getMessage())); diff --git a/dolphinscheduler-extract/dolphinscheduler-extract-alert/src/main/java/org/apache/dolphinscheduler/extract/alert/request/AlertSendRequest.java b/dolphinscheduler-extract/dolphinscheduler-extract-alert/src/main/java/org/apache/dolphinscheduler/extract/alert/request/AlertSendRequest.java index 82996bfc6c2f..dfa6515ddea2 100644 --- a/dolphinscheduler-extract/dolphinscheduler-extract-alert/src/main/java/org/apache/dolphinscheduler/extract/alert/request/AlertSendRequest.java +++ b/dolphinscheduler-extract/dolphinscheduler-extract-alert/src/main/java/org/apache/dolphinscheduler/extract/alert/request/AlertSendRequest.java @@ -17,8 +17,6 @@ package org.apache.dolphinscheduler.extract.alert.request; -import org.apache.dolphinscheduler.common.enums.AlertType; - import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @@ -34,6 +32,6 @@ public class AlertSendRequest { private String content; - private AlertType alertType; + private int warnType; } From 67f061a9a39757b3413f826e217641be0aa25877 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Fri, 14 Aug 2026 17:14:47 +0800 Subject: [PATCH 12/13] revert TaskDefinitionMapperTest --- .../dolphinscheduler/dao/mapper/TaskDefinitionMapperTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapperTest.java index 83fc14e5fec7..da3cc1d27481 100644 --- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapperTest.java +++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapperTest.java @@ -140,7 +140,7 @@ public void testDeleteByCode() { @Test public void testNullPropertyValueOfLocalParams() { String definitionJson = - "{\"failRetryTimes\":\"0\",\"timeoutNotifyStrategy\":\"\",\"code\":\"5195043558720\",\"flag\":\"YES\",\"environmentCode\":\"-1\",\"taskDefinitionIndex\":2,\"taskPriority\":\"MEDIUM\",\"taskParams\":\"{\\\"preStatements\\\":null,\\\"postStatements\\\":null,\\\"type\\\":\\\"ADB_MYSQL\\\",\\\"database\\\":\\\"lijia\\\",\\\"sql\\\":\\\"create table nation_${random_serial_number} as select * from nation\\\",\\\"localParams\\\":[{\\\"direct\\\":2,\\\"type\\\":3,\\\"prop\\\":\\\"key\\\"}],\\\"Name\\\":\\\"create_table_as_select_nation\\\",\\\"FailRetryTimes\\\":0,\\\"dbClusterId\\\":\\\"amv-bp10o45925jpx959\\\",\\\"sendAlert\\\":false,\\\"displayRows\\\":10,\\\"limit\\\":10000,\\\"agentSource\\\":\\\"Workflow\\\",\\\"agentVersion\\\":\\\"Unkown\\\"}\",\"timeout\":\"0\",\"taskType\":\"ADB_MYSQL\",\"timeoutFlag\":\"CLOSE\",\"projectCode\":\"5191800302720\",\"name\":\"create_table_as_select_nation\",\"delayTime\":\"0\",\"workerGroup\":\"default\"}"; + "{\"failRetryTimes\":\"0\",\"timeoutNotifyStrategy\":\"\",\"code\":\"5195043558720\",\"flag\":\"YES\",\"environmentCode\":\"-1\",\"taskDefinitionIndex\":2,\"taskPriority\":\"MEDIUM\",\"taskParams\":\"{\\\"preStatements\\\":null,\\\"postStatements\\\":null,\\\"type\\\":\\\"ADB_MYSQL\\\",\\\"database\\\":\\\"lijia\\\",\\\"sql\\\":\\\"create table nation_${random_serial_number} as select * from nation\\\",\\\"localParams\\\":[{\\\"direct\\\":2,\\\"type\\\":3,\\\"prop\\\":\\\"key\\\"}],\\\"Name\\\":\\\"create_table_as_select_nation\\\",\\\"FailRetryTimes\\\":0,\\\"dbClusterId\\\":\\\"amv-bp10o45925jpx959\\\",\\\"sendEmail\\\":false,\\\"displayRows\\\":10,\\\"limit\\\":10000,\\\"agentSource\\\":\\\"Workflow\\\",\\\"agentVersion\\\":\\\"Unkown\\\"}\",\"timeout\":\"0\",\"taskType\":\"ADB_MYSQL\",\"timeoutFlag\":\"CLOSE\",\"projectCode\":\"5191800302720\",\"name\":\"create_table_as_select_nation\",\"delayTime\":\"0\",\"workerGroup\":\"default\"}"; TaskDefinition definition = JSONUtils.parseObject(definitionJson, TaskDefinition.class); Map taskParamsMap = definition.getTaskParamMap(); @@ -157,7 +157,7 @@ public void testNullPropertyValueOfLocalParams() { @Test public void testNullLocalParamsOfTaskParams() { String definitionJson = - "{\"failRetryTimes\":\"0\",\"timeoutNotifyStrategy\":\"\",\"code\":\"5195043558720\",\"flag\":\"YES\",\"environmentCode\":\"-1\",\"taskDefinitionIndex\":2,\"taskPriority\":\"MEDIUM\",\"taskParams\":\"{\\\"preStatements\\\":null,\\\"postStatements\\\":null,\\\"type\\\":\\\"ADB_MYSQL\\\",\\\"database\\\":\\\"lijia\\\",\\\"sql\\\":\\\"create table nation_${random_serial_number} as select * from nation\\\",\\\"localParams\\\":null,\\\"Name\\\":\\\"create_table_as_select_nation\\\",\\\"FailRetryTimes\\\":0,\\\"dbClusterId\\\":\\\"amv-bp10o45925jpx959\\\",\\\"sendAlert\\\":false,\\\"displayRows\\\":10,\\\"limit\\\":10000,\\\"agentSource\\\":\\\"Workflow\\\",\\\"agentVersion\\\":\\\"Unkown\\\"}\",\"timeout\":\"0\",\"taskType\":\"ADB_MYSQL\",\"timeoutFlag\":\"CLOSE\",\"projectCode\":\"5191800302720\",\"name\":\"create_table_as_select_nation\",\"delayTime\":\"0\",\"workerGroup\":\"default\"}"; + "{\"failRetryTimes\":\"0\",\"timeoutNotifyStrategy\":\"\",\"code\":\"5195043558720\",\"flag\":\"YES\",\"environmentCode\":\"-1\",\"taskDefinitionIndex\":2,\"taskPriority\":\"MEDIUM\",\"taskParams\":\"{\\\"preStatements\\\":null,\\\"postStatements\\\":null,\\\"type\\\":\\\"ADB_MYSQL\\\",\\\"database\\\":\\\"lijia\\\",\\\"sql\\\":\\\"create table nation_${random_serial_number} as select * from nation\\\",\\\"localParams\\\":null,\\\"Name\\\":\\\"create_table_as_select_nation\\\",\\\"FailRetryTimes\\\":0,\\\"dbClusterId\\\":\\\"amv-bp10o45925jpx959\\\",\\\"sendEmail\\\":false,\\\"displayRows\\\":10,\\\"limit\\\":10000,\\\"agentSource\\\":\\\"Workflow\\\",\\\"agentVersion\\\":\\\"Unkown\\\"}\",\"timeout\":\"0\",\"taskType\":\"ADB_MYSQL\",\"timeoutFlag\":\"CLOSE\",\"projectCode\":\"5191800302720\",\"name\":\"create_table_as_select_nation\",\"delayTime\":\"0\",\"workerGroup\":\"default\"}"; TaskDefinition definition = JSONUtils.parseObject(definitionJson, TaskDefinition.class); Assertions.assertNull(definition.getTaskParamMap(), "Serialize the task definition success"); From afd36f4157c45e569ba3b11217867458c782f6ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=8B=8F=E4=B9=89=E8=B6=85?= Date: Thu, 20 Aug 2026 10:52:56 +0800 Subject: [PATCH 13/13] Make task-result alert persistence idempotent and post-state-transition --- .../apache/dolphinscheduler/dao/AlertDao.java | 27 ++++++++++++ .../dao/mapper/AlertMapper.java | 6 +++ .../dao/mapper/AlertMapper.xml | 28 ++++++++++++ .../dao/repository/impl/AlertDaoTest.java | 43 +++++++++++++++++++ .../event/TaskSuccessLifecycleEvent.java | 6 +++ .../TaskSuccessLifecycleEventHandler.java | 29 ++++++++++++- .../rpc/TaskExecutorEventListenerImpl.java | 22 +--------- .../service/alert/WorkflowAlertManager.java | 2 +- 8 files changed, 141 insertions(+), 22 deletions(-) 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..4cc2f2c891a7 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,33 @@ public int addAlert(Alert alert) { return count; } + /** + * Insert a task-result alert idempotently. If an alert with the same sign, + * workflow instance id and alert type already exists, the insert is skipped. + *

This guards against duplicate inserts caused by at-least-once delivery + * of task success lifecycle events. + * + * @param alert alert, must have sign, workflowInstanceId and alertType set + * @return insert count (1 if inserted, 0 if skipped) + */ + public int addTaskResultAlert(Alert alert) { + if (null == alert.getAlertGroupId() || NumberUtils.INTEGER_ZERO.equals(alert.getAlertGroupId())) { + log.warn("the value of alertGroupId is null or 0 "); + return 0; + } + + String sign = generateSign(alert); + alert.setSign(sign); + int count = alertMapper.insertTaskResultAlertIfAbsent(alert); + if (count > 0) { + log.info("add task result alert to db , alert: {}", alert); + } else { + log.info("skip duplicate task result alert, sign: {}, workflowInstanceId: {}", sign, + alert.getWorkflowInstanceId()); + } + return count; + } + /** * update alert sending(execution) status * 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..df6a3a783556 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,34 @@ 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} + from t_ds_alert + where sign = #{alert.sign} + and workflow_instance_id = #{alert.workflowInstanceId} + and alert_type = #{alert.alertType.code} + having count(*) = 0 + +