Support Splittable DoFn in Flink Streaming Runner [python]

[damccorm]

The portable Flink runner should be able to execute SDFs in Python Streaming, and keep track of things like the watermarks for event time, and synchronized processing time.

Imported from Jira BEAM-7799. Original Jira may contain additional context.
Reported by: pabloem.

[Abacn]

Fixed by #39191