[FLINK-26126] Sink V2 will cause error numRecordsOut metric

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.15.0
• Fix Version/s: 1.15.0
• Component/s: Connectors / Common, Connectors / Kafka
• Labels: pull-request-available
• Assignee: Jing Ge
• Reporter: Ruan Hang
• Created: 14/Feb/22 09:55
• Updated: 01/Oct/22 14:34
• Resolved: 07/Mar/22 08:30

Description

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.

Issue Links

• blocks: FLINK-26420 error numRecordsOut metric in file connector
• relates to: FLINK-26282 Deprecate MetricNames.NUM_RECORDS_OUT_ERRORS
• relates to: FLINK-26492 rename numRecordsOutErrors to numRecordsSendErrors in SinkWriterMetricGroup
• links to: GitHub Pull Request #18825