Address comment

Signed-off-by: Kevin Su <[email protected]>
flyteorg · Dec 4, 2023 · b2ab9aa · b2ab9aa
1 parent d6b4b1d
commit b2ab9aa
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 15 deletions.
diff --git a/flytepropeller/pkg/compiler/workflow_compiler.go b/flytepropeller/pkg/compiler/workflow_compiler.go
@@ -227,7 +227,7 @@ func (w workflowBuilder) ValidateWorkflow(fg *flyteWorkflow, errs errors.Compile
 
 	if fg.Template.FailureNode != nil {
 		failureNode := fg.Template.FailureNode
-		v.ValidateNode(&wf, wf.GetOrCreateNodeBuilder(failureNode), false /* validateConditionTypes */, errs.NewScope())
+		v.ValidateNode(&wf, wf.GetOrCreateNodeBuilder(failureNode), false, errs.NewScope())
 		wf.AddEdges(wf.GetOrCreateNodeBuilder(failureNode), c.EdgeDirectionUpstream, errs.NewScope())
 	}
 

diff --git a/flytepropeller/pkg/compiler/workflow_compiler_test.go b/flytepropeller/pkg/compiler/workflow_compiler_test.go
@@ -148,14 +148,8 @@ func TestCompileWorkflowWithFailureNode(t *testing.T) {
 	// Detect what other workflows/tasks does this coreWorkflow reference
 	subWorkflows := make([]*core.WorkflowTemplate, 0)
 	reqs, err := GetRequirements(inputWorkflow, subWorkflows)
-	if err != nil {
-		fmt.Printf("failed to get requirements. Error: %v", err)
-		return
-	}
-
-	fmt.Printf("Needed Tasks: [%v], Needed Workflows [%v]\n",
-		strings.Join(dumpIdentifierNames(reqs.GetRequiredTaskIds()), ","),
-		strings.Join(dumpIdentifierNames(reqs.GetRequiredLaunchPlanIds()), ","))
+	assert.Nil(t, err)
+	assert.Equal(t, reqs.taskIds, []common.Identifier{{Name: "cleanup"}, {Name: "task_123"}})
 
 	// Replace with logic to satisfy the requirements
 	workflows := make([]common.InterfaceProvider, 0)
@@ -193,10 +187,7 @@ func TestCompileWorkflowWithFailureNode(t *testing.T) {
 	compiledTasks := make([]*core.CompiledTask, 0, len(tasks))
 	for _, task := range tasks {
 		compiledTask, err := CompileTask(task)
-		if err != nil {
-			fmt.Printf("failed to compile task [%v]. Error: %v", task.Id, err)
-			return
-		}
+		assert.Nil(t, err)
 
 		compiledTasks = append(compiledTasks, compiledTask)
 	}

diff --git a/flytepropeller/pkg/controller/nodes/subworkflow/subworkflow.go b/flytepropeller/pkg/controller/nodes/subworkflow/subworkflow.go
@@ -142,8 +142,7 @@ func (s *subworkflowHandler) getExecutionContextForDownstream(nCtx interfaces.No
 
 func (s *subworkflowHandler) HandleFailureNodeOfSubWorkflow(ctx context.Context, nCtx interfaces.NodeExecutionContext, subworkflow v1alpha1.ExecutableSubWorkflow, nl executors.NodeLookup) (handler.Transition, error) {
 	originalError := nCtx.NodeStateReader().GetWorkflowNodeState().Error
-	failureNode := subworkflow.GetOnFailureNode()
-	if failureNode != nil {
+	if failureNode := subworkflow.GetOnFailureNode(); failureNode != nil {
 		execContext, err := s.getExecutionContextForDownstream(nCtx)
 		if err != nil {
 			return handler.UnknownTransition, err