Hi, If you want to register a processing/event time trigger in your custom operator, you can take a look how other operators are doing it, by calling AbstractStreamOperator#getInternalTimerService [1]. You can look around the Flink’s code base for usages of this method, there are at least couple of them (like CepOperator or IntervalJoinOperator).
Hope that helps, Piotrek [1] https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.html#getInternalTimerService-java.lang.String-org.apache.flink.api.common.typeutils.TypeSerializer-org.apache.flink.streaming.api.operators.Triggerable- <https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.html#getInternalTimerService-java.lang.String-org.apache.flink.api.common.typeutils.TypeSerializer-org.apache.flink.streaming.api.operators.Triggerable-> > On 28 Oct 2019, at 10:09, Felipe Gutierrez <felipe.o.gutier...@gmail.com> > wrote: > > Hi all, > > I have my own stream operator which trigger an aggregation based on the > number of items received > (OneInputStreamOperator#processElement(StreamRecord)). However, it is > possible to not trigger my aggregation if my operator does not receive the > max items that have been set. So, I need a timeout trigger. > > I am confused if I need to extend Trigger on > MyPreAggregate-AbstractUdfStreamOperator or if I have to put a window as a > parameter of the operator class MyPreAggregate-AbstractUdfStreamOperator<K, > V, IN, OUT, W extends Window>. what is the best approach? > > Thanks, > Felipe > -- > -- Felipe Gutierrez > -- skype: felipe.o.gutierrez > -- https://felipeogutierrez.blogspot.com > <https://felipeogutierrez.blogspot.com/>
