private static void snapshotMetricStateForSinkSubMetricData(SinkMetricData sinkMetricData,
            Integer subtaskIndex, MetricState metricState) {
        if (!(sinkMetricData instanceof SinkSubMetricData)) {
            return;
        }
        SinkSubMetricData sinkSubMetricData = (SinkSubMetricData) sinkMetricData;

        Map<String, SinkMetricData> subSinkMetricMap = sinkSubMetricData.getSubSinkMetricMap();
        if (subSinkMetricMap != null && !subSinkMetricMap.isEmpty()) {
            Map<String, MetricState> subMetricStateMap = new HashMap<>();
            Set<Entry<String, SinkMetricData>> entries = subSinkMetricMap.entrySet();
            for (Entry<String, SinkMetricData> entry : entries) {
                Map<String, Long> subMetricDataMap = new HashMap<>();
                subMetricDataMap.put(NUM_RECORDS_OUT, entry.getValue().getNumRecordsOut().getCount());
                subMetricDataMap.put(NUM_BYTES_OUT, entry.getValue().getNumBytesOut().getCount());
                subMetricDataMap.put(DIRTY_RECORDS_OUT, entry.getValue().getDirtyRecordsOut().getCount());
                subMetricDataMap.put(DIRTY_BYTES_OUT, entry.getValue().getDirtyBytesOut().getCount());
                subMetricStateMap.put(entry.getKey(), new MetricState(subtaskIndex, subMetricDataMap));
            }
            metricState.setSubMetricStateMap(subMetricStateMap);
        }
    }
Snapshot sub metric state data for {@link SinkSubMetricData}
@param sinkMetricData {@link SinkMetricData} A collection class for handling metrics
@param subtaskIndex subtask index
@param metricState state of source metric data
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/util/MetricStateUtils.java