[BEAM-10976] Bundle finalization: Harness and some exec changes

[damccorm]

This is part 1 of 2 to add bundle finalization support to the Go Sdk.

Summary of Overall Changes

Bundle finalization enables a DoFn to perform side effects after a runner has acknowledged that it has durably persisted the output. Right now, Java and Python support bundle finalization by allowing a user to register a callback function which is invoked when the runner acknowledges that it has persisted all output, but Go does not have any such support. This is part of a larger change to add support to the Go Sdk as outlined in this design doc.

I've completed all the changes (sans some better testing on the parts not in this PR), you can see the remaining files in the diff here - https://github.com/apache/beam/compare/master...damccorm:users/damccorm/bundle-finalization?expand=1

Summary of Changes in this PR

This PR adds most of the non user facing changes needed to enable this change. There are basically 2 major components:

1. Changes to exec/pardo.go and exec/fn.go to add the bundleFinalizer type and pass it into the user's DoFn when appropriate.
2. Changes to harness.go and plan.go to manage plans that require finalization, respond to the runner sending the bundle finalization message, and to invoke the callbacks.

Additional testing done

On top of the units added, using my full implementation (not just this partial one) I also was able to run an E2E example on Dataflow (FWIW, not all runners have finalization support but I found that Dataflow does). In that example, I hijacked the wordcount example and added a bundleFinalizer to write a file to persistent storage for each line that had at least 3 words (chosen pretty randomly to minimize the chances of collisions). I'll omit the whole sample since its long, but it produced a bunch of files like this:

This indeed ran after the other data was persisted

Next Steps

After this, I'll add the user facing functionality in a follow up pr, along with testing for that and 1+ integration test for the whole flow.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[codecov]

Codecov Report

Merging #16980 (36e1450) into master (daabeba) will increase coverage by 0.68%.
The diff coverage is 22.13%.

@@            Coverage Diff             @@
##           master   #16980      +/-   ##
==========================================
+ Coverage   73.05%   73.74%   +0.68%     
==========================================
  Files         660      667       +7     
  Lines       86785    87356     +571     
==========================================
+ Hits        63402    64420    +1018     
+ Misses      22384    21829     -555     
- Partials      999     1107     +108     

Flag	Coverage Δ
go	48.93% <22.13%> (+3.03%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
sdks/go/pkg/beam/core/runtime/exec/util.go	74.28% <ø> (ø)
sdks/go/pkg/beam/core/runtime/harness/harness.go	11.26% <0.00%> (-1.43%)	⬇️
sdks/go/pkg/beam/runners/direct/buffer.go	77.46% <ø> (+77.46%)	⬆️
sdks/go/pkg/beam/core/runtime/exec/plan.go	48.61% <22.22%> (-8.80%)	⬇️
sdks/go/pkg/beam/core/runtime/exec/pardo.go	50.45% <25.00%> (ø)
sdks/go/pkg/beam/core/runtime/exec/combine.go	60.43% <66.66%> (ø)
sdks/go/pkg/beam/core/runtime/exec/fn.go	69.26% <80.00%> (+0.56%)	⬆️
sdks/go/pkg/beam/core/graph/window/fn.go	53.84% <0.00%> (ø)
sdks/go/pkg/beam/x/debug/debug.shims.go	73.78% <0.00%> (ø)
sdks/go/pkg/beam/x/debug/head.go	100.00% <0.00%> (ø)
... and 20 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update daabeba...36e1450. Read the comment docs.

[damccorm]

R: @lostluck

[lostluck]

Is there any reason the callbacks have to live in each ParDo, rather than an independant structure that's referenced in the ParDos at Plan construction time, that can be swapped out/cleared when the plan is re-used/garbage collected?

That would avoid the plumbing creep where every implementation of Unit needs new methods just to get down to the ParDos that may have callbacks.

You'll note that in Go, that it's not a great idea to simply add new methods to an interface, as it will break everything that's trying to implement the interface.

I'll also point out that we have a pattern that does this, like how we collect metrics from PCollections, by pre-collecting all the PCollections in a plan rather than plumbing a new path all the way through: See
https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/runtime/exec/plan.go#L59
https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/runtime/exec/translate.go#L82

[damccorm]

I definitely agree that this is a better approach and will update accordingly. With that said:

You'll note that in Go, that it's not a great idea to simply add new methods to an interface, as it will break everything that's trying to implement the interface.

Something I'm a little unclear on is what guarantees we actually make about compatibility outside of the beam package. Are we basically saying that we can't change any of our internal interfaces/function definitions? As best I could tell, Unit isn't something we actively expose to users through any normal path. I know we don't at this time have a true concept of internal for most of our stuff, but I'm curious how we make decisions on where its ok to make changes like this

[damccorm]

Updated! Please take another look when you have a chance. I did the same set of manual testing with these changes integrated into the other ones

[lostluck]

I still need to look at the new code but to answer your question:

WRT Compatibility, as a rule, we no longer make breaking changes to the main beam package.
Attached to this are things the main beam package make use of or that users use to build their own DoFns (eg the typex.Window interface, event time etc.)
We strongly avoid making changes to exported surfaces of our internal packages, but can if it's the best way to do things.

We have a statement about compatibility in the release announcement but this too is woefully non-specific. It would be best if we didn't force this on our users, but that's a larger set of refactorings.

In this specific case, extending the Unit interface would at worse also require changes to the direct runner implementations. Technically the direct runner is the only reason that interface is exposed at all (and frankly for most of the exec package anyway...) (This is another reason I'm not a fan of our direct runner).

As a rule for Go, adding methods to an interface is generally (but not always) the Wrong Move. It's inherently a breaking change. Interfaces are best when they're small. Especially if the cases are optional like here, it's often better to define an additional interface and type assert to check if it implements it. Now that alone wouldn't solve this plumbing necessary here.

In this case, it's a category error: One is treating the interface like a Java abstract base class, where we can have a default implementation, propagated to all subclasses automatically. However, outside of Type Embedding, there's no such feature in Go, and type embedding itself needs careful design to make use of it. In particular since it plays havok with trying to do things with reflection on the embedded type...

[lostluck]

I need to take a second look at the actual finalization logic tomorrow, but overall this is looking good.

Don't forget to update the design doc with implementation details, for later reference.

We may want to actually cobble together a contribution guide for different kinds of features, so we don't keep running into the same implementation trickiness...

[damccorm]

R: @youngoli - adding you in Rebo's absence. I think we were pretty close on this PR anyways, I've addressed all of rebo's initial comments

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[youngoli]

I have a bunch of comments on this, but they're all small things. I think you and Robert already polished up the core architecture on this, that all looked good to me.

[damccorm]

Thanks for the review @youngoli - several good catches in there. I think all your feedback should be addressed, please take another look when you have a chance!

[youngoli]

All that's left are some nits. I'm being a bit pedantic here, but these small documentation nits tend to get ignored once they're merged in so I'd prefer to get them out of the way now.