public void start() {
        log.info("Kafka MirrorMaker starting with {} herders.", herders.size());
        if (startLatch != null) {
            throw new IllegalStateException("MirrorMaker instance already started");
        }
        startLatch = new CountDownLatch(herders.size());
        stopLatch = new CountDownLatch(herders.size());
        Exit.addShutdownHook("mirror-maker-shutdown-hook", shutdownHook);
        for (Herder herder : herders.values()) {
            try {
                herder.start();
            } finally {
                startLatch.countDown();
            }
        }
        if (internalServer != null) {
            log.info("Initializing internal REST resources");
            internalServer.initializeInternalResources(herders);
        }
        log.info("Configuring connectors will happen once the worker joins the group as a leader");
        log.info("Kafka MirrorMaker started");
    }
@param config    MM2 configuration from mm2.properties file
@param clusters  target clusters for this node. These must match cluster
                 aliases as defined in the config. If null or empty list,
                 uses all clusters in the config.
@param time      time source
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMaker.java