Flink*DoFnFunction: fix check for single-output dofns

[dhalperi]

Fixes Findbugs and (presumably) increases efficiency by using the right OutputManager.

R: @aljoscha but since this is a PostCommit break we may merge sooner if tests pass.

R: @tgroh @kennknowles

[coveralls]

Changes Unknown when pulling 1590032 on dhalperi:flink-output-manager into ** on apache:master**.

[asfbot]

--none--

[dhalperi]

Build was not green - failed in integration tests.

[aljoscha]

Do you have a link to the failed integration tests?

[dhalperi]

@aljoscha :

SEVERE: Error in task code:  CHAIN MapPartition (MapPartition at ParMultiDo(AddTimestamp)) -> FlatMap (FlatMap at ParDo(AddTimestamp)/ParMultiDo(AddTimestamp).out0) -> FlatMap (Window.Into()/Window.Assign.out) (1/1)
org.apache.beam.sdk.util.UserCodeException: org.apache.flink.runtime.operators.chaining.ExceptionInChainedStubException
	at org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:36)
	at org.apache.beam.examples.WindowedWordCount$AddTimestampFn$DoFnInvoker.invokeProcessElement(Unknown Source)
	at org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:177)
	at org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:141)
	at org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement(DoFnRunnerWithMetricsUpdate.java:65)
	at org.apache.beam.runners.flink.translation.functions.FlinkDoFnFunction.mapPartition(FlinkDoFnFunction.java:120)
	at org.apache.flink.runtime.operators.MapPartitionDriver.run(MapPartitionDriver.java:103)
	at org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:490)
	at org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:355)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:665)
	at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.flink.runtime.operators.chaining.ExceptionInChainedStubException
	at org.apache.flink.runtime.operators.chaining.ChainedFlatMapDriver.collect(ChainedFlatMapDriver.java:82)
	at org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
	at org.apache.beam.runners.flink.translation.functions.FlinkDoFnFunction$DoFnOutputManager.output(FlinkDoFnFunction.java:149)
	at org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:211)
	at org.apache.beam.runners.core.SimpleDoFnRunner.access$700(SimpleDoFnRunner.java:66)
	at org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.outputWithTimestamp(SimpleDoFnRunner.java:433)
	at org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.outputWithTimestamp(SimpleDoFnRunner.java:420)
	at org.apache.beam.examples.WindowedWordCount$AddTimestampFn.processElement(WindowedWordCount.java:120)

This is from running WindowedWordCountIT in batch mode.

Since the old code would simply never have invoked the single-output DoFnFunction, maybe it's broken?

[dhalperi]

@aljoscha in e0434d3 I copied the code that replaces the value with a RawUnionValue, and that seems to have made the tests pass. Error messaging could be better here.

[dhalperi]

And flink module + flink ITs passed.

[dhalperi]

Going ahead and self-merging to unbreak the build. Please continue to review when you have time, @aljoscha

Thanks!

[coveralls]

Changes Unknown when pulling e0434d3 on dhalperi:flink-output-manager into ** on apache:master**.

[asfbot]

Build finished.
--none--

[aljoscha]

@dhalperi LGTM! Thanks for fixing this quickly and merging already. 👍