damccorm commented on a change in pull request #16980:
URL: https://github.com/apache/beam/pull/16980#discussion_r818687345
##########
File path: sdks/go/pkg/beam/core/runtime/exec/fn.go
##########
@@ -39,23 +40,47 @@ type MainInput struct {
RTracker sdf.RTracker
}
+type bundleFinalizationCallback struct {
+ callback func() error
+ validUntil time.Time
+}
+
+// bundleFinalizer holds all the user defined callbacks to be run on bundle
finalization.
+// Implements typex.BundleFinalization
+type bundleFinalizer struct {
+ callbacks []bundleFinalizationCallback
+ lastValidCallback time.Time // Used to track when we can safely gc the
bundleFinalizer
+}
+
+// RegisterCallback is used to register callbacks during DoFn execution.
+func (bf *bundleFinalizer) RegisterCallback(t time.Duration, cb func() error) {
+ callback := bundleFinalizationCallback{
+ callback: cb,
+ validUntil: time.Now().Add(t),
+ }
+ bf.callbacks = append(bf.callbacks, callback)
+ if bf.lastValidCallback.Before(callback.validUntil) {
+ bf.lastValidCallback = callback.validUntil
+ }
+}
+
// Invoke invokes the fn with the given values. The extra values must match
the non-main
// side input and emitters. It returns the direct output, if any.
-func Invoke(ctx context.Context, pn typex.PaneInfo, ws []typex.Window, ts
typex.EventTime, fn *funcx.Fn, opt *MainInput, extra ...interface{})
(*FullValue, error) {
+func Invoke(ctx context.Context, pn typex.PaneInfo, ws []typex.Window, ts
typex.EventTime, fn *funcx.Fn, opt *MainInput, bf *bundleFinalizer, extra
...interface{}) (*FullValue, error) {
Review comment:
I would argue against it in this instance. Functionally, the
bundleFinalizer is much closer to being an emitter than it is an input - its a
tool that is used to produce an output (callbacks) and thus I'd be pretty
surprised to find it in the MainInput. FWIW, I'm also kinda surprised to find
RTracker in MainInput, though I think there's a significantly stronger case
that it should be included since it does modify how we're able to access the
input (if I understand its purpose correctly 🤷)
##########
File path: sdks/go/pkg/beam/core/runtime/exec/fn.go
##########
@@ -99,6 +124,11 @@ func newInvoker(fn *funcx.Fn) *invoker {
if n.outErrIdx, ok = fn.Error(); !ok {
n.outErrIdx = -1
}
+ // TODO(@damccorm) - add this back in once BundleFinalization is
implemented
Review comment:
Done - that makes sense, and I'm much happier with that pattern 😃
--
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]
