public void outputMetrics(long rowCountSize, long rowDataSize, long dataTime) {
        outputDefaultMetrics(rowCountSize, rowDataSize);
        if (auditReporter != null) {
            for (Integer key : auditKeys) {
                auditReporter.add(
                        this.currentCheckpointId,
                        key,
                        DEFAULT_AUDIT_TAG,
                        getGroupId(),
                        getStreamId(),
                        dataTime,
                        rowCountSize,
                        rowDataSize,
                        DEFAULT_AUDIT_VERSION);
            }
        }
    }
Users can custom counter that extends from {@link Counter}
groupId and streamId and nodeId are label values,
user can use it to filter metric data when using metric reporter Prometheus
The following method is similar
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceExactlyMetric.java