[BEAM-11088] Add TestStream package to Go SDK testing capabilities #15253

jrmccluskey · 2021-07-30T15:12:36Z

Create the teststream package to provide an interface for configuring the TestStream primitive in the Go SDK. Provides a type to set up a TestStreamConfig proto and insert it into the pipeline. Currently only supported on Flink.

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.

`ValidatesRunner` compliance status (on master branch)

Lang	ULR	Twister2
Go	---	---
Java
Python	---	---
XLang		---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---			---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

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

jrmccluskey · 2021-07-30T15:26:54Z

sdks/go/pkg/beam/testing/teststream/teststream.go

+	for _, val := range outputMap {
+		ret = append(ret, val)
+	}
+	return ret


This should only ever be a single PCollection, so I was thinking this should be changed so that the return type is just a beam.PCollection, but didn't know if there was a clean looking way to do that.

Agreed. Since implementations seem to all be single output collections, lets punt on tagged multi-collection map support for now. A later iteration can add a CreateN or CreateMulticall which would return a map.

jrmccluskey · 2021-07-30T15:28:58Z

sdks/go/pkg/beam/testing/teststream/teststream.go

+
+// CreatePayload converts the Config object into a TestStreamPayload to be sent to the runner.
+func (c *Config) CreatePayload() *pipepb.TestStreamPayload {
+	return &pipepb.TestStreamPayload{CoderId: "c0", Events: c.Events, Endpoint: c.Endpoint}


Hard-coding the c0 coder is a little hacky but works. The translation code always inserts a coder for the type and it's c0 as long as the TestStream is the first thing inserted into the pipeline from what I've seen.

Correct! That's what's expected for now. Add a comment to the "teststream.Create" method to call out this current limitation that the test stream must be the first PTransform in a pipeline.

jrmccluskey · 2021-07-30T15:47:55Z

R: @lostluck

jrmccluskey · 2021-07-30T15:48:56Z

Run Go PostCommit

lostluck

Great start! Lets iterate on the user experience a bit. Ruby on Rails got popular purely by having sensible defaults, and that being a key to usability is just as true now as it was then. Requiring too many manual configurations when the system can make a sensible choice for you is death by paper cuts.

lostluck · 2021-07-30T15:56:12Z

sdks/go/pkg/beam/testing/teststream/teststream.go

+
+// Package teststream contains code configuring the TestStream primitive for
+// use in testing code that is meant to be run on streaming data sources.
+// TestStream is not supported on the Go direct runner.


Add a link to the blogpost describing test stream to the Package doc.

We can clarify that teststream is only supported on Flink.

It looks like it should be supported on Dataflow, so that's worth trying and calling out. It's only weird to mention because Dataflow costs money vs the intent of this is for testing...

sdks/go/pkg/beam/testing/teststream/teststream.go

lostluck · 2021-07-30T17:26:10Z

sdks/go/pkg/beam/testing/teststream/teststream.go

+	for _, val := range outputMap {
+		ret = append(ret, val)
+	}
+	return ret


Agreed. Since implementations seem to all be single output collections, lets punt on tagged multi-collection map support for now. A later iteration can add a CreateN or CreateMulticall which would return a map.

sdks/go/pkg/beam/testing/teststream/teststream.go

lostluck · 2021-07-30T17:28:20Z

sdks/go/pkg/beam/testing/teststream/teststream.go

+
+// CreatePayload converts the Config object into a TestStreamPayload to be sent to the runner.
+func (c *Config) CreatePayload() *pipepb.TestStreamPayload {
+	return &pipepb.TestStreamPayload{CoderId: "c0", Events: c.Events, Endpoint: c.Endpoint}


Correct! That's what's expected for now. Add a comment to the "teststream.Create" method to call out this current limitation that the test stream must be the first PTransform in a pipeline.

sdks/go/pkg/beam/testing/teststream/teststream.go

sdks/go/test/integration/primitives/teststream.go

Co-authored-by: Robert Burke <lostluck@users.noreply.github.com>

…ields and helpers

…ing coders for users

sdks/go/pkg/beam/testing/teststream/teststream.go

lostluck · 2021-07-30T20:11:52Z

sdks/go/pkg/beam/testing/teststream/teststream.go

+// SetEndpoint sets a URL for a TestStreamService that will emit events instead of having them
+// defined manually. Currently does not support authentication, so the TestStreamService should
+// be accessed in a trusted context.
+func (c *Config) SetEndpoint(url string) {


Orthogonality is nice, but will delay the error case to later, dislocating it from where the mistake occured.

Alternatively, we could have a separate constructor function for Config that takes in the EndPoint and Coder so it avoids the 2 calls, and allows a user to inline the configuration in the teststream.Create call. Or even further, we just have a CreateWithEndpoint(beam.Scope, endpoint string, c coder.Coder) beam.PCollection call which avoids users from being able to misconfigure something (mixing endpoints and events in a config). Config could still handle these details for serializing as an implementation detail, but only the CreateWithEndpoint method can access and modify them before serialization. Basically, we can ensure correctness by construction, by preventing users from building something in-advisable.

Orthogonality is great, but unless all the options make sense with each other, it leaves confusing spaces for users to make cofiguration mistakes.

sdks/go/pkg/beam/testing/teststream/teststream.go

lostluck · 2021-07-30T20:53:20Z

sdks/go/pkg/beam/testing/teststream/teststream.go

+// elements.
+func (c *Config) AddElements(timestamp int64, elements ...interface{}) error {
+	if c.elmCoder == nil {
+		elmType := reflect.TypeOf(elements[0])


I'm increasingly annoyed that the node needs to know the coder it's got. It doesn't seem like it would need to since that information is on the subsequent PCollection. But it's also not great that we need to duplicate coder inference here either.

One could work around this by having a 2nd DoFn that actually contains the serialized data and keeps it in specified order. Then instead of the real data, the TestStream is only given successive int64s with the time stamps. Means we always can pass in coder.NewVarInt(), and the 2nd DoFn spits out whatever element the returned index has from it's cache. Basically the 2nd DoFn ends up as a variant on beam.Create's implementation.

Users as a rule won't be able to tell the difference, and it would resolve a small chunk of the issues we're having without moving this to the beam package with it's own Graph and graphx plumbing.

I'm open to suggestions on how we can make it easier for external to beam packages to be able to do something like this, short of exposing everything. Essentially the risk being managed is users mis-assuming that certain hoops need to be jumped through to get something working....

Looking at it now I think we could completely ditch keeping up with the coder and keep up with the element type passed into AddElements instead. beam.Create just gets the type of the first element and defines the encoder off of that using beam.NewElementEncoder in the same way I have now. End result we can remove any references to coders in AddElement.

Changed to the new, coder-free approach

sdks/go/pkg/beam/testing/teststream/teststream_test.go

sdks/go/pkg/beam/testing/teststream/teststream.go

Update Create() comment Co-authored-by: Robert Burke <lostluck@users.noreply.github.com>

Co-authored-by: Robert Burke <lostluck@users.noreply.github.com>

jrmccluskey · 2021-08-02T19:00:37Z

Run Go PostCommit

lostluck

LGTM after some light changes! Thank you.

Once we're able to observe Panes with Riteshe's trigger work, we could add passert test helpers as described in the test stream blog post, but this is good for now.

sdks/go/pkg/beam/testing/teststream/teststream.go

sdks/go/pkg/beam/testing/teststream/teststream_test.go

…ypes

jrmccluskey · 2021-08-03T18:33:17Z

Run GoPortable PreCommit

jrmccluskey · 2021-08-03T18:33:25Z

Run Go PostCommit

lostluck · 2021-08-03T19:45:36Z

Run Go Flink ValidatesRunner

…pache#15253)

franklinlindemberg · 2023-11-28T15:56:04Z

@jrmccluskey sorry to hijack this PR, but Im trying to use testStream with Flink and Im not having much success.

I'm trying to use teststream.CreateWithEndpoint but my grpc server implementation is never called. I'm wondering if you have tested this feature with Flink and if there's any example of it

Thanks!

jrmccluskey added 2 commits July 30, 2021 15:09

[BEAM-11088] Add TestStream package to Go SDK testing capabilities

6db4a3b

[BEAM-11088] Remove reference to deleted test

f1c21de

jrmccluskey commented Jul 30, 2021

View reviewed changes

[BEAM-11088] Fix integration test filters for TestStreams tests

a0da239

jrmccluskey marked this pull request as ready for review July 30, 2021 15:47

lostluck requested changes Jul 30, 2021

View reviewed changes

jrmccluskey and others added 3 commits July 30, 2021 14:04

Update sdks/go/test/integration/primitives/teststream.go

2ed4cb7

Co-authored-by: Robert Burke <lostluck@users.noreply.github.com>

[BEAM-11088] Update comments, rename functions, change exporting on f…

4f532f4

…ields and helpers

[BEAM-11088] Refactor NewConfig() to take no arguments, automate sett…

711b2e9

…ing coders for users

lostluck requested changes Jul 31, 2021

View reviewed changes

jrmccluskey and others added 4 commits August 2, 2021 13:14

Update sdks/go/pkg/beam/testing/teststream/teststream.go

98e7b0b

Update Create() comment Co-authored-by: Robert Burke <lostluck@users.noreply.github.com>

Apply suggestions from code review

3636df1

Co-authored-by: Robert Burke <lostluck@users.noreply.github.com>

[BEAM-11088] Remove coder references from TestStreamConfig

248f487

[BEAM-11088] Add CreateWithEndpoint() and make setEndpoint() unexported

28334cd

lostluck approved these changes Aug 3, 2021

View reviewed changes

jrmccluskey added 2 commits August 3, 2021 18:08

[BEAM-11088] Address nits, add validation of variadic element input t…

6f2d0e9

…ypes

[BEAM-11088] Update validation error message to cleaner form

7739716

lostluck merged commit 92ea5f5 into apache:master Aug 3, 2021

jrmccluskey deleted the beam11088 branch August 5, 2021 14:14

Quarz0 pushed a commit to Quarz0/beam that referenced this pull request Aug 10, 2021

[BEAM-11088] Add TestStream package to Go SDK testing capabilities (a…

a421215

…pache#15253)

ryanthompson591 pushed a commit to ryanthompson591/beam that referenced this pull request Aug 10, 2021

[BEAM-11088] Add TestStream package to Go SDK testing capabilities (a…

a816382

…pache#15253)

calvinleungyk pushed a commit to calvinleungyk/beam that referenced this pull request Sep 22, 2021

[BEAM-11088] Add TestStream package to Go SDK testing capabilities (a…

820668f

…pache#15253)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-11088] Add TestStream package to Go SDK testing capabilities #15253

[BEAM-11088] Add TestStream package to Go SDK testing capabilities #15253

jrmccluskey commented Jul 30, 2021

jrmccluskey Jul 30, 2021

lostluck Jul 30, 2021

jrmccluskey Jul 30, 2021

lostluck Jul 30, 2021

jrmccluskey commented Jul 30, 2021

jrmccluskey commented Jul 30, 2021

lostluck left a comment

lostluck Jul 30, 2021

lostluck Jul 30, 2021

lostluck Jul 30, 2021

lostluck Jul 30, 2021

lostluck Jul 30, 2021

lostluck Jul 31, 2021

jrmccluskey Aug 2, 2021

jrmccluskey Aug 2, 2021

jrmccluskey commented Aug 2, 2021

lostluck left a comment

jrmccluskey commented Aug 3, 2021

jrmccluskey commented Aug 3, 2021

lostluck commented Aug 3, 2021

franklinlindemberg commented Nov 28, 2023

[BEAM-11088] Add TestStream package to Go SDK testing capabilities #15253

[BEAM-11088] Add TestStream package to Go SDK testing capabilities #15253

Conversation

jrmccluskey commented Jul 30, 2021

ValidatesRunner compliance status (on master branch)

Examples testing status on various runners

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrmccluskey commented Jul 30, 2021

jrmccluskey commented Jul 30, 2021

lostluck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrmccluskey commented Aug 2, 2021

lostluck left a comment

Choose a reason for hiding this comment

jrmccluskey commented Aug 3, 2021

jrmccluskey commented Aug 3, 2021

lostluck commented Aug 3, 2021

franklinlindemberg commented Nov 28, 2023

`ValidatesRunner` compliance status (on master branch)