Verbose print causes pipelines to get stuck on FlinkRunner.

Minimal reproduction: 
 
`def _run_pipeline(pipeline):`
`    def process_data(unused):`
`        print('a'*1000)`

`    _ = (`
`            pipeline`
`            | "Create" >> beam.Create(['a']*1000)`
`            | "Process" >> beam.Map(process_data)`
`    )`
`    pipeline.run().wait_until_finish()`

See thread at https://lists.apache.org/thread/r971f432mvko4g3hmnrodh8s4n2wkk6p

Imported from Jira [BEAM-14476](https://issues.apache.org/jira/browse/BEAM-14476). Original Jira may contain additional context.
Reported by: robertwb.

Start with the minimal _run_pipeline reproduction and the linked Apache mailing-list thread, using FlinkRunner and pipeline.run().wait_until_finish(). Trace why repeated print output causes the pipeline to stop progressing; done means the reproduction completes without getting stuck and the expected output is handled correctly.

Written by the indexing model from the issue text.