The requirement is to send alerts if the expected 'final' event (identified from the evenType field in event payload) is not received for a key within a time window of say 2 minutes on the input topic. I tried using suppress() as follows:
events
.groupByKey()
.windowedBy(TimeWindows.of(Duration.ofMinutes(2)))
.reduce((oldValue, newValue) -> newValue, Materialized.as(....))
.suppress(Suppressed.untilWindowClose(BufferConfig.unbounded()))
.filter(event -> !event.getEventType().equals("FINAL") )
.toStream()
.foreach( /* Send that email! */)
based on this blog.
However, since the suppress() is based on event-time, if the next or the 'final' event is not received, the time doesn't progress and the alert is not sent at all. How can I use wall-clock time with suppress() to emit the event when 2 minutes has elapsed based on wall-clock time?