public void startConnector(
            String connName,
            Map<String, String> connProps,
            CloseableConnectorContext ctx,
            ConnectorStatus.Listener statusListener,
            TargetState initialState,
            Callback<TargetState> onConnectorStateChange
    ) {
        final ConnectorStatus.Listener connectorStatusListener = workerMetricsGroup.wrapStatusListener(statusListener);
        try (LoggingContext loggingContext = LoggingContext.forConnector(connName)) {
            if (connectors.containsKey(connName)) {
                onConnectorStateChange.onCompletion(
                        new ConnectException("Connector with name " + connName + " already exists"),
                        null);
                return;
            }

            final WorkerConnector workerConnector;
            final String connClass = connProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
            ClassLoader connectorLoader = plugins.connectorLoader(connClass);
            try (LoaderSwap loaderSwap = plugins.withClassLoader(connectorLoader)) {
                log.info("Creating connector {} of type {}", connName, connClass);
                final Connector connector = plugins.newConnector(connClass);
                final ConnectorConfig connConfig;
                final CloseableOffsetStorageReader offsetReader;
                final ConnectorOffsetBackingStore offsetStore;
                if (ConnectUtils.isSinkConnector(connector)) {
                    connConfig = new SinkConnectorConfig(plugins, connProps);
                    offsetReader = null;
                    offsetStore = null;
                } else {
                    SourceConnectorConfig sourceConfig = new SourceConnectorConfig(plugins, connProps, config.topicCreationEnable());
                    connConfig = sourceConfig;

                    // Set up the offset backing store for this connector instance
                    offsetStore = config.exactlyOnceSourceEnabled()
                            ? offsetStoreForExactlyOnceSourceConnector(sourceConfig, connName, connector, null)
                            : offsetStoreForRegularSourceConnector(sourceConfig, connName, connector, null);
                    offsetStore.configure(config);
                    offsetReader = new OffsetStorageReaderImpl(offsetStore, connName, internalKeyConverter, internalValueConverter);
                }
                workerConnector = new WorkerConnector(
                        connName, connector, connConfig, ctx, metrics, connectorStatusListener, offsetReader, offsetStore, connectorLoader);
                log.info("Instantiated connector {} with version {} of type {}", connName, connector.version(), connector.getClass());
                workerConnector.transitionTo(initialState, onConnectorStateChange);
            } catch (Throwable t) {
                log.error("Failed to start connector {}", connName, t);
                connectorStatusListener.onFailure(connName, t);
                onConnectorStateChange.onCompletion(t, null);
                return;
            }

            WorkerConnector existing = connectors.putIfAbsent(connName, workerConnector);
            if (existing != null) {
                onConnectorStateChange.onCompletion(
                        new ConnectException("Connector with name " + connName + " already exists"),
                        null);
                // Don't need to do any cleanup of the WorkerConnector instance (such as calling
                // shutdown() on it) here because it hasn't actually started running yet
                return;
            }

            executor.submit(plugins.withClassLoader(connectorLoader, workerConnector));

            log.info("Finished creating connector {}", connName);
        }
    }
Start a connector managed by this worker.

@param connName the connector name.
@param connProps the properties of the connector.
@param ctx the connector runtime context.
@param statusListener a listener for the runtime status transitions of the connector.
@param initialState the initial state of the connector.
@param onConnectorStateChange invoked when the initial state change of the connector is completed
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java