[jira] [Commented] (FLINK-11408) ContinuousProcessingTimeTrigger: NPE on clear() and state is lost on merge

["Kezhu Wang (Jira)" <ji...@apache.org>]

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

Kezhu Wang commented on FLINK-11408:
------------------------------------

I think it is same as FLINK-5717 and similar to event time trigger counterpart FLINK-4862.

The problems is:
 * In merging, old trigger states are merged out. In later {{clear}} for old triggers, they will get no {{fireTimestampState}}. This causes Bug1.
 * In merging, only state got merged into new window namespace but processing timer was not registered there. This causes Bug2.

[~aljoscha] [~mxm] [~dwysakowicz] Could I take over this due to stale FLINK-5717 ? I plans to add test similar to {{ContinuousEventTimeTriggerTest.testMergingWindows}} to cover both cases.

> ContinuousProcessingTimeTrigger: NPE on clear() and state is lost on merge
> --------------------------------------------------------------------------
>
>                 Key: FLINK-11408
>                 URL: https://issues.apache.org/jira/browse/FLINK-11408
>             Project: Flink
>          Issue Type: Bug
>          Components: API / DataStream
>         Environment: Put both bugs in [https://github.com/casidiablo/flink-continuous-processing-trigger-bugs]
> This is running Flink 1.7.1 locally.
>            Reporter: Cristian
>            Priority: Major
>
> I was testing session windows using processing time and found a couple of problems with the 
> ContinuousProcessingTimeTrigger. The first one is an NPE in the clear method:
> [https://github.com/casidiablo/flink-continuous-processing-trigger-bugs/blob/master/src/main/java/flink/bug/Bug1.java]
> The second one, which is most likely related and the root cause of the first one, is that the way the state is merged for windows that are merged somehow makes it so that the trigger gets confused and it stops triggering:
>  
> [https://github.com/casidiablo/flink-continuous-processing-trigger-bugs/blob/master/src/main/java/flink/bug/Bug2.java]
>  
> I managed to solve both of these by using a modified version of 
> ContinuousProcessingTimeTrigger which does NOT call `ctx.mergePartitionedState(stateDesc);` in the onMerge method.
> This is what I understand it happens at the trigger level:
>  * The first element in the stream sets an initial fire time (logic is in
> ContinuousProcessingTimeTrigger#onElement()), if there is no trigger set.
>  * From then on, ContinuousProcessingTimeTrigger#onProcessingTime() is responsible for scheduling the next trigger. 
>  * When the windows are merged (in the case of session windows), somehow the clear and merge methods are called using the wrong window namespace (I think this is the root cause of the bug, but I'm not too familiar with that code).
>  * Because the state is not cleared properly in the new window namespace, the previously scheduled trigger gets executed against the window that was cleared.
>  * Moreover, the new window has the state of the previous window, which means that:
>  ## onElement will NOT schedule a fire trigger
>  ## onProcessingTime will never be called at all



--
This message was sent by Atlassian Jira
(v8.3.4#803005)