ConfigInfos validateConnectorConfig(
            Map<String, String> connectorProps,
            Function<String, TemporaryStage> reportStage,
            boolean doLog
    ) {
        String stageDescription;
        if (worker.configTransformer() != null) {
            stageDescription = "resolving transformed configuration properties for the connector";
            try (TemporaryStage stage = reportStage.apply(stageDescription)) {
                connectorProps = worker.configTransformer().transform(connectorProps);
            }
        }
        String connType = connectorProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
        if (connType == null)
            throw new BadRequestException("Connector config " + connectorProps + " contains no connector type");

        Connector connector = getConnector(connType);
        ClassLoader connectorLoader = plugins().connectorLoader(connType);
        try (LoaderSwap loaderSwap = plugins().withClassLoader(connectorLoader)) {
            org.apache.kafka.connect.health.ConnectorType connectorType;
            ConfigDef enrichedConfigDef;
            Map<String, ConfigValue> validatedConnectorConfig;
            if (connector instanceof SourceConnector) {
                connectorType = org.apache.kafka.connect.health.ConnectorType.SOURCE;
                enrichedConfigDef = ConnectorConfig.enrich(plugins(), SourceConnectorConfig.configDef(), connectorProps, false);
                stageDescription = "validating source connector-specific properties for the connector";
                try (TemporaryStage stage = reportStage.apply(stageDescription)) {
                    validatedConnectorConfig = validateSourceConnectorConfig((SourceConnector) connector, enrichedConfigDef, connectorProps);
                }
            } else {
                connectorType = org.apache.kafka.connect.health.ConnectorType.SINK;
                enrichedConfigDef = ConnectorConfig.enrich(plugins(), SinkConnectorConfig.configDef(), connectorProps, false);
                stageDescription = "validating sink connector-specific properties for the connector";
                try (TemporaryStage stage = reportStage.apply(stageDescription)) {
                    validatedConnectorConfig = validateSinkConnectorConfig((SinkConnector) connector, enrichedConfigDef, connectorProps);
                }
            }

            connectorProps.entrySet().stream()
                .filter(e -> e.getValue() == null)
                .map(Map.Entry::getKey)
                .forEach(prop ->
                    validatedConnectorConfig.computeIfAbsent(prop, ConfigValue::new)
                        .addErrorMessage("Null value can not be supplied as the configuration value.")
            );

            List<ConfigValue> configValues = new ArrayList<>(validatedConnectorConfig.values());
            Map<String, ConfigKey> configKeys = new LinkedHashMap<>(enrichedConfigDef.configKeys());
            Set<String> allGroups = new LinkedHashSet<>(enrichedConfigDef.groups());

            // do custom connector-specific validation
            ConfigDef configDef;
            stageDescription = "retrieving the configuration definition from the connector";
            try (TemporaryStage stage = reportStage.apply(stageDescription)) {
                configDef = connector.config();
            }
            if (null == configDef) {
                throw new BadRequestException(
                        String.format(
                                "%s.config() must return a ConfigDef that is not null.",
                                connector.getClass().getName()
                        )
                );
            }

            Config config;
            stageDescription = "performing multi-property validation for the connector";
            try (TemporaryStage stage = reportStage.apply(stageDescription)) {
                config = connector.validate(connectorProps);
            }
            if (null == config) {
                throw new BadRequestException(
                        String.format(
                                "%s.validate() must return a Config that is not null.",
                                connector.getClass().getName()
                        )
                );
            }
            configKeys.putAll(configDef.configKeys());
            allGroups.addAll(configDef.groups());
            configValues.addAll(config.configValues());

            // do custom converter-specific validation
            ConfigInfos headerConverterConfigInfos = validateHeaderConverterConfig(
                    connectorProps,
                    validatedConnectorConfig.get(HEADER_CONVERTER_CLASS_CONFIG),
                    reportStage
            );
            ConfigInfos keyConverterConfigInfos = validateKeyConverterConfig(
                    connectorProps,
                    validatedConnectorConfig.get(KEY_CONVERTER_CLASS_CONFIG),
                    reportStage
            );
            ConfigInfos valueConverterConfigInfos = validateValueConverterConfig(
                    connectorProps,
                    validatedConnectorConfig.get(VALUE_CONVERTER_CLASS_CONFIG),
                    reportStage
            );

            ConfigInfos configInfos = generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups));
            AbstractConfig connectorConfig = new AbstractConfig(new ConfigDef(), connectorProps, doLog);
            String connName = connectorProps.get(ConnectorConfig.NAME_CONFIG);
            ConfigInfos producerConfigInfos = null;
            ConfigInfos consumerConfigInfos = null;
            ConfigInfos adminConfigInfos = null;

            if (connectorUsesProducer(connectorType, connectorProps)) {
                stageDescription = "validating producer config overrides for the connector";
                try (TemporaryStage stage = reportStage.apply(stageDescription)) {
                    producerConfigInfos = validateClientOverrides(
                            connName,
                            ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX,
                            connectorConfig,
                            ProducerConfig.configDef(),
                            connector.getClass(),
                            connectorType,
                            ConnectorClientConfigRequest.ClientType.PRODUCER,
                            connectorClientConfigOverridePolicy);
                }
            }
            if (connectorUsesAdmin(connectorType, connectorProps)) {
                stageDescription = "validating admin config overrides for the connector";
                try (TemporaryStage stage = reportStage.apply(stageDescription)) {
                    adminConfigInfos = validateClientOverrides(
                            connName,
                            ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX,
                            connectorConfig,
                            AdminClientConfig.configDef(),
                            connector.getClass(),
                            connectorType,
                            ConnectorClientConfigRequest.ClientType.ADMIN,
                            connectorClientConfigOverridePolicy);
                }
            }
            if (connectorUsesConsumer(connectorType, connectorProps)) {
                stageDescription = "validating consumer config overrides for the connector";
                try (TemporaryStage stage = reportStage.apply(stageDescription)) {
                    consumerConfigInfos = validateClientOverrides(
                            connName,
                            ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX,
                            connectorConfig,
                            ConsumerConfig.configDef(),
                            connector.getClass(),
                            connectorType,
                            ConnectorClientConfigRequest.ClientType.CONSUMER,
                            connectorClientConfigOverridePolicy);
                }
            }
            return mergeConfigInfos(connType,
                    configInfos,
                    producerConfigInfos,
                    consumerConfigInfos,
                    adminConfigInfos,
                    headerConverterConfigInfos,
                    keyConverterConfigInfos,
                    valueConverterConfigInfos
            );
        }
    }
Build the {@link RestartPlan} that describes what should and should not be restarted given the restart request
and the current status of the connector and task instances.

@param request the restart request; may not be null
@return the restart plan, or empty if this worker has no status for the connector named in the request and therefore the
        connector cannot be restarted
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java