flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Kostas Kloudas (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-7606) CEP operator leaks state
Date Fri, 15 Sep 2017 16:44:00 GMT

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

Kostas Kloudas commented on FLINK-7606:

Now that I think about it, can you check your watermarks and their values? And if the watermark
advances past the 10 min + 10 sec (your bounded out-of-orderness) in event time.

If the watermark does not advance then the patterns do not time out.
This, in combination with a potential re-ordering of events (as I described before) can lead
to the NFA always having some stale "Idle" elements that are waiting for their corresponding
"Start" which never arrives because it was dropped as late.

To verify this, could you set the parallelism of your job to 1, make sure that "Idle" precedes
"Start" and at the end of the iteration, your source sends a Watermark with a high timestamp
so that all buffered elements are flushed or timed-out. This is a very controlled experiment
just to see where we stand.

It would also be useful if you have the timestamp generator for your input elements. 
To see how time and the watermark advances.
Then I could give you more detailed answers.

> CEP operator leaks state
> ------------------------
>                 Key: FLINK-7606
>                 URL: https://issues.apache.org/jira/browse/FLINK-7606
>             Project: Flink
>          Issue Type: Bug
>          Components: CEP
>    Affects Versions: 1.3.1
>            Reporter: Matteo Ferrario
>         Attachments: heap-dump1.png, heap-dump2.png, heap-dump3.png
> The NestedMapsStateTable grows up continuously without free the heap memory.
> We created a simple job that processes a stream of messages and uses CEP to generate
an outcome message when a specific pattern is identified.
> The messages coming from the stream are grouped by a key defined in a specific field
of the message.
> We've also added the "within" clause (set as 5 minutes), indicating that two incoming
messages match the pattern only if they come in a certain time window.
> What we've seen is that for every key present in the message, an NFA object is instantiated
in the NestedMapsStateTable and it is never deallocated.
> Also the "within" clause didn't help: we've seen that if we send messages that don't
match the pattern, the memory grows up (I suppose that the state of NFA is updated) but it
is not cleaned also after the 5 minutes of time window defined in "within" clause.
> If you need, I can provide more details about the job we've implemented and also the
screenshots about the memory leak.

This message was sent by Atlassian JIRA

View raw message