[jira] [Commented] (FLINK-13383) Customize the problem in the trigger

["Aljoscha Krettek (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-13383?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16890910#comment-16890910 ] 

Aljoscha Krettek commented on FLINK-13383:
------------------------------------------

This can happen when the sources didn't yet emit watermarks (or when we haven't yet seen a watermark for all the parallel instances/partitions of a source). In that case, this is the expected behaviour. Could that be the case for you?

> Customize the problem in the trigger
> ------------------------------------
>
>                 Key: FLINK-13383
>                 URL: https://issues.apache.org/jira/browse/FLINK-13383
>             Project: Flink
>          Issue Type: Bug
>          Components: API / DataStream
>    Affects Versions: 1.8.0
>         Environment: The development environment is idea, the flink version is 1.8
>            Reporter: jinguishi
>            Priority: Blocker
>             Fix For: 1.8.0
>
>         Attachments: WX20190723-174236.png, WechatIMG2.png
>
>
> Using a Tumbling time window, I created a time-based and counter trigger. The parameters in the onElement method TriggerContext, ctx.getCurrentWatermark(), get negative values, 
> There are screenshots in the attachment。
> code show as below
> {code:java}
> public class CountTrigger extends Trigger<Object, TimeWindow> {
>     private static final long serialVersionUID = 1L;
>     private CountTrigger(int count) {
>         this.threshold = count;
>     }
>     private int count = 0;
>     private int threshold;
>     @Override
>     public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
>         long watermark = ctx.getCurrentWatermark();
>         ctx.registerEventTimeTimer(window.maxTimestamp());
>         if (count > threshold) {
>             count = 0;
>             return TriggerResult.FIRE;
>         } else {
>             count++;
>         }
>         System.out.println("onElement: " + element);
>         return TriggerResult.CONTINUE;
>     }
>     @Override
>     public TriggerResult onEventTime(long time, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
>         return TriggerResult.CONTINUE;
>     }
>     @Override
>     public TriggerResult onProcessingTime(long time, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
>         return TriggerResult.FIRE;
>     }
>     @Override
>     public void clear(TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
>         ctx.deleteProcessingTimeTimer(window.maxTimestamp());
>     }
>     @Override
>     public String toString() {
>         return "CountTrigger";
>     }
>     public static <W extends Window> CountTrigger of(int threshold) {
>         return new CountTrigger(threshold);
>     }
> }
> {code}
> {code:java}
>   public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
>         ReducingState<Long> fireTimestamp = ctx.getPartitionedState(stateDesc);
>         timestamp = ctx.getCurrentWatermark();
>         long end = window.maxTimestamp();
>         System.out.println(element + "  " + timestamp + "  " + window.maxTimestamp() + "  " + fireTimestamp.get());
>         if (fireTimestamp.get() == null) {
>             long start = timestamp - (timestamp % interval);
>             long nextFireTimestamp = start + interval;
>             ctx.registerEventTimeTimer(nextFireTimestamp);
>             fireTimestamp.add(nextFireTimestamp);
>         
>             return TriggerResult.CONTINUE;
>         }
>        
>         return TriggerResult.CONTINUE;
>     }
> {code}



--
This message was sent by Atlassian JIRA
(v7.6.14#76016)