Add Stream Operator that Discards Late Elements

We should introduce an operator (and API call on DataStream) that allows users to discard late elements (with respect to their timestamps and the current watermark). The operator could have two modes, one for silently discarding elements that are late and one for failing the job if late elements arrive (as a sort of sanity check).