private Iterator<Tuple2<TopicSubscription, MessageId>> tryMigrateState(OperatorStateStore stateStore)
            throws Exception {
        log.info("restore old state version {}", oldStateVersion);
        PulsarSourceStateSerializer stateSerializer =
                new PulsarSourceStateSerializer(getRuntimeContext().getExecutionConfig());
        // Since stateStore.getUnionListState gets the data of a state point,
        // it can only be registered once and will fail to register again,
        // so it only allows the user to set a version number.
        ListState<?> rawStates = stateStore.getUnionListState(new ListStateDescriptor<>(
                OFFSETS_STATE_NAME,
                stateSerializer.getSerializer(oldStateVersion)));

        ListState<String> oldUnionSubscriptionNameStates =
                stateStore.getUnionListState(
                        new ListStateDescriptor<>(
                                OFFSETS_STATE_NAME + "_subName",
                                TypeInformation.of(new TypeHint<String>() {
                                })));
        final Iterator<String> subNameIterator = oldUnionSubscriptionNameStates.get().iterator();

        Iterator<?> tuple2s = rawStates.get().iterator();
        log.info("restore old state has data {}", tuple2s.hasNext());
        final List<Tuple2<TopicSubscription, MessageId>> records = new ArrayList<>();
        while (tuple2s.hasNext()) {
            final Object next = tuple2s.next();
            Tuple2<TopicSubscription, MessageId> tuple2 = stateSerializer.deserialize(oldStateVersion, next);

            String subName = tuple2.f0.getSubscriptionName();
            if (subNameIterator.hasNext()) {
                subName = subNameIterator.next();
            }
            final TopicSubscription topicSubscription = TopicSubscription.builder()
                    .topic(tuple2.f0.getTopic())
                    .range(tuple2.f0.getRange())
                    .subscriptionName(subName)
                    .build();
            final Tuple2<TopicSubscription, MessageId> record = Tuple2.of(topicSubscription, tuple2.f1);
            log.info("migrationState {}", record);
            records.add(record);
        }
        rawStates.clear();
        oldUnionSubscriptionNameStates.clear();
        return records.listIterator();
    }
Try to restore the old save point.

@param stateStore state store
@return state data
@throws Exception Type incompatibility, serialization failure
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSource.java