private void validateSourceConnectorTransactionBoundary(
            Map<String, String> rawConfig,
            Map<String, ConfigValue> validatedConfig,
            SourceConnector connector) {
        ConfigValue validatedTransactionBoundary = validatedConfig.get(SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG);
        if (validatedTransactionBoundary.errorMessages().isEmpty()) {
            // Should be safe to parse the enum from the user-provided value since it's passed validation so far
            SourceTask.TransactionBoundary transactionBoundary =
                    SourceTask.TransactionBoundary.fromProperty(Objects.toString(validatedTransactionBoundary.value()));
            if (SourceTask.TransactionBoundary.CONNECTOR.equals(transactionBoundary)) {
                try {
                    ConnectorTransactionBoundaries connectorTransactionSupport = connector.canDefineTransactionBoundaries(rawConfig);
                    if (connectorTransactionSupport == null) {
                        validatedTransactionBoundary.addErrorMessage(
                                "This connector has returned a null value from its canDefineTransactionBoundaries method, which is not permitted. " +
                                        "The connector will be treated as if it cannot define its own transaction boundaries, and cannot be configured with " +
                                        "'" + SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG + "' set to '" + SourceTask.TransactionBoundary.CONNECTOR + "'."
                        );
                    } else if (!ConnectorTransactionBoundaries.SUPPORTED.equals(connectorTransactionSupport)) {
                        validatedTransactionBoundary.addErrorMessage(
                                "The connector does not support connector-defined transaction boundaries with the given configuration. "
                                        + "Please reconfigure it to use a different transaction boundary definition.");
                    }
                } catch (Exception e) {
                    log.error("Failed while validating connector support for defining its own transaction boundaries", 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.";
                    }
                    validatedTransactionBoundary.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