[Feature] Flink Connector support pendingRecords metric

[luoyuxia]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Although we has coverd most of standard metrics defined in FLIP-33: Standardize Connector Metrics, we still miss the metrics of pendingRecords which help us to trace data backlog.

Solution

pending records = latest log offset - fetched log offset.

For snapshot read phase of primary key table, it should be total records in kv snapshot - records emited, but to simplify, we can just trace the phase of incremental reading which read from change log.

Anything else?

No response

Willingness to contribute

• I'm willing to submit a PR!

[Alibaba-HZY]

I would like to contribute this issue, please assign it to me

[wuchong]

convert to issue #138