private boolean startTask(ConnectorTaskId taskId) {
        log.info("Starting task {}", taskId);
        Map<String, String> connProps = configState.connectorConfig(taskId.connector());
        switch (connectorType(connProps)) {
            case SINK:
                return worker.startSinkTask(
                        taskId,
                        configState,
                        connProps,
                        configState.taskConfig(taskId),
                        this,
                        configState.targetState(taskId.connector())
                );
            case SOURCE:
                if (config.exactlyOnceSourceEnabled()) {
                    int taskGeneration = configState.taskConfigGeneration(taskId.connector());
                    return worker.startExactlyOnceSourceTask(
                            taskId,
                            configState,
                            connProps,
                            configState.taskConfig(taskId),
                            this,
                            configState.targetState(taskId.connector()),
                            () -> {
                                FutureCallback<Void> preflightFencing = new FutureCallback<>();
                                fenceZombieSourceTasks(taskId, preflightFencing);
                                try {
                                    preflightFencing.get();
                                } catch (InterruptedException e) {
                                    throw new ConnectException("Interrupted while attempting to perform round of zombie fencing", e);
                                } catch (ExecutionException e) {
                                    Throwable cause = e.getCause();
                                    throw ConnectUtils.maybeWrap(cause, "Failed to perform round of zombie fencing");
                                }
                            },
                            () -> verifyTaskGenerationAndOwnership(taskId, taskGeneration)
                    );
                } else {
                    return worker.startSourceTask(
                            taskId,
                            configState,
                            connProps,
                            configState.taskConfig(taskId),
                            this,
                            configState.targetState(taskId.connector())
                    );
                }
            default:
                throw new ConnectException("Failed to start task " + taskId + " since it is not a recognizable type (source or sink)");
        }
    }
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