private SinkMetricData buildSinkMetricData(String topic, MetricState metricState, SinkMetricData sinkMetricData) {
        Map<String, String> labels = sinkMetricData.getLabels();
        String metricGroupLabels = labels.entrySet().stream().map(entry -> entry.getKey() + "=" + entry.getValue())
                .collect(Collectors.joining(DELIMITER));

        MetricOption metricOption = MetricOption.builder()
                .withInlongLabels(metricGroupLabels + DELIMITER + Constants.TOPIC_NAME + "=" + topic)
                .withInitRecords(metricState != null ? metricState.getMetricValue(NUM_RECORDS_OUT) : 0L)
                .withInitBytes(metricState != null ? metricState.getMetricValue(NUM_BYTES_OUT) : 0L)
                .withInitDirtyRecords(metricState != null ? metricState.getMetricValue(DIRTY_RECORDS_OUT) : 0L)
                .withInitDirtyBytes(metricState != null ? metricState.getMetricValue(DIRTY_BYTES_OUT) : 0L)
                .withRegisterMetric(RegisteredMetric.ALL)
                .build();
        return new SinkMetricData(metricOption, sinkMetricData.getMetricGroup());
    }
register sub sink metrics group from metric state

@param metricState MetricState
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SinkTopicMetricData.java