[Fix-17854] [Worker&SQL Task] Fix SQL task query result alert not being sent - #18549
[Fix-17854] [Worker&SQL Task] Fix SQL task query result alert not being sent#18549njnu-seafish wants to merge 51 commits into
Conversation
…r into Fix-17854-2
# Conflicts: # docs/docs/en/guide/upgrade/incompatible.md # docs/docs/zh/guide/upgrade/incompatible.md
SbloodyS
changed the title
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
Preserve the existing sendEmail field name
SqlParameters renames the persisted/API field from sendEmail to sendAlert, and the UI now only reads/writes sendAlert.
Although @JsonAlias("sendEmail") keeps deserialization compatible, serialization and UI payloads use the new name. This can cause existing clients, SDKs, integrations, or mixed-version components to silently lose the setting. It also introduces an unnecessary database migration and an incompatible public contract change.
Please keep sendEmail as the field name and only update the semantic description/UI label to indicate that alerts can use channels other than email. If the rename is still required, please provide an explicit compatibility strategy covering API clients, UI loading of legacy definitions, and rolling upgrades.
Keep AlertSendRequest wire-compatible
AlertSendRequest changes warnType: int to alertType: AlertType.
This changes both the field name and the serialized type of the Master–Alert RPC request. During a rolling upgrade, an old Alert Server will still expect warnType, while a new Alert Server may receive a request without alertType from an old Master. The result can be a default/incorrect alert type or a NullPointerException at alertType.getCode().
Please retain the existing warnType field for compatibility, or support both fields with explicit conversion and add a mixed-version serialization test.
…er in incompatible.md
thanks. doned |
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
TaskExecutionContext.getVarPool() is not populated with the predecessor-scoped VarPool
The latest change uses:
taskExecutionContext.getVarPool()
However, TaskExecutionContextFactory.createTaskExecutionContext() only writes the result of generateTaskInstanceVarPool() to:
taskInstance.setVarPool(VarPoolUtils.serializeVarPool(varPools));
TaskExecutionContextBuilder.buildTaskInstanceRelatedInfo() does not copy taskInstance.varPool into TaskExecutionContext, and TaskExecutionContext.varPool has no default value. Therefore, for a newly initialized sub-workflow logic task, taskExecutionContext.getVarPool() is normally null.
As a result, the current one-line change still drops runtime OUT parameters from upstream tasks. A manual test may appear to pass when the same parameter is also present in global parameters or the original workflow start parameters, but it does not verify propagation from the predecessor task's runtime output.
Please explicitly propagate the predecessor-scoped VarPool into the task execution context, or read the scoped VarPool from the current task instance. Also add automated regression tests covering:
- An OUT parameter produced only at runtime by an upstream task is passed to the sub-workflow.
- An OUT parameter from an unrelated sibling branch is not passed to the sub-workflow.
- Conflicting global/start/upstream parameters retain the intended precedence.
Thanks @SbloodyS. I'd like to clarify the scope here, because I believe this concern is out of scope for this PR. .varPool(taskExecutionContext.getVarPool()) is pre-existing code on dev. In the diff of this PR it appears as an unchanged context line; this PR only adds the two adjacent lines .needAlert(...) and .taskAlertInfo(...): .varPool(taskExecutionContext.getVarPool()) // unchanged (pre-existing on dev)
This PR does not modify TaskExecutionContextFactory or TaskExecutionContextBuilder. The alert path is decoupled from varPool. The SQL task-result alert travels through needAlert / taskAlertInfo, an independent channel that does not read taskExecutionContext.getVarPool(). SqlTask#prepareTaskResultAlert() only sets the alert info; it never reads varPool. So this PR neither affects nor depends on varPool propagation. The varPool-propagation issue on sub-workflow logical tasks looks like a real pre-existing bug on dev, and I agree it's worth fixing — but it touches the shared TaskExecutionContext build path that all task types go through, so it belongs in a separate issue/PR with its own regression tests, rather than being mixed into an alert-focused PR (close #17854). Could you double-check whether this comment was intended for another PR that actually touches the varPool/sub-workflow path? Could you double-check whether this comment was intended for another PR that actually touches the varPool/sub-workflow path? Happy to file a separate issue to track the varPool propagation fix if you confirm. |
| <insert id="insertTaskResultAlertIfAbsent" databaseId="mysql"> | ||
| INSERT INTO t_ds_alert(sign, title, content, alert_status, warning_type, log, alertgroup_id, | ||
| create_time, update_time, project_code, workflow_definition_code, | ||
| workflow_instance_id, alert_type) | ||
| VALUES (#{alert.sign}, #{alert.title}, #{alert.content}, #{alert.alertStatus.code}, | ||
| #{alert.warningType.code}, #{alert.log}, #{alert.alertGroupId}, #{alert.createTime}, | ||
| #{alert.updateTime}, #{alert.projectCode}, #{alert.workflowDefinitionCode}, | ||
| #{alert.workflowInstanceId}, #{alert.alertType.code}) | ||
| ON DUPLICATE KEY UPDATE id = id | ||
| </insert> | ||
|
|
||
| <!-- H2 --> | ||
| <insert id="insertTaskResultAlertIfAbsent" databaseId="h2"> | ||
| MERGE 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) | ||
| KEY(sign, workflow_instance_id, alert_type) | ||
| VALUES (#{alert.sign}, #{alert.title}, #{alert.content}, #{alert.alertStatus.code}, | ||
| #{alert.warningType.code}, #{alert.log}, #{alert.alertGroupId}, #{alert.createTime}, | ||
| #{alert.updateTime}, #{alert.projectCode}, #{alert.workflowDefinitionCode}, | ||
| #{alert.workflowInstanceId}, #{alert.alertType.code}) | ||
| </insert> | ||
|
|
||
| <!-- PostgreSQL --> | ||
| <insert id="insertTaskResultAlertIfAbsent" databaseId="postgresql"> | ||
| INSERT INTO t_ds_alert(sign, title, content, alert_status, warning_type, log, alertgroup_id, | ||
| create_time, update_time, project_code, workflow_definition_code, | ||
| workflow_instance_id, alert_type) | ||
| VALUES (#{alert.sign}, #{alert.title}, #{alert.content}, #{alert.alertStatus.code}, | ||
| #{alert.warningType.code}, #{alert.log}, #{alert.alertGroupId}, #{alert.createTime}, | ||
| #{alert.updateTime}, #{alert.projectCode}, #{alert.workflowDefinitionCode}, | ||
| #{alert.workflowInstanceId}, #{alert.alertType.code}) | ||
| ON CONFLICT (sign, workflow_instance_id, alert_type) DO NOTHING | ||
| </insert> |
There was a problem hiding this comment.
Don't use this writing method, which will lead to poor database performance.
There was a problem hiding this comment.
Don't use this writing method, which will lead to poor database performance.
Replaced the dialect-specific upsert (ON DUPLICATE KEY UPDATE / MERGE INTO / ON CONFLICT DO NOTHING) with a single dialect-neutral INSERT ... SELECT ... WHERE NOT EXISTS statement.
The uk_alert_dedup unique constraint is retained as a concurrent-safety net. If a race condition causes two threads to pass NOT EXISTS simultaneously, the constraint catches the duplicate insert.
The DAO layer catches DuplicateKeyException and returns 0, ensuring idempotency without propagating exceptions to the caller.
String sign = generateSign(alert);
alert.setSign(sign);
try {
int count = alertMapper.insertTaskResultAlertIfAbsent(alert);
if (count > 0) {
log.info("add task result alert to db , alert: {}", alert);
} else {
log.info("skip duplicate task result alert, sign: {}, workflowInstanceId: {}", sign,
alert.getWorkflowInstanceId());
}
return count;
} catch (DuplicateKeyException e) {
// Concurrent race: NOT EXISTS passed but another thread inserted first.
// The uk_alert_dedup unique constraint caught it — treat as a skip.
log.info("skip duplicate task result alert (concurrent race), sign: {}, workflowInstanceId: {}", sign,
alert.getWorkflowInstanceId());
return 0;
}
fully implements a dual-safeguard idempotent write mechanism combining application-layer deduplication with a database unique constraint as a fallback.
There was a problem hiding this comment.
Don't use this writing method, which will lead to poor database performance.
- Unique Index Added
A unique composite index uk_alert_dedup has been created on the columns (sign, workflow_instance_id, alert_type) in the t_ds_alert table. This serves as the ultimate safeguard at the database level, ensuring that no duplicate alert records can ever be persisted, regardless of what happens at the application or SQL layer.
ALTER TABLE
t_ds_alertADD UNIQUE INDEXuk_alert_dedup(sign,workflow_instance_id,alert_type);
- Optimistic Check in SQL — Low Cost, No Exception Overhead
The INSERT statement uses a WHERE NOT EXISTS clause to perform an optimistic duplicate check before inserting. In the vast majority of cases, if the alert already exists, the subquery detects it via the unique index and simply returns zero affected rows — no exception is thrown, no lock contention is incurred, and no error-handling overhead is introduced. This makes the common path both fast and lightweight.
INSERT INTO t_ds_alert(sign, title, content, alert_status, warning_type, log, alertgroup_id,
create_time, update_time, project_code, workflow_definition_code,
workflow_instance_id, alert_type)
SELECT #{alert.sign}, #{alert.title}, #{alert.content}, #{alert.alertStatus.code},
#{alert.warningType.code}, #{alert.log}, #{alert.alertGroupId}, #{alert.createTime},
#{alert.updateTime}, #{alert.projectCode}, #{alert.workflowDefinitionCode},
#{alert.workflowInstanceId}, #{alert.alertType.code}
WHERE NOT EXISTS (
SELECT 1 FROM t_ds_alert
WHERE sign = #{alert.sign}
AND workflow_instance_id = #{alert.workflowInstanceId}
AND alert_type = #{alert.alertType.code}
)
</insert>
- Pessimistic Fallback via DuplicateKeyException — Ensuring Consistency Under Extreme Concurrency
In rare race conditions where two or more threads simultaneously pass the NOT EXISTS check and attempt to insert the same alert, the unique index will reject one of them, causing a DuplicateKeyException. The application catches this exception as a pessimistic fallback, treating it as a skip rather than a failure. This guarantees data consistency even under extreme concurrent load.
try {
int count = alertMapper.insertTaskResultAlertIfAbsent(alert);
if (count > 0) {
log.info("add task result alert to db , alert: {}", alert);
} else {
log.info("skip duplicate task result alert, sign: {}, workflowInstanceId: {}", sign,
alert.getWorkflowInstanceId());
}
return count;
} catch (DuplicateKeyException e) {
// Concurrent race: NOT EXISTS passed but another thread inserted first.
// The uk_alert_dedup unique constraint caught it — treat as a skip.
log.info("skip duplicate task result alert (concurrent race), sign: {}, workflowInstanceId: {}", sign,
alert.getWorkflowInstanceId());
return 0;
}
I clean up redundant comments and code.
@SbloodyS When you have a moment, could you please review the code again? Thanks so much!
…TS for task-result alert idempotency
|
@SbloodyS When you have a moment, could you please review the code again? Thanks so much! |
| <insert id="insertTaskResultAlertIfAbsent"> | ||
| INSERT INTO t_ds_alert(sign, title, content, alert_status, warning_type, log, alertgroup_id, | ||
| create_time, update_time, project_code, workflow_definition_code, | ||
| workflow_instance_id, alert_type) | ||
| SELECT #{alert.sign}, #{alert.title}, #{alert.content}, #{alert.alertStatus.code}, | ||
| #{alert.warningType.code}, #{alert.log}, #{alert.alertGroupId}, #{alert.createTime}, | ||
| #{alert.updateTime}, #{alert.projectCode}, #{alert.workflowDefinitionCode}, | ||
| #{alert.workflowInstanceId}, #{alert.alertType.code} | ||
| WHERE NOT EXISTS ( | ||
| SELECT 1 FROM t_ds_alert | ||
| WHERE sign = #{alert.sign} | ||
| AND workflow_instance_id = #{alert.workflowInstanceId} | ||
| AND alert_type = #{alert.alertType.code} | ||
| ) | ||
| </insert> |
There was a problem hiding this comment.
Such logic should not be handled in the database, but should be handled by code.
There was a problem hiding this comment.
Such logic should not be handled in the database, but should be handled by code.
Thanks for the feedback. I've moved the dedup logic out of SQL and into application code.
The INSERT ... SELECT ... WHERE NOT EXISTS statement has been replaced with a plain INSERT ... VALUES. Idempotency is now handled entirely in code: AlertDao#addTaskResultAlert performs a direct insert and catches DuplicateKeyException (thrown by the uk_alert_dedup unique constraint) to treat the duplicate as a skip. The unique constraint remains as the DB-level safety net, but no dedup logic lives in the SQL itself.
2 participants
Was this PR generated or assisted by AI?
Yes, I design the architecture and write the core code myself, then use an LLM to review and optimize the logic.
Purpose of the pull request
close #17854
Brief change log
Purpose
Fix #17854: the SQL task "query result" alert feature silently stopped working after the
task-executor refactoring (DSIP-73). The alert flag and payload (
needAlert/taskAlertInfo) used to live onAbstractTask, but no component consumed them anymore,so enabling "Send Alert" on a SQL task had no effect.
Root cause
The
needAlert/taskAlertInfofields were only defined and set in the task plugin(
AbstractTask), while the Master never read them. After the task-executor modulerefactor, the success lifecycle event did not carry the alert information to the Master,
so the alert was never persisted/sent.
What changed
Task plugin side
needAlert/taskAlertInfofromAbstractTaskintoTaskExecutionContextso they can be carried across the Worker -> Master RPC.
SqlTask: prepare the alert info (title,alertGroupId,AlertType.TASK_RESULT)and truncate the query result to
displayRows(default if unset) to avoid oversizedRPC payloads; empty result sets are also covered.
sendEmailtosendAlert(kept@JsonAlias("sendEmail")for backward-compatible deserialization) and removed the obsolete
showTypefield.Event / Master
TaskExecutorSuccessLifecycleEventnow carriesneedAlertandtaskAlertInfo.TaskExecutorEventListenerImplconsumes the success event: whenneedAlertis trueand a valid
alertGroupIdis present, it delegates toWorkflowAlertManager.sendTaskResultAlert(with project / workflow / task context filled in); otherwise it logs a warning instead
of silently dropping the alert.
Alert chain
AlertType.TASK_RESULT (8).AlertSendRequestnow carriesAlertTypeinstead of a plain intwarnType;AlertSender.syncHandlerandAlertOperatorImplpropagate it intoAlertData.Data migration & docs
sendEmail->sendAlertint_ds_task_definitionandt_ds_task_definition_log(null-safe guards added).incompatible.md(en/zh).Verification
SqlParametersTest: JSON backward compatibility (sendEmail->sendAlert) andnew field name.
AlertSenderTest:syncHandlerwith the newAlertTypeargument.mvn compile).I previously submitted a PR proposing that the Worker role should directly send RPC requests to the Master to transmit SQL result set alerts. The proposal was rejected. (#17856)
Verify this pull request
This pull request is code cleanup without any test coverage.
(or)
This pull request is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(or)
Pull Request Notice
Pull Request Notice
If your pull request contains incompatible change, you should also add it to
docs/docs/en/guide/upgrade/incompatible.md