@Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        TypeInformation<Tuple2<String, Long>> typeInformation =
                new TupleTypeInfo<>(STRING_TYPE_INFO, LONG_TYPE_INFO);
        ListStateDescriptor<Tuple2<String, Long>> stateDescriptor =
                new ListStateDescriptor<>(TUBE_OFFSET_STATE, typeInformation);

        OperatorStateStore stateStore = context.getOperatorStateStore();
        offsetsState = stateStore.getListState(stateDescriptor);
        currentOffsets = new HashMap<>();
        if (context.isRestored()) {
            for (Tuple2<String, Long> tubeOffset : offsetsState.get()) {
                currentOffsets.put(tubeOffset.f0, tubeOffset.f1);
            }
            LOG.info("Successfully restore the offsets {}.", currentOffsets);
        } else {
            LOG.info("No restore offsets.");
        }
    }
Build a TubeMQ source function

@param masterAddress         the master address of TubeMQ
@param topic                 the topic name
@param streamIdSet                the  topic's filter condition items
@param consumerGroup         the consumer group name
@param deserializationSchema the deserialize schema
@param configuration         the configure
@param sessionKey            the tube session key
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java