public void start()

in src/main/java/com/aliyun/dms/subscribe/clients/DTSConsumerWithDBMapping.java [29:66]

27 lines of code
5 McCabe index (conditional complexity)


    public void start() {

        //check firstly
        boolean checkResult = check();

        if (!checkResult) {
            log.error("DTS precheck failed, dts consumer exit.");
            throw new CriticalException("DTS precheck failed, dts consumer exit.");
        }

        synchronized (this) {
            initLog4j();
            if (started) {
                throw new IllegalStateException("The client has already been started");
            }

            KafkaRecordFetcher recordFetcher = new KafkaRecordFetcher(consumerContext, toProcessRecords);

            UserRecordGeneratorWithDBMapping userRecordGenerator = new UserRecordGeneratorWithDBMapping(consumerContext, toProcessRecords,
                    defaultUserRecords,
                    (tp, timestamp, offset, metadata) -> recordFetcher.setToCommitCheckpoint(new Checkpoint(tp, timestamp, offset, metadata)));

            //processor
            EtlRecordProcessor etlRecordProcessor = new EtlRecordProcessor(consumerContext, defaultUserRecords, recordListeners);

            List<WorkThread> startStream = startWorker(etlRecordProcessor, userRecordGenerator, recordFetcher);

            while (!consumerContext.isExited()) {
                sleepMS(1000);
            }
            log.info("DTS Consumer: shutting down...");
            for (WorkThread workThread : startStream) {
                workThread.stop();
            }

            started = true;
        }
    }