@Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        deserializationSchema.setCurrentCheckpointId(context.getCheckpointId());
        // Make sure snapshot all tube partitions' offset to avoid inconsistency.
        Map<String, ConsumeOffsetInfo> curConsumedPartitions = messagePullConsumer.getCurConsumedPartitions();
        for (Map.Entry<String, ConsumeOffsetInfo> consumedPartition : curConsumedPartitions.entrySet()) {
            ConsumeOffsetInfo consumeOffsetInfo = consumedPartition.getValue();
            if (!currentOffsets.containsKey(consumeOffsetInfo.getPartitionKey())) {
                LOG.info("snapshot assigned but not consume partition: {}", consumedPartition.getValue());
                currentOffsets.put(consumeOffsetInfo.getPartitionKey(), consumedPartition.getValue().getCurrOffset());
            }
        }
        offsetsState.clear();
        for (Map.Entry<String, Long> entry : currentOffsets.entrySet()) {
            offsetsState.add(new Tuple2<>(entry.getKey(), entry.getValue()));
        }

        LOG.info("Successfully save the offsets in checkpoint {}: {}.",
                context.getCheckpointId(), currentOffsets);
    }
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.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/FlinkTubeMQConsumer.java