Hi all,
In my pipeline setup I cannot see side outputs for Session Window (Flink
1.9.1)
What I have is:
messageStream.
.keyBy(tradeKeySelector)
.window(ProcessingTimeSessionWindows.withDynamicGap(new
TradeAggregationGapExtractor()))
.sideOutputLateData(lateTradeMessages)
.process(new CumulativeTransactionOperator())
.name("Aggregate Transaction Builder");
lateTradeMessages implementes SessionWindowTimeGapExtractor and returns 5
secodns.
Further I have:
messageStream.getSideOutput(lateTradeMessages)
.keyBy(tradeKeySelector)
.process(new KeyedProcessFunction<Long, EnrichedMessage, Transaction>() {
@Override
public void processElement(EnrichedMessage value, Context ctx,
Collector<Transaction> out) throws Exception {
System.out.println("Process Late messages For
Aggregation");
out.collect(new Transaction());
}
})
.name("Process Late messages For Aggregation");
The problem is that I never see "Process Late messages For Aggregation" when
Im sending Messages with same key.
When Session Window passes and I "immediately" sent a new message for the
same Key it triggerts new Session Window, without going into Late Side
Output.
Not sure What I'm doing wrong here.
What I would like to achieve heve is to catch "late events" and try to
reprocess them againts state that was builder for "on time" events for this
Window or if its is impossible, report late events into special Sink.
I will appreciate any help.
However it seems I do not see have any late Events.
--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
