@Override
    public void startConnect() {
        log.info("Starting standalone Connect worker");

        workerProps.put(BOOTSTRAP_SERVERS_CONFIG, kafka().bootstrapServers());
        // use a random available port
        workerProps.put(LISTENERS_CONFIG, "HTTP://" + REST_HOST_NAME + ":0");

        workerProps.putIfAbsent(OFFSET_STORAGE_FILE_FILENAME_CONFIG, offsetsFile);
        workerProps.putIfAbsent(KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
        workerProps.putIfAbsent(VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
        workerProps.putIfAbsent(PLUGIN_DISCOVERY_CONFIG, "hybrid_fail");

        ConnectStandalone cli = new ConnectStandalone();
        Connect<StandaloneHerder> connect = cli.startConnect(workerProps);
        connectWorker = new WorkerHandle("standalone", connect);
        cli.processExtraArgs(connect, connectorConfigFiles());
    }
Start a standalone embedded connect worker. Internally, this class will spin up a Kafka cluster,
set up any tmp directories. and clean them up on exit. Methods on the same
{@code EmbeddedConnectStandalone} are not guaranteed to be thread-safe.
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectStandalone.java