[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 42 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.
… TaskExecutionContext
…onstraint and atomic upsert
Thanks for the thorough review! Both issues have been addressed.
Thanks for the thorough review! Both issues have been addressed. |
SbloodyS
left a comment
SbloodyS
left a comment
There was a problem hiding this comment.
The previous ACK-ordering, task-identity, and database-constraint concerns have mostly been addressed, but the latest version still has the following issues.
1. The database-specific insert statements will not be selected
DaoConfiguration creates a plain VendorDatabaseIdProvider without defining any vendor mappings. Therefore, MyBatis uses the JDBC product names as database IDs: MySQL, PostgreSQL, and H2.
However, the new mapper statements use lowercase IDs:
databaseId="mysql"databaseId="postgresql"databaseId="h2"
These values are case-sensitive and do not match the IDs returned by the provider. As a result, MyBatis selects the generic plain INSERT statement instead. A duplicated lifecycle event will then violate uk_alert_dedup and throw an exception instead of returning zero, potentially preventing the Worker ACK and causing repeated retries.
Please either configure explicit mappings in VendorDatabaseIdProvider or use the actual database product names in the mapper statements.
The concurrent test currently hides this problem by catching and ignoring every exception. Future.isDone() only indicates completion; it does not report the exception. Please do not swallow these exceptions—call Future#get() and require every losing insert to return zero without throwing.
Also, consider avoiding MySQL INSERT IGNORE, because it can suppress unrelated data errors in addition to duplicate-key violations.
2. The protected AbstractTask fields are still removed
The deprecated getter and setter methods were restored, but the protected needAlert and taskAlertInfo fields are still removed.
Existing third-party task plugins may access these protected fields directly. Such plugins were compiled with field references to AbstractTask.needAlert and AbstractTask.taskAlertInfo, so they can still fail with NoSuchFieldError after upgrading.
Please retain the protected fields as deprecated compatibility members as well. Restoring only the methods does not preserve binary compatibility.
3. The null-alert-type fallback cannot protect an old Alert Server
The new fallback in AlertSender#getAlertData() cannot fix the documented rolling-upgrade scenario. An old Alert Server is running the old implementation, so it does not contain this null guard.
On a new Alert Server, TASK_RESULT is already known and will not be mapped to null. Therefore, this fallback only handles corrupted data or values introduced by an even newer version, and converting either case to WORKFLOW_INSTANCE_FAILURE silently sends an alert with the wrong semantic type.
Please rely on and enforce the documented component upgrade order, or preserve the unknown numeric value through deserialization. Do not silently relabel an unknown alert as a workflow failure.
…r into Fix-17854-2
…stractTask for binary compatibility
…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.
This approach follows the same pattern already used by insertAlertWhenServerCrash in the same mapper file, and avoids the following performance concerns with upsert syntax:
MySQL ON DUPLICATE KEY UPDATE: triggers a unique-index probe and may acquire gap/next-key locks even when the "update" is a no-op (id = id), adding lock contention under concurrent access.
H2 MERGE INTO: performs a separate SELECT then conditional INSERT, adding overhead on every call.
PostgreSQL ON CONFLICT: similarly incurs an extra index probe per insert.
The new SQL uses a single INSERT ... SELECT ... WHERE NOT EXISTS query. The NOT EXISTS subquery short-circuits on the first matching row via the uk_alert_dedup index, avoiding full-row scanning or locking overhead. This is one SQL statement, no dialect branches, and no upsert-specific locking.
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.
…TS for task-result alert idempotency
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