@Override
    public void open(Configuration parameters) throws Exception {

        deserializationSchema.open(null);
        ConsumerConfig consumerConfig = new ConsumerConfig(masterAddress, consumerGroup);
        consumerConfig.setConsumePosition(consumeFromMax
                ? ConsumePosition.CONSUMER_FROM_LATEST_OFFSET
                : ConsumePosition.CONSUMER_FROM_FIRST_OFFSET);

        consumerConfig.setMsgNotFoundWaitPeriodMs(messageNotFoundWaitPeriod.toMillis());

        final int numTasks = getRuntimeContext().getNumberOfParallelSubtasks();
        messageSessionFactory = new TubeSingleSessionFactory(consumerConfig);
        messagePullConsumer = messageSessionFactory.createPullConsumer(consumerConfig);
        messagePullConsumer.subscribe(topic, streamIdSet);
        String jobId = getRuntimeContext().getJobId().toString();
        String attemptNumber = String.valueOf(getRuntimeContext().getAttemptNumber());
        String startSessionKey = sessionKey.concat(UNDERSCORE).concat(jobId).concat(UNDERSCORE).concat(attemptNumber);
        LOG.info("start to init tube mq consumer, session key={}", startSessionKey);
        messagePullConsumer.completeSubscribe(startSessionKey, numTasks, true, currentOffsets);

        running = true;
    }
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