[jira] [Commented] (FLINK-29494) ChangeLogNormalize operator causes unexpected firing of past windows after state restoration

["David Anderson (Jira)" <ji...@apache.org>]

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

David Anderson commented on FLINK-29494:
----------------------------------------

[~twalthr] [~jingzhang] [~godfrey] This is an interesting problem; it feels to me like we've missed something in the overall design of these operators.

One solution might be to implement watermark checkpointing / recovery, either globally, or just in the ChangeLogNormalize operator. But it also seems problematic that the window operator is consuming -U events. But I probably don't have the big picture. What do you think?

> ChangeLogNormalize operator causes unexpected firing of past windows after state restoration
> --------------------------------------------------------------------------------------------
>
>                 Key: FLINK-29494
>                 URL: https://issues.apache.org/jira/browse/FLINK-29494
>             Project: Flink
>          Issue Type: Bug
>          Components: Connectors / Kafka, Table SQL / Runtime
>    Affects Versions: 1.14.2
>         Environment: Flink version: 1.14.2
> API: Flink SQL
>            Reporter: Rashmin Patel
>            Priority: Critical
>
> *Issue Summary:*
> While doing GroupWindowAggregation on stream produced by `upsert-kafka` connector, I am facing an unexpected behaviour, where restoring a job from checkpoint/savepoint is causing past windows(wrt last watermark generated by previous job run) to fire.
> *Detailed Description:* 
> My program is written in Flink SQL.
> Watermark Strategy: max-bounded-out-of-orderness with periodic generation (with default 200ms interval)
> Rowtime field: `updated_at_ts` which is monotonically increasing field in changelog stream produced by debezium.
> Below is the runtime topology of Flink Job
> Kafka Source (upsert mode) >>  ChangeLogNormalize >> GroupWindowAggregate >> PostgresSink
> *Job Logic Context:*
> I am reading a cdc-stream from kafka and record schema looks something like this:
> (pk, loan_acc_no, status, created_at, *updated_at,* __op).
> Now I want to count number of distinct loan_acc_no with *hourly* window. So I have created watermark on {{updated_at}} field and hence tumbling also on {{updated_at}}
> *Usual scenario which triggers unexpected late windows:*
> Now suppose that for the previous job run, the latest running window was{color:#0747a6} {{2022-09-10 08:59:59}}{color} (win_end time) and job had processed events till {{{}08:30{}}}.
> Now upon restarting a job, suppose I got a first cdc event like (pk1, loan_1, "approved", {color:#00875a}{{2022-09-02 00:00:00}}{color}, {color:#00875a}{{2022-09-10 08:45:00}}{color}, "u")  say it {*}E1{*}, which is not a late event wrt the last watermark generated by source operator in previous job run.
> Now there is ChangeLogNormalize operator in between kafka source and window operator. So, when kafka source forwards this *E1* to ChangeLogNormalize, it will emit two records which will be of type -U and +U, and will be passed as input to window operator.
>  -U (pk1, loan_1, "pending", {color:#00875a}{{2022-09-02 00:00:00}}{color}, {color:#00875a}{{2022-09-05 00:00:00}}{color}, "u") => previous state of record with key `{_}pk1{_}`
> +U (pk1, loan_1, "approved", {color:#00875a}{{2022-09-02 00:00:00}}{color}, {color:#00875a}{{2022-09-10 08:45:00}}{color}, "u") => same as E1
> So this -U type of events are causing the problem since their {{updated_at}} can be of any timestamp in the past and we are tumbling on this field. As per periodic watermarks, during the first watermark interval (i.e 200 ms default), no events will be considered late, so these -U events will create their window state and upon receiving first high watermark, windows created by these events will fire.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)