private boolean startTask(
            ConnectorTaskId id,
            Map<String, String> connProps,
            Map<String, String> taskProps,
            ClusterConfigState configState,
            TaskStatus.Listener statusListener,
            TaskBuilder<?, ?> taskBuilder
    ) {
        final WorkerTask<?, ?> workerTask;
        final TaskStatus.Listener taskStatusListener = workerMetricsGroup.wrapStatusListener(statusListener);
        try (LoggingContext loggingContext = LoggingContext.forTask(id)) {
            log.info("Creating task {}", id);

            if (tasks.containsKey(id))
                throw new ConnectException("Task already exists in this worker: " + id);

            connectorStatusMetricsGroup.recordTaskAdded(id);
            String connType = connProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
            ClassLoader connectorLoader = plugins.connectorLoader(connType);

            try (LoaderSwap loaderSwap = plugins.withClassLoader(connectorLoader)) {
                final ConnectorConfig connConfig = new ConnectorConfig(plugins, connProps);

                int maxTasks = connConfig.tasksMax();
                int numTasks = configState.taskCount(id.connector());
                checkTasksMax(id.connector(), numTasks, maxTasks, connConfig.enforceTasksMax());

                final TaskConfig taskConfig = new TaskConfig(taskProps);
                final Class<? extends Task> taskClass = taskConfig.getClass(TaskConfig.TASK_CLASS_CONFIG).asSubclass(Task.class);
                final Task task = plugins.newTask(taskClass);
                log.info("Instantiated task {} with version {} of type {}", id, task.version(), taskClass.getName());

                // By maintaining connector's specific class loader for this thread here, we first
                // search for converters within the connector dependencies.
                // If any of these aren't found, that means the connector didn't configure specific converters,
                // so we should instantiate based upon the worker configuration
                Converter keyConverter = plugins.newConverter(connConfig, WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, ClassLoaderUsage
                                                                                                                           .CURRENT_CLASSLOADER);
                Converter valueConverter = plugins.newConverter(connConfig, WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, ClassLoaderUsage.CURRENT_CLASSLOADER);
                HeaderConverter headerConverter = plugins.newHeaderConverter(connConfig, WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG,
                                                                             ClassLoaderUsage.CURRENT_CLASSLOADER);
                if (keyConverter == null) {
                    keyConverter = plugins.newConverter(config, WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, ClassLoaderUsage.PLUGINS);
                    log.info("Set up the key converter {} for task {} using the worker config", keyConverter.getClass(), id);
                } else {
                    log.info("Set up the key converter {} for task {} using the connector config", keyConverter.getClass(), id);
                }
                if (valueConverter == null) {
                    valueConverter = plugins.newConverter(config, WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, ClassLoaderUsage.PLUGINS);
                    log.info("Set up the value converter {} for task {} using the worker config", valueConverter.getClass(), id);
                } else {
                    log.info("Set up the value converter {} for task {} using the connector config", valueConverter.getClass(), id);
                }
                if (headerConverter == null) {
                    headerConverter = plugins.newHeaderConverter(config, WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG, ClassLoaderUsage
                                                                                                                             .PLUGINS);
                    log.info("Set up the header converter {} for task {} using the worker config", headerConverter.getClass(), id);
                } else {
                    log.info("Set up the header converter {} for task {} using the connector config", headerConverter.getClass(), id);
                }

                workerTask = taskBuilder
                        .withTask(task)
                        .withConnectorConfig(connConfig)
                        .withKeyConverter(keyConverter)
                        .withValueConverter(valueConverter)
                        .withHeaderConverter(headerConverter)
                        .withClassloader(connectorLoader)
                        .build();

                workerTask.initialize(taskConfig);
            } catch (Throwable t) {
                log.error("Failed to start task {}", id, t);
                connectorStatusMetricsGroup.recordTaskRemoved(id);
                taskStatusListener.onFailure(id, t);
                return false;
            }

            WorkerTask<?, ?> existing = tasks.putIfAbsent(id, workerTask);
            if (existing != null)
                throw new ConnectException("Task already exists in this worker: " + id);

            executor.submit(plugins.withClassLoader(connectorLoader, workerTask));
            if (workerTask instanceof WorkerSourceTask) {
                sourceTaskOffsetCommitter.ifPresent(committer -> committer.schedule(id, (WorkerSourceTask) workerTask));
            }
            return true;
        }
    }
Start a task managed by this worker.

@param id the task ID.
@param connProps the connector properties.
@param taskProps the tasks properties.
@param configState the most recent {@link ClusterConfigState} known to the worker
@param statusListener a listener for the runtime status transitions of the task.
@param taskBuilder the {@link TaskBuilder} used to create the {@link WorkerTask} that manages the lifecycle of the task.
@return true if the task started successfully.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java