Add support for global sequence processing to the "ordered" extension in Java SDK

[slilichenko]

Global sequence processing is used to ensure that events for a given key are only processed when it's guaranteed that they all the elements for this particular have been received.

Consider a PCollection which contains these event tuples (first element of the tuple is the global sequence number):
[1, key1, data], [2, key2, data], [3, key1, data], [4, key1, data], [5, key2, data], [7, key2, data]. Elements for key1 must be processed in the following order: 1, 3, 4. Elements for key2 must be processed in the following order: 2, 5. Event with sequence 7 can't be processed because there is a missing sequence 6.

The approach used to implement ordered processing in the presence of global sequencing:

• Generate (periodically for streaming pipelines, once for batch) the side input which contains the maximum contiguous range of sequences across all keys and the maximum timestamp of events in that range.
• Use this side input by the DoFns which process events to a) store in the per-key processing state the latest maximum range and b) set up an event time based timer to fire off at the latest timestamp.
• Save all the events for a particular key into the ordered list state (with some optimization exceptions) because the events received in this DoFn are per key - they are not guaranteed to be contiguous and can't be processed right away.
• Once the timer fires off there is a guarantee that all the events for a given key have been received up to the firing timestamp. The latest contiguous range stored in the processing state is used to limit the events in the ordered list state that can be safely processed in a loop.

This high level diagram illustrates the overall approach.

There are dedicated unit tests to cover both per-key and global sequence processing. Please refer to them to understand the details of use cases.

Note that the batch unit tests for global processing don't automatically run under global sequencing. This is due to the apparent incorrectness of the DirectRunner implementation (it is supposed to block processing event processing DoFns until the side input is calculated once). The batch processing tests were successfully run manually using DataflowRunner. Additional work will be needed to either fix the DirectRunner, switch to PrismRunner (when it supports all the primitives used in this transform), or enable test to run a DataflowRunner.

[github-actions]

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @m-trieu for label java.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[github-actions]

Reminder, please take a look at this pr: @m-trieu

[github-actions]

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @damondouglas for label java.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

[damccorm]

I'm going to try to take this in 2 chunks, starting with the side input pieces.

The code itself looks good so far, thanks for the thorough tests in particular

[damccorm]

Thanks - mostly LGTM outside of pending comments (mostly cosmetic)

[damccorm]

This LGTM, thanks! I noticed that the java test suite failed, but it looks unrelated to your change (FlinkRequiresStableInputTest failed). I'm rerunning to hopefully get a green signal before merging