protected void updateConcurrentConsumer(PulsarClient pulsarClient) {
        List<Consumer<byte[]>> list = topicConsumerMap.computeIfAbsent(mqConfig.getPulsarTopic(),
                key -> new ArrayList<Consumer<byte[]>>());
        int currentConsumerNum = list.size();
        int createNum = mqConfig.getConcurrentConsumerNum() - currentConsumerNum;
        /*
         * add consumer
         */
        if (createNum > 0) {
            for (int i = 0; i < mqConfig.getConcurrentConsumerNum(); i++) {
                Consumer<byte[]> consumer = createConsumer(pulsarClient, mqConfig.getPulsarTopic());
                if (consumer != null) {
                    list.add(consumer);
                }
            }
        } else if (createNum < 0) {
            /*
             * delete consumer
             */
            int removeIndex = currentConsumerNum - 1;
            for (int i = createNum; i < 0; i++) {
                Consumer<byte[]> consumer = list.remove(removeIndex);
                consumer.closeAsync();
                removeIndex -= 1;
            }
        }
    }
Constructor
@param insertServiceList
@param storeConfig
@param mqConfig
inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/service/consume/PulsarConsume.java