Treat VarInt encoding as a Beam primitive encoding in Dataflow runner

[CraigChambersG]

This PR causes the Dataflow runner to treat VarInt encoding akin to Bytes encoding, rather than a custom encoding. This fixes an issue afflicting the C++ unified worker, since it doesn't have the Java custom encoders available in the runner side of the FnAPI.

R: @robertwb

---

Post-Commit Tests Status (on master branch)

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

[kennknowles]

Tests should be running: https://github.com/apache/beam/blob/master/.test-infra/jenkins/job_PreCommit_Java.groovy#L29

[kennknowles]

retest this please

[robertwb]

Run Java PostCommit

[lgajowy]

While experimenting with load tests with @kkucharc we've faced the following error:

java.lang.ExceptionInInitializerError
	at org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$2.typedApply(IntrinsicMapTaskExecutorFactory.java:344)
	at org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$2.typedApply(IntrinsicMapTaskExecutorFactory.java:338)
	at org.apache.beam.runners.dataflow.worker.graph.Networks$TypeSafeNodeFunction.apply(Networks.java:63)
	at org.apache.beam.runners.dataflow.worker.graph.Networks$TypeSafeNodeFunction.apply(Networks.java:50)
	at org.apache.beam.runners.dataflow.worker.graph.Networks.replaceDirectedNetworkNodes(Networks.java:87)
	at org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory.create(IntrinsicMapTaskExecutorFactory.java:120)
	at org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.doWork(BatchDataflowWorker.java:337)
	at org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.getAndPerformWork(BatchDataflowWorker.java:291)
	at org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.doWork(DataflowBatchWorkerHarness.java:135)
	at org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call(DataflowBatchWorkerHarness.java:115)
	at org.apache.beam.runners.dataflow.worker.DataflowBatchWorkerHarness$WorkerThread.call(DataflowBatchWorkerHarness.java:102)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IllegalArgumentException: Multiple entries with same key: kind:varint=org.apache.beam.runners.dataflow.util.CloudObjectTranslators$8@39b69c48 and kind:varint=org.apache.beam.runners.dataflow.worker.RunnerHarnessCoderCloudObjectTranslatorRegistrar$1@7966f294
	at org.apache.beam.repackaged.beam_runners_google_cloud_dataflow_java.com.google.common.collect.ImmutableMap.checkNoConflict(ImmutableMap.java:136)
	at org.apache.beam.repackaged.beam_runners_google_cloud_dataflow_java.com.google.common.collect.RegularImmutableMap.checkNoConflictInKeyBucket(RegularImmutableMap.java:100)
	at org.apache.beam.repackaged.beam_runners_google_cloud_dataflow_java.com.google.common.collect.RegularImmutableMap.fromEntryArray(RegularImmutableMap.java:86)
	at org.apache.beam.repackaged.beam_runners_google_cloud_dataflow_java.com.google.common.collect.ImmutableMap$Builder.build(ImmutableMap.java:300)
	at org.apache.beam.runners.dataflow.util.CloudObjects.populateCloudObjectTranslators(CloudObjects.java:60)
	at org.apache.beam.runners.dataflow.util.CloudObjects.<clinit>(CloudObjects.java:39)
	... 15 more

Is this PR causing our issue?

The Jenkins logs (success, but only because we do not pass Dataflow error status by mistake). The error stacktrace is in Dataflow logs.

We generate KV<byte[], byte[]> records so my guess is that it has something to do with ByteArrayCoder that uses VarInt class. This is not really my area though so I might be missing something.

[CraigChambersG]

That stack trace looks like you've somehow not got all of the changes in the PR. Note in particular that the code that would produce the kind:varint=org.apache.beam.runners.dataflow.worker.RunnerHarnessCoderCloudObjectTranslatorRegistrar$1@7966f294 binding has been deleted in the PR.

[lgajowy]

Does the PR delete the binding fully or something else is required? I noticed this in the PR and it seemed strange to me too that I keep getting this error. However, after I reverted this commit and run the test I stopped seeing the issue.

I created an issue for this: https://issues.apache.org/jira/browse/BEAM-6349