@Override
    public void run(SourceContext<T> ctx) throws Exception {

        Instant lastConsumeInstant = Instant.now();

        while (running) {
            ConsumerResult consumeResult = messagePullConsumer.getMessage();
            if (!consumeResult.isSuccess()) {
                if (!(consumeResult.getErrCode() == TErrCodeConstants.BAD_REQUEST
                        || consumeResult.getErrCode() == TErrCodeConstants.NOT_FOUND
                        || consumeResult.getErrCode() == TErrCodeConstants.ALL_PARTITION_FROZEN
                        || consumeResult.getErrCode() == TErrCodeConstants.NO_PARTITION_ASSIGNED
                        || consumeResult.getErrCode() == TErrCodeConstants.ALL_PARTITION_WAITING
                        || consumeResult.getErrCode() == TErrCodeConstants.ALL_PARTITION_INUSE)) {
                    LOG.info("Could not consume messages from tubemq (errcode: {},errmsg: {}).",
                            consumeResult.getErrCode(),
                            consumeResult.getErrMsg());
                }

                Duration idleTime = Duration.between(lastConsumeInstant, Instant.now());
                if (idleTime.compareTo(maxIdleTime) > 0) {
                    ctx.markAsTemporarilyIdle();
                }

                continue;
            }

            List<Message> messageList = consumeResult.getMessageList();
            lastConsumeInstant = Instant.now();

            List<T> records = new ArrayList<>();

            synchronized (ctx.getCheckpointLock()) {
                lastConsumeInstant = getRecords(lastConsumeInstant, messageList, records);
                for (T record : records) {
                    ctx.collect(record);
                }

                ConsumerResult confirmResult = confirmOffset(consumeResult);
                if (confirmResult != null && confirmResult.isSuccess()) {
                    currentOffsets.put(confirmResult.getPartitionKey(), confirmResult.getCurrOffset());
                } else {
                    LOG.warn("Confirm offset failed, fallback to use offset in consume result.");
                    currentOffsets.put(consumeResult.getPartitionKey(), consumeResult.getCurrOffset());
                }
            }
        }
    }
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