public void outputMetrics(long rowCountSize, long rowDataSize, long fetchDelay, long emitDelay) {
        outputDefaultMetrics(rowCountSize, rowDataSize, fetchDelay, emitDelay);

        if (auditOperator != null) {
            for (Integer key : auditKeys) {
                auditOperator.add(
                        key,
                        getGroupId(),
                        getStreamId(),
                        System.currentTimeMillis(),
                        rowCountSize,
                        rowDataSize);
            }

        }
    }
User can use custom counter that extends from {@link Counter}
groupId and streamId and nodeId are label value, user can use it filter metric data when use metric reporter
prometheus
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceMetricData.java