private void validateSourceConnectorExactlyOnceSupport(
            Map<String, String> rawConfig,
            Map<String, ConfigValue> validatedConfig,
            SourceConnector connector) {
        ConfigValue validatedExactlyOnceSupport = validatedConfig.get(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG);
        if (validatedExactlyOnceSupport.errorMessages().isEmpty()) {
            // Should be safe to parse the enum from the user-provided value since it's passed validation so far
            SourceConnectorConfig.ExactlyOnceSupportLevel exactlyOnceSupportLevel =
                    SourceConnectorConfig.ExactlyOnceSupportLevel.fromProperty(Objects.toString(validatedExactlyOnceSupport.value()));
            if (SourceConnectorConfig.ExactlyOnceSupportLevel.REQUIRED.equals(exactlyOnceSupportLevel)) {
                if (!config.exactlyOnceSourceEnabled()) {
                    validatedExactlyOnceSupport.addErrorMessage("This worker does not have exactly-once source support enabled.");
                }

                try {
                    ExactlyOnceSupport exactlyOnceSupport = connector.exactlyOnceSupport(rawConfig);
                    if (!ExactlyOnceSupport.SUPPORTED.equals(exactlyOnceSupport)) {
                        final String validationErrorMessage;
                        // Would do a switch here but that doesn't permit matching on null values
                        if (exactlyOnceSupport == null) {
                            validationErrorMessage = "The connector does not implement the API required for preflight validation of exactly-once "
                                    + "source support. Please consult the documentation for the connector to determine whether it supports exactly-once "
                                    + "semantics, and then consider reconfiguring the connector to use the value \""
                                    + SourceConnectorConfig.ExactlyOnceSupportLevel.REQUESTED
                                    + "\" for this property (which will disable this preflight check and allow the connector to be created).";
                        } else if (ExactlyOnceSupport.UNSUPPORTED.equals(exactlyOnceSupport)) {
                            validationErrorMessage = "The connector does not support exactly-once semantics with the provided configuration.";
                        } else {
                            throw new ConnectException("Unexpected value returned from SourceConnector::exactlyOnceSupport: " + exactlyOnceSupport);
                        }
                        validatedExactlyOnceSupport.addErrorMessage(validationErrorMessage);
                    }
                } catch (Exception e) {
                    log.error("Failed while validating connector support for exactly-once semantics", e);
                    String validationErrorMessage = "An unexpected error occurred during validation";
                    String failureMessage = e.getMessage();
                    if (failureMessage != null && !failureMessage.trim().isEmpty()) {
                        validationErrorMessage += ": " + failureMessage.trim();
                    } else {
                        validationErrorMessage += "; please see the worker logs for more details.";
                    }
                    validatedExactlyOnceSupport.addErrorMessage(validationErrorMessage);
                }
            }
        }
    }
Close resources managed by this herder but which are not explicitly started.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java