Hi Ori, one way to do it is to implement a basic ProcessFunction. ProcessFunction.processElement(I value, Context ctx, Collector<O> out) offers access to the context through which you can access the current watermark timestamp using ctx.timerService().currentWatermark(). That you can use to filter out delayed events.
Best, Matthias On Thu, Sep 24, 2020 at 9:59 AM Ori Popowski <ori....@gmail.com> wrote: > I need to drop elements which are delayed by more than a certain amount of > time from the current watermark. > > I wanted to create a FilterFunction where I get the current watermark, > and if the difference between the watermark and my element's timestamp is > greater than X - drop the element. > > However, I do not have access to the current watermark inside any of > Flink's operators/functions including FilterFunction. > > How can such functionality be achieved? > -- Matthias Pohl | Engineer Follow us @VervericaData Ververica <https://www.ververica.com/> -- Join Flink Forward <https://flink-forward.org/> - The Apache Flink Conference Stream Processing | Event Driven | Real Time -- Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany -- Ververica GmbH Registered at Amtsgericht Charlottenburg: HRB 158244 B Managing Directors: Yip Park Tung Jason, Jinwei (Kevin) Zhang, Karl Anton Wehner
