Since filter and flatMap can change the stream the pipeline sink implementations for those operations should not pass size information on Sink.begin to the downstream sink, instead -1 should be passed, indicated the size is unknown.