protected MessageConsumer subscribe(String topic,
            TreeSet<String> filterConds,
            MessageListener messageListener) throws TubeClientException {
        this.checkClientRunning();
        if (TStringUtils.isBlank(topic)) {
            throw new TubeClientException("Parameter error: topic is Blank!");
        }
        if ((filterConds != null) && (!filterConds.isEmpty())) {
            if (filterConds.size() > TBaseConstants.CFG_FLT_MAX_FILTER_ITEM_COUNT) {
                throw new TubeClientException(new StringBuilder(256)
                        .append("Parameter error: Over max allowed filter count, allowed count is ")
                        .append(TBaseConstants.CFG_FLT_MAX_FILTER_ITEM_COUNT).toString());
            }
            for (String filter : filterConds) {
                if (TStringUtils.isBlank(filter)) {
                    throw new TubeClientException(
                            "Parameter error: blank filter value in parameter filterConds!");
                }
                if (filter.length() > TBaseConstants.CFG_FLT_MAX_FILTER_ITEM_LENGTH) {
                    throw new TubeClientException(new StringBuilder(256)
                            .append("Parameter error: over max allowed filter length, allowed length is ")
                            .append(TBaseConstants.CFG_FLT_MAX_FILTER_ITEM_LENGTH).toString());
                }
            }
        }
        if ((messageListener == null) && (!this.isPullConsume)) {
            throw new IllegalArgumentException("Parameter error: null messageListener");
        }
        TopicProcessor topicProcessor = this.consumeSubInfo.getTopicProcessor(topic);
        if (topicProcessor == null) {
            final TopicProcessor oldProcessor =
                    this.consumeSubInfo.putIfAbsentTopicProcessor(topic,
                            new TopicProcessor(messageListener, filterConds));
            if (oldProcessor != null) {
                throw new TubeClientException(new StringBuilder(256).append("Topic=")
                        .append(topic).append(" has been subscribed").toString());
            }
            return this;
        } else {
            throw new TubeClientException(new StringBuilder(256).append("Topic=")
                    .append(topic).append(" has been subscribed").toString());
        }
    }
Subscribe a topic.

@param topic           topic name
@param filterConds     filter conditions
@param messageListener message listener
@return message consumer
@throws TubeClientException
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/BaseMessageConsumer.java