You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lo...@apache.org on 2022/05/26 22:57:43 UTC
[beam] branch master updated: [BEAM-11106] Validate that DoFn returns Process continuation when Truncating (#17770)
This is an automated email from the ASF dual-hosted git repository.
lostluck pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 075757d7369 [BEAM-11106] Validate that DoFn returns Process continuation when Truncating (#17770)
075757d7369 is described below
commit 075757d7369d2ebdb96b1d9f38fbe31b15a74e6d
Author: Ritesh Ghorse <ri...@gmail.com>
AuthorDate: Thu May 26 18:57:36 2022 -0400
[BEAM-11106] Validate that DoFn returns Process continuation when Truncating (#17770)
---
sdks/go/pkg/beam/core/graph/fn.go | 8 ++++++++
sdks/go/pkg/beam/core/graph/fn_test.go | 18 ++++++++++++++----
.../go/pkg/beam/core/runtime/exec/sdf_invokers_test.go | 10 +++++++---
3 files changed, 29 insertions(+), 7 deletions(-)
diff --git a/sdks/go/pkg/beam/core/graph/fn.go b/sdks/go/pkg/beam/core/graph/fn.go
index a37397977de..5eea802d5fa 100644
--- a/sdks/go/pkg/beam/core/graph/fn.go
+++ b/sdks/go/pkg/beam/core/graph/fn.go
@@ -1003,6 +1003,14 @@ func validateSdfSigTypes(fn *Fn, num int) error {
"Ensure that all restrictions in an SDF are the same type.",
truncateRestrictionName, 0, method.Ret[0].T, restrictionT, createInitialRestrictionName)
}
+ processFn := fn.methods[processElementName]
+ if _, exists := processFn.ProcessContinuation(); !exists {
+ err := errors.Errorf("missing return value in %v: return value of type %v is not present",
+ processElementName, reflect.TypeOf((*sdf.ProcessContinuation)(nil)).Elem())
+ return errors.SetTopLevelMsgf(err, "Missing output value in method %v, "+
+ "%v method should return %v when %v method is defined.",
+ processElementName, processElementName, reflect.TypeOf((*sdf.ProcessContinuation)(nil)).Elem(), truncateRestrictionName)
+ }
}
}
diff --git a/sdks/go/pkg/beam/core/graph/fn_test.go b/sdks/go/pkg/beam/core/graph/fn_test.go
index 9333db34734..0612f0ec4cb 100644
--- a/sdks/go/pkg/beam/core/graph/fn_test.go
+++ b/sdks/go/pkg/beam/core/graph/fn_test.go
@@ -23,6 +23,7 @@ import (
"testing"
"time"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
)
@@ -192,6 +193,7 @@ func TestNewDoFnSdf(t *testing.T) {
{dfn: &BadSdfReturnsRestSize{}},
{dfn: &BadSdfReturnsCreateTracker{}},
{dfn: &BadSdfReturnsTruncateRestriction{}},
+ {dfn: &BadSdfReturnsProcessElement{}},
// Validate element types consistent with ProcessElement.
{dfn: &BadSdfElementTCreateRest{}},
{dfn: &BadSdfElementTSplitRest{}},
@@ -760,8 +762,8 @@ func (fn *GoodSdf) CreateTracker(RestT) *RTrackerT {
return &RTrackerT{}
}
-func (fn *GoodSdf) ProcessElement(*RTrackerT, int) int {
- return 0
+func (fn *GoodSdf) ProcessElement(*RTrackerT, int) (int, sdf.ProcessContinuation) {
+ return 0, sdf.StopProcessing()
}
func (fn *GoodSdf) TruncateRestriction(*RTrackerT, int) RestT {
@@ -788,8 +790,8 @@ func (fn *GoodSdfKv) CreateTracker(RestT) *RTrackerT {
return &RTrackerT{}
}
-func (fn *GoodSdfKv) ProcessElement(*RTrackerT, int, int) int {
- return 0
+func (fn *GoodSdfKv) ProcessElement(*RTrackerT, int, int) (int, sdf.ProcessContinuation) {
+ return 0, sdf.StopProcessing()
}
func (fn *GoodSdfKv) TruncateRestriction(*RTrackerT, int, int) RestT {
@@ -981,6 +983,14 @@ func (fn *BadSdfReturnsTruncateRestriction) TruncateRestriction(*RTrackerT, int)
return RestT{}, 0
}
+type BadSdfReturnsProcessElement struct {
+ *GoodSdf
+}
+
+func (fn *BadSdfReturnsProcessElement) ProcessElement(*RTrackerT, int) int {
+ return 0
+}
+
// Examples with element types inconsistent with ProcessElement.
type BadSdfElementTCreateRest struct {
diff --git a/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers_test.go b/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers_test.go
index de016a327a8..ba33113bcb9 100644
--- a/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers_test.go
+++ b/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers_test.go
@@ -21,6 +21,7 @@ import (
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf"
"github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
"github.com/google/go-cmp/cmp"
)
@@ -554,12 +555,13 @@ func (fn *VetSdf) CreateWatermarkEstimator() *VetWatermarkEstimator {
// Note that emitting restrictions is discouraged in normal usage. It is only
// done here to allow validating that ProcessElement is being executed
// properly.
-func (fn *VetSdf) ProcessElement(rt *VetRTracker, i int, emit func(*VetRestriction)) {
+func (fn *VetSdf) ProcessElement(rt *VetRTracker, i int, emit func(*VetRestriction)) sdf.ProcessContinuation {
rest := rt.Rest
rest.Key = nil
rest.Val = i
rest.ProcessElm = true
emit(rest)
+ return sdf.ResumeProcessingIn(1 * time.Second)
}
type VetSdfStatefulWatermark struct {
@@ -669,12 +671,13 @@ func (fn *VetKvSdf) CreateTracker(rest *VetRestriction) *VetRTracker {
// Note that emitting restrictions is discouraged in normal usage. It is only
// done here to allow validating that ProcessElement is being executed
// properly.
-func (fn *VetKvSdf) ProcessElement(rt *VetRTracker, i, j int, emit func(*VetRestriction)) {
+func (fn *VetKvSdf) ProcessElement(rt *VetRTracker, i, j int, emit func(*VetRestriction)) sdf.ProcessContinuation {
rest := rt.Rest
rest.Key = i
rest.Val = j
rest.ProcessElm = true
emit(rest)
+ return sdf.ResumeProcessingIn(1 * time.Second)
}
// VetEmptyInitialSplitSdf runs an SDF in order to test that these methods get called properly,
@@ -722,10 +725,11 @@ func (fn *VetEmptyInitialSplitSdf) CreateTracker(rest *VetRestriction) *VetRTrac
// Note that emitting restrictions is discouraged in normal usage. It is only
// done here to allow validating that ProcessElement is being executed
// properly.
-func (fn *VetEmptyInitialSplitSdf) ProcessElement(rt *VetRTracker, i int, emit func(*VetRestriction)) {
+func (fn *VetEmptyInitialSplitSdf) ProcessElement(rt *VetRTracker, i int, emit func(*VetRestriction)) sdf.ProcessContinuation {
rest := rt.Rest
rest.Key = nil
rest.Val = i
rest.ProcessElm = true
emit(rest)
+ return sdf.ResumeProcessingIn(1 * time.Second)
}