Ensure that the payload is consumed on Elements.Data and Elements.Timers when is_last = true

[damccorm]

BeamFnDataInboundObserver drops the bytes instead of passing it forward on the last message:

beam/sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java

Line 58 in cd4b7f3

if (isLast) {

data_plane.py drops the timers payload but seems to do the right thing for data:

beam/sdks/python/apache_beam/runners/worker/data_plane.py

Line 343 in cd4b7f3

if not element.is_last:

data_mgr.go drops the data payload (timers don't seem to be supported):

beam/sdks/go/pkg/beam/core/runtime/harness/datamgr.go

Line 297 in cd4b7f3

continue

Imported from Jira BEAM-13142. Original Jira may contain additional context.
Reported by: lcwik.

[damccorm]

@lostluck should this be closed or is it just fixed for Go?

[lostluck]

Ah good catch.

Java is correct: Bytes are not dropped for either data or timers.

Python remains how it was on first filing:
Data is always propagated, timers are only propagated if not last in the In Memory channel:
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/worker/data_plane.py#L363
The GRPC channel just drops both.
https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/worker/data_plane.py#L516

I'm still unfamiliar with timers in that respect though (ask me again after I add them to Prism). I'd expect the intent behind this issue is to allow runners to avoid extra messages, which does matter for streaming. That benefit is superseded if the SDK supports the "Elements in ProcessBundle" extensions though, which if the Python SDK supports, then this issue may be obselete.

[lostluck]

It has been determined that Go SDK was breaching FnAPI contract in how it implemented timer writers. By not eagerly creating them, we never had a clear is_last signal for timers for the runner to key off of.

That specific issue is filed in #29605