Dropping messages based on timestamp.

[Joe Malt]

Hi,

I'm working on a custom TimestampAssigner which will do different things depending on the value of the extracted timestamp. One of the actions I want to take is to drop messages entirely if their timestamp meets certain criteria.

Of course there's no direct way to do this in the TimestampAssigner, but I'd like to keep this logic as close to the TimestampAssigner as possible since this is going to be a pluggable component used in a bunch of different Flink apps.

What would be the best way to implement this?

Thanks,

Joe

[rmetzger0]

Hi Joe,

my gut feeling is that a flatMap() is what you are looking for.

Best,

Robert

On Thu, May 28, 2020 at 7:21 PM Joe Malt <[hidden email]> wrote:

Hi,

I'm working on a custom TimestampAssigner which will do different things depending on the value of the extracted timestamp. One of the actions I want to take is to drop messages entirely if their timestamp meets certain criteria.

Of course there's no direct way to do this in the TimestampAssigner, but I'd like to keep this logic as close to the TimestampAssigner as possible since this is going to be a pluggable component used in a bunch of different Flink apps.

What would be the best way to implement this?

Thanks,

Joe

[Arvid Heise-3]

Although flatMap() is a valid choice, it would be more idiomatic to use filter(). I'd apply that even before running TimestampAssigner, except when extracting the timestamp is rather complicated. But if it's a simple field, then it feels better to first filter bad data, and then apply any kind of logic.

On Fri, May 29, 2020 at 1:13 PM Robert Metzger <[hidden email]> wrote:

Hi Joe,

my gut feeling is that a flatMap() is what you are looking for.

Best,

Robert

On Thu, May 28, 2020 at 7:21 PM Joe Malt <[hidden email]> wrote:

Hi,

I'm working on a custom TimestampAssigner which will do different things depending on the value of the extracted timestamp. One of the actions I want to take is to drop messages entirely if their timestamp meets certain criteria.

Of course there's no direct way to do this in the TimestampAssigner, but I'd like to keep this logic as close to the TimestampAssigner as possible since this is going to be a pluggable component used in a bunch of different Flink apps.

What would be the best way to implement this?

Thanks,

Joe

--