@Override
    public void close() throws Exception {

        cancel();

        if (messagePullConsumer != null) {
            try {
                messagePullConsumer.shutdown();
            } catch (Throwable t) {
                LOG.warn("Could not properly shutdown the tubemq pull consumer.", t);
            }
        }

        if (messageSessionFactory != null) {
            try {
                messageSessionFactory.shutdown();
            } catch (Throwable t) {
                LOG.warn("Could not properly shutdown the tubemq session factory.", t);
            }
        }

        super.close();

        LOG.info("Closed the tubemq source.");
    }
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