@Override
    public void collect(T record) {
        if (sourceMetricsReporter != null) {
            sourceMetricsReporter.outputMetricsWithEstimate(record, timestampMillis);
        }
        collector.collect(record);
    }
sending metrics each time a record is collected.
@param <T>
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricsCollector.java