Hi everyone, The pipeline I am trying to implement is an aggregation of a data stream across a given window, that uses a custom trigger to count events and outputs a new event for further processing by a downstream operator when that count is met. However, I do not want to start counting messages again until the end of the window. I am attempting to use a custom count trigger to output from my aggregation before the window is closed, but wait to start counting again until the end of the window. I am successfully triggering and creating a new event, but Flink is waiting for my window to close before sending my new event downstream. Does anyone know how to accomplish this? Here is some example code of what I am running:

DataStream<Event> output =
      keyedStream
        .window(TumblingEventTimeWindows.of(Time.seconds(60)))
        .trigger(CustomCountTrigger.of(5))
        .aggregate(new MyCustomAggregateFunction(), new MyCustomProcessWindowFunction());

In this example,

MyCustomProcessWindowFunction

is calling

collector.collect(new Event())

when my count is met, but my output does not contain the new Event until after 60 seconds.