Triggering savepoint does not trigger operator notifyCheckpointComplete

When I trigger a savepoint with the Flink K8s operator, I verified for two sources (KafkaSource and MultiClusterKafkaSource) do not invoke notifyCheckpointComplete. This is easily reproducible in a simple pipeline (e.g. KafkaSource -> print). In this case, the savepoint is complete and successful, which is verified by the Flink Checkpoint UI tab and the jobmanager logs. e.g. `
Triggering checkpoint 3 (type=SavepointType{name='Savepoint', postCheckpointAction=NONE, formatType=CANONICAL})`
 

However, when the checkpoint occurs via the interval, I do see the sources checkpointing properly and expected logs in the output.

After the ticket was initially filed, I also checked with other stateful UDFs and observed the same behavior.