Copilot commented on code in PR #17953:
URL:
https://github.com/apache/dolphinscheduler/pull/17953#discussion_r2767484694
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java:
##########
@@ -49,12 +48,63 @@ public class WorkflowExecuteContext implements
IWorkflowExecuteContext {
private final IWorkflowGraph workflowGraph;
- private final IWorkflowExecutionGraph workflowExecutionGraph;
+ private volatile IWorkflowExecutionGraph workflowExecutionGraph;
+
+ private final IWorkflowExecutionGraphAssembler
workflowExecutionGraphAssembler;
private final WorkflowEventBus workflowEventBus;
private final List<IWorkflowLifecycleListener>
workflowInstanceLifecycleListeners;
+ public WorkflowExecuteContext(Command command,
+ WorkflowDefinition workflowDefinition,
+ Project project,
+ WorkflowInstance workflowInstance,
+ IWorkflowGraph workflowGraph,
+ IWorkflowExecutionGraph
workflowExecutionGraph,
+ IWorkflowExecutionGraphAssembler
workflowExecutionGraphAssembler,
+ WorkflowEventBus workflowEventBus,
+ List<IWorkflowLifecycleListener>
workflowInstanceLifecycleListeners) {
+ this.command = command;
+ this.workflowDefinition = workflowDefinition;
+ this.project = project;
+ this.workflowInstance = workflowInstance;
+ this.workflowGraph = workflowGraph;
+ this.workflowExecutionGraph = workflowExecutionGraph;
+ this.workflowExecutionGraphAssembler = workflowExecutionGraphAssembler;
+ this.workflowEventBus = workflowEventBus;
+ this.workflowInstanceLifecycleListeners =
workflowInstanceLifecycleListeners;
+ }
+
+ /**
+ * Initialize the workflow execution graph using the assembler.
+ * This method should be called when the workflow is ready to start
execution,
+ * typically during the handling of WorkflowStartLifecycleEvent.
+ *
+ * @throws IllegalStateException if the execution graph is already
initialized
+ * or no assembler is available
+ */
+ public void initializeWorkflowExecutionGraph() {
+ if (workflowExecutionGraph != null) {
+ return;
+ }
+ if (workflowExecutionGraphAssembler == null) {
+ return;
+ }
+ synchronized (this) {
+ if (workflowExecutionGraph == null) {
+ workflowExecutionGraph =
workflowExecutionGraphAssembler.assemble();
+ }
+ }
+ }
+
+ /**
+ * Check if the workflow execution graph has been initialized.
+ */
Review Comment:
This method overrides
[IWorkflowExecuteContext.isWorkflowExecutionGraphInitialized](1); it is
advisable to add an Override annotation.
```suggestion
*/
@Override
```
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java:
##########
@@ -49,12 +48,63 @@ public class WorkflowExecuteContext implements
IWorkflowExecuteContext {
private final IWorkflowGraph workflowGraph;
- private final IWorkflowExecutionGraph workflowExecutionGraph;
+ private volatile IWorkflowExecutionGraph workflowExecutionGraph;
Review Comment:
This method overrides
[IWorkflowExecuteContext.getWorkflowExecutionGraph](1); it is advisable to add
an Override annotation.
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java:
##########
@@ -49,12 +48,63 @@ public class WorkflowExecuteContext implements
IWorkflowExecuteContext {
private final IWorkflowGraph workflowGraph;
- private final IWorkflowExecutionGraph workflowExecutionGraph;
+ private volatile IWorkflowExecutionGraph workflowExecutionGraph;
+
+ private final IWorkflowExecutionGraphAssembler
workflowExecutionGraphAssembler;
private final WorkflowEventBus workflowEventBus;
private final List<IWorkflowLifecycleListener>
workflowInstanceLifecycleListeners;
+ public WorkflowExecuteContext(Command command,
+ WorkflowDefinition workflowDefinition,
+ Project project,
+ WorkflowInstance workflowInstance,
+ IWorkflowGraph workflowGraph,
+ IWorkflowExecutionGraph
workflowExecutionGraph,
+ IWorkflowExecutionGraphAssembler
workflowExecutionGraphAssembler,
+ WorkflowEventBus workflowEventBus,
+ List<IWorkflowLifecycleListener>
workflowInstanceLifecycleListeners) {
+ this.command = command;
+ this.workflowDefinition = workflowDefinition;
+ this.project = project;
+ this.workflowInstance = workflowInstance;
+ this.workflowGraph = workflowGraph;
+ this.workflowExecutionGraph = workflowExecutionGraph;
+ this.workflowExecutionGraphAssembler = workflowExecutionGraphAssembler;
+ this.workflowEventBus = workflowEventBus;
+ this.workflowInstanceLifecycleListeners =
workflowInstanceLifecycleListeners;
+ }
+
+ /**
+ * Initialize the workflow execution graph using the assembler.
+ * This method should be called when the workflow is ready to start
execution,
+ * typically during the handling of WorkflowStartLifecycleEvent.
+ *
+ * @throws IllegalStateException if the execution graph is already
initialized
+ * or no assembler is available
Review Comment:
The Javadoc states that this method throws IllegalStateException if the
execution graph is already initialized or no assembler is available. However,
the implementation simply returns early in both cases without throwing an
exception. The Javadoc should be updated to accurately reflect the method's
behavior, or the method should be changed to throw exceptions as documented.
```suggestion
* <p>
* If the execution graph is already initialized or no assembler is
available,
* this method returns without making any changes.
```
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteContext.java:
##########
@@ -49,12 +48,63 @@ public class WorkflowExecuteContext implements
IWorkflowExecuteContext {
private final IWorkflowGraph workflowGraph;
- private final IWorkflowExecutionGraph workflowExecutionGraph;
+ private volatile IWorkflowExecutionGraph workflowExecutionGraph;
+
+ private final IWorkflowExecutionGraphAssembler
workflowExecutionGraphAssembler;
private final WorkflowEventBus workflowEventBus;
private final List<IWorkflowLifecycleListener>
workflowInstanceLifecycleListeners;
+ public WorkflowExecuteContext(Command command,
+ WorkflowDefinition workflowDefinition,
+ Project project,
+ WorkflowInstance workflowInstance,
+ IWorkflowGraph workflowGraph,
+ IWorkflowExecutionGraph
workflowExecutionGraph,
+ IWorkflowExecutionGraphAssembler
workflowExecutionGraphAssembler,
+ WorkflowEventBus workflowEventBus,
+ List<IWorkflowLifecycleListener>
workflowInstanceLifecycleListeners) {
+ this.command = command;
+ this.workflowDefinition = workflowDefinition;
+ this.project = project;
+ this.workflowInstance = workflowInstance;
+ this.workflowGraph = workflowGraph;
+ this.workflowExecutionGraph = workflowExecutionGraph;
+ this.workflowExecutionGraphAssembler = workflowExecutionGraphAssembler;
+ this.workflowEventBus = workflowEventBus;
+ this.workflowInstanceLifecycleListeners =
workflowInstanceLifecycleListeners;
+ }
+
+ /**
+ * Initialize the workflow execution graph using the assembler.
+ * This method should be called when the workflow is ready to start
execution,
+ * typically during the handling of WorkflowStartLifecycleEvent.
+ *
+ * @throws IllegalStateException if the execution graph is already
initialized
+ * or no assembler is available
+ */
Review Comment:
This method overrides
[IWorkflowExecuteContext.initializeWorkflowExecutionGraph](1); it is advisable
to add an Override annotation.
```suggestion
*/
@Override
```
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/WorkflowRunningStateAction.java:
##########
@@ -46,8 +46,18 @@ public class WorkflowRunningStateAction extends
AbstractWorkflowStateAction {
public void onStartEvent(final IWorkflowExecutionRunnable
workflowExecutionRunnable,
final WorkflowStartLifecycleEvent
workflowStartEvent) {
throwExceptionIfStateIsNotMatch(workflowExecutionRunnable);
+
+ // Initialize the workflow execution graph if not already initialized
+ // This is deferred from command handling to reduce transaction time
+
workflowExecutionRunnable.getWorkflowExecuteContext().initializeWorkflowExecutionGraph();
Review Comment:
If `initializeWorkflowExecutionGraph()` throws an exception during graph
assembly (e.g., from the assembler's `assemble()` method), the exception will
propagate up and be caught by the event bus fire worker's catch block. However,
this will only log the error without properly marking the workflow as failed.
The PR description mentions that "Workflow can be properly marked as failed
with WorkflowFailedLifecycleEvent", but there's no try-catch block around the
initialization call to publish a WorkflowFailedLifecycleEvent on failure.
Consider wrapping the initialization in a try-catch block and publishing
WorkflowFailedLifecycleEvent on exception to ensure proper workflow failure
handling.
--
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]
