ruanwenjun commented on code in PR #18722:
URL:
https://github.com/apache/dolphinscheduler/pull/18722#discussion_r4332613801
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/AbstractWorkflowStateAction.java:
##########
@@ -153,25 +154,34 @@ protected void
tryToTriggerSuccessorsAfterTaskFinish(final IWorkflowExecution wo
protected void workflowFinish(final IWorkflowExecution workflowExecution,
final WorkflowExecutionStatus
workflowExecutionStatus) {
- // todo: add transaction configuration in lifecycle event, all sync
lifecycle should be in transaction
+ final WorkflowInstance workflowInstance =
workflowExecution.getWorkflowInstance();
+ final WorkflowInstance workflowInstanceToUpdate = new
WorkflowInstance();
+ BeanUtils.copyProperties(workflowInstance, workflowInstanceToUpdate);
+ // Lifecycle events are dispatched without an enclosing transaction,
so execute commits before returning.
transactionTemplate.execute(status -> {
- final WorkflowInstance workflowInstance =
workflowExecution.getWorkflowInstance();
- workflowInstance.setEndTime(new Date());
- transformWorkflowInstanceState(workflowExecution,
workflowExecutionStatus);
- WorkflowInstanceMetrics.recordWorkflowInstanceFinish(
- workflowExecutionStatus,
- workflowInstance.getWorkflowDefinitionCode());
- if
(workflowExecution.getWorkflowExecuteContext().getWorkflowDefinition().getExecutionType()
- .isSerial()) {
- if
(serialCommandDao.deleteByWorkflowInstanceId(workflowInstance.getId()) > 0) {
- log.info("Success clear SerialCommand for
WorkflowExecuteRunnable: {}",
- workflowExecution.getName());
- }
+ workflowInstanceToUpdate.setState(workflowExecutionStatus);
+ workflowInstanceToUpdate.setEndTime(new Date());
+ workflowInstanceDao.updateById(workflowInstanceToUpdate);
+ if
(workflowExecution.getWorkflowExecuteContext().getWorkflowDefinition().getExecutionType().isSerial()
+ &&
serialCommandDao.deleteByWorkflowInstanceId(workflowInstance.getId()) > 0) {
+ log.info("Deleted SerialCommand within workflow finish
transaction for WorkflowExecuteRunnable: {}",
+ workflowExecution.getName());
}
- workflowExecution.getWorkflowEventBus()
-
.publish(WorkflowFinalizeLifecycleEvent.of(workflowExecution));
return null;
});
+ final WorkflowExecutionStatus originState =
workflowInstance.getState();
+ workflowInstance.setState(workflowInstanceToUpdate.getState());
+ workflowInstance.setEndTime(workflowInstanceToUpdate.getEndTime());
+ log.info("Success set WorkflowExecuteRunnable: {} state from: {} to
{}",
+ workflowInstance.getName(), originState.name(),
workflowInstance.getState().name());
+ try {
+
WorkflowInstanceMetrics.recordWorkflowInstanceFinish(workflowInstance.getState(),
+ workflowInstance.getWorkflowDefinitionCode());
+ } catch (Exception ex) {
+ // Metrics must not prevent releasing a workflow whose transaction
has already committed.
+ log.warn("Record completion metric failed for workflow: {}",
workflowExecution.getName(), ex);
+ }
Review Comment:
NIT: We can use another PR to simplify this. Since most of the state changes
here involve updating the database and updating memory, we can move the memory
updates into a separate method, similar to the implementation for task instance
updates.
At the same time, exception handling can be placed in
`recordWorkflowInstanceFinish`; I believe that any exceptions that occur while
recording metrics should not affect the upper layers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
