@Override
    public void initializeState(FunctionInitializationContext functionInitializationContext) throws Exception {
        if (sinkOptions.getSemantic() != StarRocksSinkSemantic.EXACTLY_ONCE) {
            return;
        }

        String transactionStateName = "starrocks-sink-transaction"
                + (StringUtils.isNullOrWhitespaceOnly(stateKey) ? "" : "-" + stateKey);
        ListStateDescriptor<byte[]> descriptor =
                new ListStateDescriptor<>(
                        transactionStateName,
                        TypeInformation.of(new TypeHint<byte[]>() {
                        }));

        ListState<byte[]> listState = functionInitializationContext.getOperatorStateStore().getListState(descriptor);
        snapshotStates = new SimpleVersionedListState<>(listState, new StarRocksVersionedSerializer());

        String legacyStateName = "buffered-rows"
                + (StringUtils.isNullOrWhitespaceOnly(stateKey) ? "" : "-" + stateKey);
        // old version
        ListStateDescriptor<Map<String, StarRocksSinkBufferEntity>> legacyDescriptor =
                new ListStateDescriptor<>(
                        legacyStateName,
                        TypeInformation.of(new TypeHint<Map<String, StarRocksSinkBufferEntity>>() {
                        }));
        legacyState = functionInitializationContext.getOperatorStateStore().getListState(legacyDescriptor);

        if (functionInitializationContext.isRestored()) {
            for (StarrocksSnapshotState state : snapshotStates.get()) {
                for (Map.Entry<Long, List<StreamLoadSnapshot>> entry : state.getData().entrySet()) {
                    snapshotMap.compute(entry.getKey(), (k, v) -> {
                        if (v == null) {
                            return new ArrayList<>(entry.getValue());
                        }
                        v.addAll(entry.getValue());
                        return v;
                    });
                }
            }

            legacyData = new ArrayList<>();
            for (Map<String, StarRocksSinkBufferEntity> entry : legacyState.get()) {
                legacyData.addAll(entry.values());
            }
            log.info("There are {} items from legacy state", legacyData.size());
        }
    }
The map to store the start time of each checkpoint.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/table/StarRocksDynamicSinkFunctionV2.java