public HeaderConverter newHeaderConverter(AbstractConfig config, String classPropertyName, ClassLoaderUsage classLoaderUsage) {
        Class<? extends HeaderConverter> klass = null;
        switch (classLoaderUsage) {
            case CURRENT_CLASSLOADER:
                if (!config.originals().containsKey(classPropertyName)) {
                    // This connector configuration does not define the header converter via the specified property name
                    return null;
                }
                // Attempt to load first with the current classloader, and plugins as a fallback.
                // Note: we can't use config.getConfiguredInstance because we have to remove the property prefixes
                // before calling config(...)
                klass = pluginClassFromConfig(config, classPropertyName, HeaderConverter.class, scanResult.headerConverters());
                break;
            case PLUGINS:
                // Attempt to load with the plugin class loader, which uses the current classloader as a fallback.
                // Note that there will always be at least a default header converter for the worker
                String converterClassOrAlias = config.getClass(classPropertyName).getName();
                try {
                    klass = pluginClass(
                            delegatingLoader,
                            converterClassOrAlias,
                            HeaderConverter.class
                    );
                } catch (ClassNotFoundException e) {
                    throw new ConnectException(
                            "Failed to find any class that implements HeaderConverter and which name matches "
                                    + converterClassOrAlias
                                    + ", available header converters are: "
                                    + pluginNames(scanResult.headerConverters())
                    );
                }
        }
        if (klass == null) {
            throw new ConnectException("Unable to initialize the HeaderConverter specified in '" + classPropertyName + "'");
        }

        String configPrefix = classPropertyName + ".";
        Map<String, Object> converterConfig = config.originalsWithPrefix(configPrefix);
        converterConfig.put(ConverterConfig.TYPE_CONFIG, ConverterType.HEADER.getName());
        log.debug("Configuring the header converter with configuration keys:{}{}", System.lineSeparator(), converterConfig.keySet());

        HeaderConverter plugin;
        try (LoaderSwap loaderSwap = withClassLoader(klass.getClassLoader())) {
            plugin = newPlugin(klass);
            plugin.configure(converterConfig);
        }
        return plugin;
    }
If the given configuration defines a {@link HeaderConverter} using the named configuration property, return a new configured
instance.

@param config             the configuration containing the {@link Converter}'s configuration; may not be null
@param classPropertyName  the name of the property that contains the name of the {@link Converter} class; may not be null
@param classLoaderUsage   which classloader should be used
@return the instantiated and configured {@link HeaderConverter}; null if the configuration did not define the specified property
@throws ConnectException if the {@link HeaderConverter} implementation class could not be found
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java