[BEAM-9624] Adds Convert to Accumulators operator for use in combiner lifting for streaming pipelines

[acrites]

The Convert To Accumulators operation is defined here:

https://s.apache.org/beam-runner-api-combine-model#heading=h.h5697l1scd9x

It is used by streaming pipelines where we want to lift the combiner into the GBK, but don't want to also lift the combiner into a PGBK before the GBK. This PGBK can behave badly with certain triggers.

This PR also adds an implementation of Convert to Accumulators for the Go, Java, and Python SDKs.

R: @robertwb, @lukecwik, @lostluck

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.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go		---	---		---	---
Java
Python		---			---	---
XLang	---	---	---		---	---

Pre-Commit Tests Status (on master branch)

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

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

[lukecwik]

I didn't see a place where you could add a Python test that was similar to the Go/Java ones.

@udim Any suggestions?

[lukecwik]

retest this please

[lukecwik]

Java Precommit failed due to:

19:54:56 > Task :sdks:java:harness:compileTestJava
19:54:56 /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Commit/src/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/CombineRunnersTest.java:350: error: method createRunnerForPTransform in interface PTransformRunnerFactory<T> cannot be applied to given types;
19:54:56         .createRunnerForPTransform(
19:54:56         ^
19:54:56   required: PipelineOptions,BeamFnDataClient,BeamFnStateClient,String,PTransform,Supplier<String>,Map<String,PCollection>,Map<String,Coder>,Map<String,WindowingStrategy>,PCollectionConsumerRegistry,PTransformFunctionRegistry,PTransformFunctionRegistry,Consumer<ThrowingRunnable>,BundleSplitListener,BundleFinalizer
19:54:56   found: PipelineOptions,<null>,<null>,String,PTransform,<null>,Map<Object,Object>,Map<Object,Object>,Map<Object,Object>,PCollectionConsumerRegistry,PTransformFunctionRegistry,PTransformFunctionRegistry,<null>,<null>
19:54:56   reason: actual and formal argument lists differ in length
19:54:56   where T is a type-variable:
19:54:56     T extends Object declared in interface PTransformRunnerFactory

[lostluck]

Other than a missing test that Luke's already suggested, LGTM for Go. The implementation matches the spec description.

[acrites]

I fixed the problem in :sdks:java:harness:compileTestJava. It turns out my branch was old and createRunnerForPTransform only had 14 parameters, but now it's 15.

[lukecwik]

retest this please

[lukecwik]

retest this please

[lukecwik]

retest this please

[udim]

I didn't see a place where you could add a Python test that was similar to the Go/Java ones.

@udim Any suggestions?

My suggestion is to look at combiners_test.py and add the appropriate test there.

[acrites]

The main issue is that we need to be able to test one of the pieces of the CombinePerKey operation (ConvertToAccumulators) independently of the others like MergeAccumulators and Extract. In Go and Java there were already some test frameworks in place to do that, but in Python we can only test the entire thing.