DirectRunner: watermarks are updated asynchronously from bundle processing

Watermarks are updated in QuiescenceDriver (by calling fireTimers, which calls forceRefresh()) on WatermarkManager. This results in creating timer bundles, that are then processed asynchronously as DirectTransformExecutor. Because of that, watermarks (input watermarks mostly) might be updated while bundle is being processed. That violates assumption, that bundle processing should be atomical (with identical external conditions during processing of whole bundle).