[
https://issues.apache.org/jira/browse/FLINK-9547?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16505782#comment-16505782
]
Dawid Wysakowicz commented on FLINK-9547:
-----------------------------------------
Now that I have whole picture I can explain it a bit. In the example you
provided the first version does not produce any results because you enabled
{{EventTime}} and do not generate timestamps nor watermarks. Therefore no
events are processed, cause the CEP library sorts events on watermark and only
then processes them.
Try running in ProcessingTime or implement proper
[Watermarking|https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/event_timestamps_watermarks.html#assigning-timestamps.
Also make sure you emit enough events to close window.
To sum up I don't think it is a valid bug, therefore will close this ticket. If
you need more help please ask on user mailing list:
[http://flink.apache.org/community.html|http://flink.apache.org/community.html.]
> CEP pattern not called on windowed stream
> -----------------------------------------
>
> Key: FLINK-9547
> URL: https://issues.apache.org/jira/browse/FLINK-9547
> Project: Flink
> Issue Type: Bug
> Components: CEP
> Affects Versions: 1.3.2, 1.5.0
> Reporter: Lucas Resch
> Priority: Major
>
> When trying to match a pattern on a stream that was windowed the pattern will
> not be called. The following shows example code where the issue was noticed:
> {code:java}
> // Set up stream
> SingleOutputStreamOperator<ForceZ> forces = ...
> .filter(new FilterForcesFunction())
> .process(new ProcessForcesFunction());
> // Define mock pattern
> Pattern<ForceZ, ?> forcesMock = Pattern.<ForceZ>begin("start").where(new
> SimpleCondition<ForceZ>() {
> @Override
> public boolean filter(ForceZ value) {
> // This is called as expected
> return true;
> }
> });
> // Print pattern results
> // This actually prints all incoming events as expected
> CEP.pattern(forcesMock, mock)
> .select(new PatternSelectFunction<ForceZ, ForceZ>() {
> @Override
> public ForceZ select(Map<String, List<ForceZ>> pattern){
> return pattern.get("start").get(0);
> }
> }).print();
> // Create another stream based on a sliding window over the input stream
> SingleOutputStreamOperator<Interval> intervals = forces
> .countWindowAll(2, 1)
> .process(new ForceWindowFunction());
> // Define mock pattern
> Pattern<Interval, Interval> intervalMock =
> Pattern.<Interval>begin("start").where(new SimpleCondition<Interval>() {
> @Override
> public boolean filter(Interval value) throws Exception {
> // This is never called
> return true;
> }
> });
> // Print pattern results
> // Doesn't print anything since the mock condition is never called
> CEP.pattern(intervals, intervalMock)
> .select(new PatternSelectFunction<Interval, Interval>() {
> @Override
> public Interval select(Map<String, List<Interval>> pattern)
> throws Exception {
> return pattern.get("start").get(0);
> }
> }).print();
> {code}
> Either I'm doing something wrong or this is a major bug.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
