Apache Flink CEP с дополнительным событием не работает должным образом ⇐ JAVA

[Anonymous]

У меня есть такой шаблон:

Код: Выделить всё

Pattern pattern = Pattern.begin("A", AfterMatchSkipStrategy.noSkip())
.where(new IterativeCondition() {
@Override
public boolean filter(Event value, Context ctx) {
return value.getType().contains("A");
}
})
.next("B").where(new IterativeCondition() {
@Override
public boolean filter(Event value, Context ctx) {
return value.getType().contains("B");
}
})
.next("C").where(new IterativeCondition() {
@Override
public boolean filter(Event value, Context ctx) {
return value.getType().contains("C");
}
})
.optional()
.within(Duration.ofSeconds(10));

Я хочу, чтобы событие C было необязательным. Мой поток ввода:

Код: Выделить всё

DataStream eventStream = env.fromData(
new Event(1, "A"),
new Event(1, "B"),
new Event(1, "C")

)
.assignTimestampsAndWatermarks(new IngestionTimeExtractor());

И утонем:

Код: Выделить всё

PatternStream patternStream = CEP.pattern(eventStream.keyBy(Event::getId), pattern);

SingleOutputStreamOperator eventsBuildStream = patternStream
.process(new EventsToContainerPPF());

eventsBuildStream.addSink(new SinkFunction() {
@Override
public void invoke(MatchedEventsContainer value, Context context) {
log.debug(">>> sink: {}", value);
}
});

Я добавил вывод в журнал объекта, который содержит все соответствующие события.
В журнале я вижу следующее:

Код: Выделить всё

>>> sink: OptionalEventTestJob.MatchedEventsContainer(events={A=[OptionalEventTestJob.Event(id=1, type=A)], B=[OptionalEventTestJob.Event(id=1, type=B)]})
>>> sink: OptionalEventTestJob.MatchedEventsContainer(events={A=[OptionalEventTestJob.Event(id=1, type=A)], B=[OptionalEventTestJob.Event(id=1, type=B)], C=[OptionalEventTestJob.Event(id=1, type=C)]})

Как видите, приемник вызывался два раза, в первом случае для A и B, во втором случае для A, B и C.
Но я ожидал увидеть только одну строку, в которой будут A, B , и события C. А если событие C отсутствует во входящем потоке, то только A и B.
Почему происходит двукратный вывод и как мне получить ожидаемый результат?

Подробнее здесь: https://stackoverflow.com/questions/790 ... s-expected