uranusjr commented on code in PR #75443:
URL: https://github.com/apache/airflow/pull/75443#discussion_r4357388592
##########
go-sdk/internal/bundle/task.go:
##########
@@ -183,28 +183,31 @@ func (f *taskFunction) call(
)
}
}
+ // The task pushes no XCom when fn failed: its result is the zero
value, and pushing it would
+ // feed a value the task never produced to the downstream tasks that
read it.
+ if err != nil {
+ return err
+ }
if branch != nil {
// The task pushes the value that decide returns, and decide
runs only on a result that fn
- // returned without an error. So the task pushes no XCom when
fn or decide fails.
- if err != nil {
- return err
- }
+ // returned without an error.
Review Comment:
I can’t do inline suggestion but this comment is not really needed. Maybe
the first part is (up until the comma).
##########
go-sdk/internal/bundle/task_test.go:
##########
@@ -453,3 +453,53 @@ func (s *TaskSuite)
TestBranchFunctionFailsWhenItCannotSkip() {
})
}
}
+
+// runTask runs task as the task instance extract of dag1 with client.
+func runTask(task Task, client *branchClient) error {
+ ctx := context.WithValue(
+ context.Background(),
+ sdkcontext.SdkClientContextKey,
+ sdk.Client(client),
+ )
+ ctx = context.WithValue(ctx, sdkcontext.RuntimeContextKey,
sdk.NewTIRunContext(
+ context.Background(),
+ sdk.TaskInstance{DagID: "dag1", RunID: "run1", TaskID:
"extract"},
+ sdk.DagRun{DagID: "dag1", RunID: "run1"},
+ ))
+ return task.Execute(ctx, slog.New(logging.NewTeeLogger()), nil)
+}
+
+func (s *TaskSuite) TestPushesTheReturnValue() {
+ task, err := NewTaskFunction(func(contexttest.Context) (int, error) {
return 7, nil })
+ s.Require().NoError(err)
+
+ client := &branchClient{}
+ s.Require().NoError(runTask(task, client))
+
+ s.Equal([]string{"PushXCom extract return_value 7"}, client.calls)
+}
+
+// A failed push fails the task, so the downstream tasks do not fail later on
an XCom that is
+// not there.
Review Comment:
```suggestion
```
##########
go-sdk/internal/bundle/task.go:
##########
@@ -183,28 +183,31 @@ func (f *taskFunction) call(
)
}
}
+ // The task pushes no XCom when fn failed: its result is the zero
value, and pushing it would
+ // feed a value the task never produced to the downstream tasks that
read it.
Review Comment:
```suggestion
```
##########
go-sdk/internal/bundle/task_test.go:
##########
@@ -453,3 +453,53 @@ func (s *TaskSuite)
TestBranchFunctionFailsWhenItCannotSkip() {
})
}
}
+
+// runTask runs task as the task instance extract of dag1 with client.
+func runTask(task Task, client *branchClient) error {
+ ctx := context.WithValue(
+ context.Background(),
+ sdkcontext.SdkClientContextKey,
+ sdk.Client(client),
+ )
+ ctx = context.WithValue(ctx, sdkcontext.RuntimeContextKey,
sdk.NewTIRunContext(
+ context.Background(),
+ sdk.TaskInstance{DagID: "dag1", RunID: "run1", TaskID:
"extract"},
+ sdk.DagRun{DagID: "dag1", RunID: "run1"},
+ ))
+ return task.Execute(ctx, slog.New(logging.NewTeeLogger()), nil)
+}
+
+func (s *TaskSuite) TestPushesTheReturnValue() {
+ task, err := NewTaskFunction(func(contexttest.Context) (int, error) {
return 7, nil })
+ s.Require().NoError(err)
+
+ client := &branchClient{}
+ s.Require().NoError(runTask(task, client))
+
+ s.Equal([]string{"PushXCom extract return_value 7"}, client.calls)
+}
+
+// A failed push fails the task, so the downstream tasks do not fail later on
an XCom that is
+// not there.
+func (s *TaskSuite) TestFailsWhenItCannotPushTheReturnValue() {
+ task, err := NewTaskFunction(func(contexttest.Context) (int, error) {
return 7, nil })
+ s.Require().NoError(err)
+
+ client := &branchClient{failOn: "PushXCom extract return_value 7"}
+ s.EqualError(
+ runTask(task, client),
+ "setting the return value XCom: xcom refused",
+ )
+}
+
+// A task whose fn failed pushes nothing: its result is the zero value, not
one it produced.
Review Comment:
```suggestion
```
##########
go-sdk/internal/bundle/task.go:
##########
@@ -183,28 +183,31 @@ func (f *taskFunction) call(
)
}
}
+ // The task pushes no XCom when fn failed: its result is the zero
value, and pushing it would
+ // feed a value the task never produced to the downstream tasks that
read it.
+ if err != nil {
+ return err
+ }
if branch != nil {
// The task pushes the value that decide returns, and decide
runs only on a result that fn
- // returned without an error. So the task pushes no XCom when
fn or decide fails.
- if err != nil {
- return err
- }
+ // returned without an error.
value, skipped, err := f.decide(retValues[0].Interface())
if err != nil {
return err
}
rv := reflect.ValueOf(value)
if rv.Kind() != reflect.Ptr || !rv.IsNil() {
- f.sendXcom(ctx, value, sdkClient, logger)
+ if err := f.sendXcom(ctx, value, sdkClient); err != nil
{
+ return err
+ }
Review Comment:
This should be covered in a test.
##########
go-sdk/internal/bundle/task_test.go:
##########
@@ -453,3 +453,53 @@ func (s *TaskSuite)
TestBranchFunctionFailsWhenItCannotSkip() {
})
}
}
+
+// runTask runs task as the task instance extract of dag1 with client.
+func runTask(task Task, client *branchClient) error {
+ ctx := context.WithValue(
+ context.Background(),
+ sdkcontext.SdkClientContextKey,
+ sdk.Client(client),
+ )
+ ctx = context.WithValue(ctx, sdkcontext.RuntimeContextKey,
sdk.NewTIRunContext(
+ context.Background(),
+ sdk.TaskInstance{DagID: "dag1", RunID: "run1", TaskID:
"extract"},
+ sdk.DagRun{DagID: "dag1", RunID: "run1"},
+ ))
+ return task.Execute(ctx, slog.New(logging.NewTeeLogger()), nil)
+}
+
+func (s *TaskSuite) TestPushesTheReturnValue() {
+ task, err := NewTaskFunction(func(contexttest.Context) (int, error) {
return 7, nil })
+ s.Require().NoError(err)
+
+ client := &branchClient{}
+ s.Require().NoError(runTask(task, client))
+
+ s.Equal([]string{"PushXCom extract return_value 7"}, client.calls)
+}
+
+// A failed push fails the task, so the downstream tasks do not fail later on
an XCom that is
+// not there.
+func (s *TaskSuite) TestFailsWhenItCannotPushTheReturnValue() {
+ task, err := NewTaskFunction(func(contexttest.Context) (int, error) {
return 7, nil })
+ s.Require().NoError(err)
+
+ client := &branchClient{failOn: "PushXCom extract return_value 7"}
+ s.EqualError(
+ runTask(task, client),
+ "setting the return value XCom: xcom refused",
+ )
+}
+
+// A task whose fn failed pushes nothing: its result is the zero value, not
one it produced.
+func (s *TaskSuite) TestThatFailsPushesNoReturnValue() {
Review Comment:
```suggestion
func (s *TaskSuite) TestFailedTaskPushesNoReturnValue() {
```
##########
go-sdk/internal/bundle/task.go:
##########
@@ -268,16 +271,12 @@ func (b *branchRun) skipDownstream(
return nil
}
-func (f *taskFunction) sendXcom(
- ctx context.Context,
- value any,
- c sdk.XComClient,
- logger *slog.Logger,
-) {
+// sendXcom pushes the task's return value. A failure fails the task: leaving
it successful would
+// only move the error to the downstream tasks, which fail later on the XCom
that is not there.
+func (f *taskFunction) sendXcom(ctx context.Context, value any, c
sdk.XComClient) error {
runtimeContext, ok :=
ctx.Value(sdkcontext.RuntimeContextKey).(sdk.TIRunContext)
if !ok {
- logger.ErrorContext(ctx, "Unable to set XCom", "err", "task
runtime context is missing")
- return
+ return errors.New("setting the return value XCom: task runtime
context is missing")
Review Comment:
This runtime-context lookup duplicates `startBranch` and in binding. Not
blocking, but a small `taskInstanceFrom(ctx)` helper would remove three copies.
##########
go-sdk/internal/bundle/task.go:
##########
@@ -268,16 +271,12 @@ func (b *branchRun) skipDownstream(
return nil
}
-func (f *taskFunction) sendXcom(
- ctx context.Context,
- value any,
- c sdk.XComClient,
- logger *slog.Logger,
-) {
+// sendXcom pushes the task's return value. A failure fails the task: leaving
it successful would
+// only move the error to the downstream tasks, which fail later on the XCom
that is not there.
+func (f *taskFunction) sendXcom(ctx context.Context, value any, c
sdk.XComClient) error {
Review Comment:
Since `f` is never used, this can be a plain function.
While you're touching it, L281–287 rebuild an `sdk.TaskInstance` field by
field from `ti`. Maybe `ti` can just be passed in directly?
##########
go-sdk/internal/bundle/task_test.go:
##########
@@ -453,3 +453,53 @@ func (s *TaskSuite)
TestBranchFunctionFailsWhenItCannotSkip() {
})
}
}
+
+// runTask runs task as the task instance extract of dag1 with client.
+func runTask(task Task, client *branchClient) error {
Review Comment:
This copies `runBranch`. `branchClient` is now used by non-branch tests too,
so it should be renamed.
--
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]
