Oh I see what you mean now. I think the problem is that onProcessingTime changes nextFireTimestamp without actually setting a Trigger, as you said.
I think changing onProcessingTime to this should have the correct result: @Override public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception { ValueState<Long> fireState = ctx.getPartitionedState(stateDesc); long nextFireTimestamp = fireState.value(); // only fire if an element didn't already fire long currentTime = System.currentTimeMillis(); if (currentTime > nextFireTimestamp) { fireState.update(0); // <- set to 0 so that onElement will set a timer return TriggerResult.FIRE; } return TriggerResult.CONTINUE; } What do you think? This should have the behavior that it continuously fires, but only if new elements arrive. Cheers, Aljoscha On Thu, 31 Mar 2016 at 14:46 Hironori Ogibayashi <ogibaya...@gmail.com> wrote: > Aljoscha, > > Thanks for your response. > I understood that trigger is only set when new elements arrive, but in > my previous example, trigger fired at > 20:51:40.002, then new element arrived at 20:51:41, 42, 43. So why > next trigger did not set at 20:51:45? > > It looks like the following situation. > - 20:51:40.002 onProcessingTime called, and the trigger fires. In the > same method, fireState was updated to 20:51:45. but > registerProcessingTimeTimer wad not called, so next timer was not > actually set. > - 20:51:41 next element comes and onElement called. Since > currentTime(21:51:41) < nextFireTimeStamp (20:51:45), > it just return TriggerResult.CONTINUE. Next timer was not set. > > I think next time should be set at 20:51:45 when an element comes at > 20:51:41. > Am I mis-understanding? > > Regards, > Hironori > > 2016-03-31 18:08 GMT+09:00 Aljoscha Krettek <aljos...@apache.org>: > > Hi, > > yes, right now this is expected behavior. But I see that it can be a bit, > > well, unexpected. > > > > The continuous trigger is only set when new elements arrive, so only when > > you put new elements does the trigger fire again after five seconds. If > you > > want it to truly continuously fire every five seconds even though no new > > elements arrived you can change the "onProcessingTime" method to this: > > > > @Override > > public TriggerResult onProcessingTime(long time, W window, TriggerContext > > ctx) throws Exception { > > > > ValueState<Long> fireState = ctx.getPartitionedState(stateDesc); > > long nextFireTimestamp = fireState.value(); > > > > // only fire if an element didn't already fire > > long currentTime = System.currentTimeMillis(); > > if (currentTime > nextFireTimestamp) { > > long start = currentTime - (currentTime % interval); > > fireState.update(start + interval); > > ctx.registerProcessingTimeTimer(start + interval); // <-- I > added > > this call > > return TriggerResult.FIRE; > > } > > return TriggerResult.CONTINUE; > > } > > > > I hope this helps. As I mentioned in the other thread I'm currently > thinking > > about how to make the triggers more intuitive since right now they are > not > > very easy to comprehend because the names can also be misleading. > > > > Cheers, > > Aljoscha > > > > On Wed, 30 Mar 2016 at 14:33 Hironori Ogibayashi <ogibaya...@gmail.com> > > wrote: > >> > >> Hi > >> > >> I noticed that ContinuousProcessingTimeTrigger sometimes does not fire. > >> > >> I asked similar question before and applied this patch. > >> > >> > https://github.com/apache/flink/commit/607892314edee95da56f4997d85610f17a0dd470#diff-19bbcb3ea1403e483327408badfcd3f8 > >> It looked work but still I have strange behavior. > >> > >> The code is: > >> > >> ---- > >> val env = StreamExecutionEnvironment.getExecutionEnvironment > >> val input = > >> > >> > env.readFileStream(fileName,100,FileMonitoringFunction.WatchType.PROCESS_ONLY_APPENDED) > >> .flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } } > >> .windowAll(TumblingProcessingTimeWindows.of(Time.days(1))) > >> .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(5))) > >> .fold(Set[String]()){(r,i) => { r + i}} > >> .map{x => (new Timestamp(System.currentTimeMillis()), x.size)} > >> > >> input print > >> --- > >> > >> This case, the base window is long, so I just expect cumulative > >> distinct count of the value every 5 seconds. > >> > >> Appended 8 strings to the input file with 1 second interval. > >> > >> --- > >> % for i in `seq 1 8`; do date; echo "aa${i}" >> ~/tmp/input.txt; sleep > >> 1; done > >> Wed Mar 30 20:51:36 JST 2016 > >> Wed Mar 30 20:51:37 JST 2016 > >> Wed Mar 30 20:51:38 JST 2016 > >> Wed Mar 30 20:51:39 JST 2016 > >> Wed Mar 30 20:51:40 JST 2016 > >> Wed Mar 30 20:51:41 JST 2016 > >> Wed Mar 30 20:51:42 JST 2016 > >> Wed Mar 30 20:51:43 JST 2016 > >> --- > >> > >> But I only received 1 output event. I should receive one more event 5 > >> seconds later, but actually nothing. > >> > >> (2016-03-30 20:51:40.002,4) > >> > >> Later, if I put additional line to the file. I got these events. > >> > >> (2016-03-30 21:12:05.39,9) > >> (2016-03-30 21:12:10.001,9) > >> > >> I slightly modified ContinuousProcessingTimeTrigger.java and added > >> logging in onProcessingTime method. It looks like the method was > >> called at 20:51:40 and 21:12:10, not at 20:51:45 and 21:12:05. > >> > >> ---- > >> 2016-03-30 20:51:40,002 INFO > >> > >> > org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger > >> - onProcessingTime called: 2016-03-30 20:51:40.002 > >> ... > >> 2016-03-30 21:12:10,001 INFO > >> > >> > org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger > >> - onProcessingTime called: 2016-03-30 21:12:10.001 > >> ---- > >> > >> Is this an expected behavior? > >> > >> Regards, > >> Hironori >