private ConsumerResult confirmOffset(ConsumerResult consumeResult) {
        try {
            ConsumerResult confirmResult = messagePullConsumer
                    .confirmConsume(consumeResult.getConfirmContext(), true);
            if (!confirmResult.isSuccess()) {
                if (!(confirmResult.getErrCode() == TErrCodeConstants.BAD_REQUEST
                        || confirmResult.getErrCode() == TErrCodeConstants.NOT_FOUND
                        || confirmResult.getErrCode() == TErrCodeConstants.ALL_PARTITION_FROZEN
                        || confirmResult.getErrCode() == TErrCodeConstants.NO_PARTITION_ASSIGNED
                        || confirmResult.getErrCode() == TErrCodeConstants.ALL_PARTITION_WAITING
                        || confirmResult.getErrCode() == TErrCodeConstants.ALL_PARTITION_INUSE)) {
                    LOG.warn("Could not confirm messages to tubemq (errcode: {},errmsg: {}).",
                            confirmResult.getErrCode(),
                            confirmResult.getErrMsg());
                }
            }
            return confirmResult;
        } catch (Exception e) {
            LOG.error("Confirm tube offset exception.", e);
        }
        return null;
    }
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