private void startConnector(String connectorName, Callback<Void> callback) {
        log.info("Starting connector {}", connectorName);
        final Map<String, String> configProps = configState.connectorConfig(connectorName);
        final CloseableConnectorContext ctx = new HerderConnectorContext(this, connectorName);
        final TargetState initialState = configState.targetState(connectorName);
        final Callback<TargetState> onInitialStateChange = (error, newState) -> {
            if (error != null) {
                callback.onCompletion(new ConnectException("Failed to start connector: " + connectorName, error), null);
                return;
            }

            // Use newState here in case the connector has been paused right after being created
            if (newState == TargetState.STARTED) {
                addRequest(
                    () -> {
                        // Request configuration since this could be a brand new connector. However, also only update those
                        // task configs if they are actually different from the existing ones to avoid unnecessary updates when this is
                        // just restoring an existing connector.
                        reconfigureConnectorTasksWithRetry(time.milliseconds(), connectorName);
                        callback.onCompletion(null, null);
                        return null;
                    },
                    forwardErrorAndTickThreadStages(callback)
                );
            } else {
                callback.onCompletion(null, null);
            }
        };
        callback.recordStage(new Stage("starting the connector", time.milliseconds()));
        worker.startConnector(connectorName, configProps, ctx, this, initialState, onInitialStateChange);
    }
Try to read to the end of the config log within the given timeout and capture a fresh snapshot via
{@link ConfigBackingStore#snapshot()}

@param timeoutMs maximum time to wait to sync to the end of the log
@return true if successful; false if timed out
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java