We found that the new sink v2 interface will have a wrong numRecordsOut metric for the sink writers. We send a fixed number of records to the source, but the numRecordsOut of the sink continues to increase by the time.
The problem lies in the method `emitCommittables` in the class `SinkWriterOperator`. The field `output` in its parent class `AbstractStreamOperator` uses the same counter object as the `KafkaWriter`. It will cause the numRecordsOut increasing when doing the checkpoint.
I found this problem when we implement the metric test in the testframe, now I disable this metric test in the PR We could reopen this test case after the fix.