notifyCheckpointComplete is never called in Flink

Question

I wrote my own operator extends from AbstractStreamOperator and OneInputStreamOperator

    implements OneInputStreamOperator<GenericRecord, Void> 


    @Override
    public void notifyCheckpointComplete(long checkpointId) throws Exception {
      ...
    }

then I transform my DataStream

OneOperator oneOperator = new OneOperator();
        input.transform(oneOperator.getClass().getSimpleName(), Types.VOID, oneOperator)
                .setParallelism(1)
                .setMaxParallelism(1)
                .addSink(new DiscardingSink<>())
                .setParallelism(1)
                .uid("oneOperator");

so that I can use the oneOperator to do something.

in my unit test, i set checkpointing to be 500ms and autowatermarkInterval to be 10, then I pass 2 events into input, the timestamp between 2 events are 2 hours apart, which is enough to trigger checkpointing.

So, in my understanding, the notifyCheckpointComplete will be called once the checkpoint is triggered.

But when I run my unit test, notifyCheckpointComplete function is never called.

Did I miss anything?

Thank you.

Answer 1

Checkpointing has nothing to do (most of the times) with Watermarking :) So, when You set Your checkpoiting to be 500ms it will be 500ms of processing time, so it really doesn't matter which data You inject. It is possible that the checkpoint is not triggered if the test finishes very quickly, so You may need to decrease the checkpointing time or add some sleep to wait for checkpoint to be triggered.