[
https://issues.apache.org/jira/browse/BEAM-10976?focusedWorklogId=735976&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-735976
]
ASF GitHub Bot logged work on BEAM-10976:
-----------------------------------------
Author: ASF GitHub Bot
Created on: 03/Mar/22 14:05
Start Date: 03/Mar/22 14:05
Worklog Time Spent: 10m
Work Description: 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]
Issue Time Tracking
-------------------
Worklog Id: (was: 735976)
Time Spent: 3.5h (was: 3h 20m)
> Enable Bundle Finalization in Go SDK
> ------------------------------------
>
> Key: BEAM-10976
> URL: https://issues.apache.org/jira/browse/BEAM-10976
> Project: Beam
> Issue Type: Sub-task
> Components: sdk-go
> Reporter: Robert Burke
> Assignee: Danny McCormick
> Priority: P3
> Time Spent: 3.5h
> Remaining Estimate: 0h
>
> Eg. to support acking pubsub/kafka messages as processed after the results
> have been properly committed by the runner.
> Note, that due to BEAM-10959 that when implementing this, an instruction must
> remain "active" until it's finalization occurs as well. Specifically, we
> should probably keep another map around for "to be finalized" process bundle
> instructions so we can return the appropriate "empty" response and not
> accidently evict them from the nearly equivalent inactive state until after
> finalization.
> [https://s.apache.org/beam-finalizing-bundles]
> Â
> (To be updated once [https://github.com/apache/beam/pull/13160]Â is merged and
> the programming guide updated with SDF content.)
> See also Java and Python approaches
> https://beam.apache.org/documentation/programming-guide/#bundle-finalization
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
