DirectRunner: race condition in watermark update

When watermark is updated in upstream PTransform, it is instantly propagated as input watermark to all directly connected downstream PTransforms. We must ensure that all output bundles that were emitted before the watermark update are processed before we allow updating input watermark of downstream PTransform.