Stop the flow before examining the stream to avoid `ConcurrentModificationException`. **cherry-pick to 5.0.x**