@Override
    public void start() {
        serverUrl = mqConfig.getKafkaServerUrl();
        topic = mqConfig.getKafkaTopic();
        boolean isAutoCommit = Boolean.getBoolean(mqConfig.getEnableAutoCommit());
        Preconditions.checkArgument(StringUtils.isNotEmpty(serverUrl), "no kafka server url specified");
        Preconditions.checkArgument(StringUtils.isNotEmpty(mqConfig.getKafkaTopic()),
                "no kafka topic topic specified");
        Preconditions.checkArgument(StringUtils.isNotEmpty(mqConfig.getKafkaConsumerName()),
                "no kafka consume name specified");

        // create topic if need
        createTopic();

        initConsumer(mqConfig);

        Thread thread = new Thread(new Fetcher(consumer, topic, isAutoCommit, mqConfig.getFetchWaitMs()),
                "KafkaConsume_Fetcher_Thread");
        thread.start();
    }
Constructor

@param insertServiceList
@param storeConfig
@param mqConfig
inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/service/consume/KafkaConsume.java