public Connect<H> startConnect(Map<String, String> workerProps) {
        log.info("Kafka Connect worker initializing ...");
        long initStart = time.hiResClockMs();

        WorkerInfo initInfo = new WorkerInfo();
        initInfo.logAll();

        log.info("Scanning for plugin classes. This might take a moment ...");
        Plugins plugins = new Plugins(workerProps);
        plugins.compareAndSwapWithDelegatingLoader();
        T config = createConfig(workerProps);
        log.debug("Kafka cluster ID: {}", config.kafkaClusterId());

        RestClient restClient = new RestClient(config);

        ConnectRestServer restServer = new ConnectRestServer(config.rebalanceTimeout(), restClient, config.originals());
        restServer.initializeServer();

        URI advertisedUrl = restServer.advertisedUrl();
        String workerId = advertisedUrl.getHost() + ":" + advertisedUrl.getPort();

        ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy = plugins.newPlugin(
                config.getString(WorkerConfig.CONNECTOR_CLIENT_POLICY_CLASS_CONFIG),
                config, ConnectorClientConfigOverridePolicy.class);

        H herder = createHerder(config, workerId, plugins, connectorClientConfigOverridePolicy, restServer, restClient);

        final Connect<H> connect = new Connect<>(herder, restServer);
        log.info("Kafka Connect worker initialization took {}ms", time.hiResClockMs() - initStart);
        try {
            connect.start();
        } catch (Exception e) {
            log.error("Failed to start Connect", e);
            connect.stop();
            Exit.exit(3);
        }

        return connect;
    }
Initialize and start an instance of {@link Connect}

@param workerProps the worker properties map used to initialize the {@link WorkerConfig}
@return a started instance of {@link Connect}
connect/runtime/src/main/java/org/apache/kafka/connect/cli/AbstractConnectCli.java