[FLINK-37153] Monitor late event count in temporal join

[grzegorz8]

What is the purpose of the change

In temporal join, events on the right side (build side, versioned table) are not discarded even if they arrive late (event timestamp < current watermark), which is generally good.

As far as I understand, late events can sometimes lead to incorrect results. This happens when the latest right-side event is delayed for some reason, and as the watermark progresses, a left-side event gets joined with the most recently known right-side event (stored in Flink's state), which might not be the true latest one.

I believe it would be valuable to have a metric tracking late event count on the right side, just to be aware if such situation might have occurred.

Brief change log

Expose rightNumLateRecords metric in temporal join.

Verifying this change

This change is a trivial rework.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not documented)

[davidradl]

I notice that we already have numLateRecordsDropped metric. Is this sufficient?
You are proposing a new metric for a subset of operators (ones with 2 inputs) and only for one side, where all the processing is the same for the left and right sides at the moment.
I wonder if the metric should be when the late record is dropped in line with the existing metric
We would need documentation to be added for this.

[grzegorz8]

I notice that we already have numLateRecordsDropped metric. Is this sufficient?

Yes, but not for temporal join. There is numLateRecordsDropped metric in window operators or cep operator.
What is more, late events ARE NOT dropped in this operator, that is why I named it rightNumLateRecordsDropped.

You are proposing a new metric for a subset of operators (ones with 2 inputs) and only for one side, where all the processing is the same for the left and right sides at the moment.

I just wanted to prompt a discussion with this PR :) We can easily add such metric for left side as well.

[flinkbot]

CI report:

• 37241bc Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build