Flink KeyedProcessFunction<K,I,O>.Context .timerService().currentWatermark() is always Long.MIN

user_19180624 0
Viewed 56

I am currently trying to write some tests for our flink stream processing, and have something such as

env.fromCollection(events)
            .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<>(Time.milliseconds(1000)) {
                @Override
                public long extractTimestamp(CallEvent element) {
                    return element.getEventTimeStamp();
                }
            })
            .keyBy(CALL_EV_KEY_SELECTOR)
            .process(new CallEventKeyedProcessFunction())
            .map(new CallTimedEntityMapper())
            .addSink(sink);

    env.execute();

in CallEventKeyedProcessFunction, when attempting to call ctx.timerService().currentWatermark();, this will always return Long.MIN. Is there anything obvious I might be missing?

apache-flink watermark java
1 Answers

You should set periodic watermark interval:

env.getConfig().setAutoWatermarkInterval(millis);

Also, for Flink < 1.12 you should enable event time:

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/event-time/generating_watermarks/#writing-a-periodic-watermarkgenerator

If you are trying to write tests, I would recommend using one of the provided test harnesses, since then you can input your own (deterministic) watermark and do not have to rely on auto watermark interval, which can give you different results on every run. In your case, the full input collection could be processed before a single watermark is generated, even if you set the auto watermark interval to 1 ms.

https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/testing/#unit-testing-stateful-or-timely-udfs--custom-operators

user_46489690
Related