I am trying to use process function to some processing on a set of events. I am using event time and keyed stream. The issue i am facing is Watermark value is always coming as 9223372036854725808. i have put print statement to debug and it shows like this:
timestamp------1583128014000 extractedTimestamp 1583128014000 currentwatermark-----9223372036854775808
timestamp------1583128048000 extractedTimestamp 1583128048000 currentwatermark-----9223372036854775808
timestamp------1583128089000 extractedTimestamp 1583128089000 currentwatermark-----9223372036854775808
So timestamp and extractedTimestamp changing but watermark not getting updated.So no record is getting in queue as context.timestamp is never less than watermark.
DataStream<GenericRecord> dataStream = env.addSource(searchConsumer).name("search_list_keyless");
DataStream<GenericRecord> dataStreamWithWaterMark = dataStream.assignTimestampsAndWatermarks(new SessionAssigner());
try {
dataStreamWithWaterMark.keyBy((KeySelector<GenericRecord, String>) record -> {
StringBuilder builder = new StringBuilder();
builder.append(record.get("session_id"));
builder.append(record.get("user_id"));
return builder.toString();
}).process(new MatchFunction()).print();
}
catch (Exception e){
e.printStackTrace();
}
env.execute("start session process");
}
public static class SessionAssigner implements AssignerWithPunctuatedWatermarks<GenericRecord> {
@Override
public long extractTimestamp(GenericRecord record, long previousElementTimestamp) {
long timestamp = (long) record.get("event_ts");
System.out.println("timestamp------"+ timestamp);
return timestamp;
}
@Override
public Watermark checkAndGetNextWatermark(GenericRecord record, long extractedTimestamp) {
// simply emit a watermark with every event
System.out.println("extractedTimestamp "+extractedTimestamp);
return new Watermark(extractedTimestamp - 30000);
}
}
This is the code for processFunction ....
public class MatchFunction extends KeyedProcessFunction<String, GenericRecord, Object> {
private ValueState<Tuple2<Long, PriorityQueue<GenericRecord>>> queueState = null;
@Override
public void open(Configuration config) throws Exception {
System.out.println("open");
ValueStateDescriptor<Tuple2<Long, PriorityQueue<GenericRecord>>> descriptor = new ValueStateDescriptor<>(
"sorted-events", TypeInformation.of(new TypeHint<Tuple2<Long, PriorityQueue<GenericRecord>>>() {
})
);
queueState = getRuntimeContext().getState(descriptor);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Object> out) throws Exception {
Tuple2<Long, PriorityQueue<GenericRecord>> tuple = queueState.value();
PriorityQueue<GenericRecord> records = tuple.f1;
}
@Override
public void processElement(GenericRecord record, Context context, Collector<Object> collector) throws Exception {
TimerService timerService = context.timerService();
System.out.println("currentwatermark----"+ timerService.currentWatermark());
if (context.timestamp() > timerService.currentWatermark()) {
Tuple2<Long, PriorityQueue<GenericRecord>> queueval = queueState.value();
PriorityQueue<GenericRecord> queue = queueval.f1;
long startTime = queueval.f0;
System.out.println("starttime----"+ startTime);
if (queue == null) {
queue = new PriorityQueue<>(10, new TimeStampComprator());
startTime = (long) record.get("event_ts");
}
queueState.update(new Tuple2<>(startTime, queue));
timerService.registerEventTimeTimer(startTime + 5 * 60 * 1000);
}
}
}
