[BEAM-7192] Fix partitioning of buffered elements during checkpointing

[mxm]

When a Flink checkpoint is taken, the current bundle is finalized. The
finalization happens when the checkpoint barrier has already been sent
downstream; emitting elements at this point would violate the checkpoint barrier
alignment.

When elements are emitted during checkpointing they are buffered until the
checkpoint is complete. We should ensure that they are keyed correctly and
emission of the buffered elements does not interfere with any concurrent state
requests (in case of portability).

Post-Commit Tests Status (on master branch)

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

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.

[mxm]

Run Java PreCommit

[mxm]

Run Java PreCommit

[mxm]

Flaky Dataflow test in https://builds.apache.org/job/beam_PreCommit_Java_Phrase/1066/

[tweise]

I don't think we need a lock for the output buffer since elements are only emitted by the operator thread.

I will look at the changes in more detail later.

[mxm]

Thanks for taking an initial look.

I don't think we need a lock for the output buffer since elements are only emitted by the operator thread.

I think we need a lock because elements are buffered to the state backend while the SDK harness can still process elements and make state requests.

[tweise]

Output elements are buffered by the same thread (the operator thread) that also performs shapshotState. That is why I think that no lock is necessary.

[mxm]

There are two threads:

1. Main operator thread which emits elements and buffers them in the state backend
2. GRPC thread which delegates state requests of pending elements to the state backend

So there can be concurrent writes to the state backend which must be avoided. We don't need the lock in case only the main operator thread can have access, as it is the case for flushBuffer() where the bundle is ensured to be finished. That's why there is only a lock in the buffer method.

[mxm]

Run Java PreCommit

[tweise]

There is no need to buffer output except during snapshotState. But since the bundle is still processing and may access state until finishBundle (remoteBundle.close()) returns, we need the lock?

[mxm]

That's correct.

[mxm]

Run Flink ValidatesRunner