ConfigDef getConfigDefFromConfigProvidingClass(String key, Class<?> cls) {
            if (cls == null) {
                throw new ConfigException(key, null, "Not a " + baseClass.getSimpleName());
            }
            Utils.ensureConcreteSubclass(baseClass, cls);

            T pluginInstance;
            try {
                pluginInstance = Utils.newInstance(cls, baseClass);
            } catch (Exception e) {
                throw new ConfigException(key, String.valueOf(cls), "Error getting config definition from " + baseClass.getSimpleName() + ": " + e.getMessage());
            }
            ConfigDef configDef = config(pluginInstance);
            if (null == configDef) {
                throw new ConnectException(
                    String.format(
                        "%s.config() must return a ConfigDef that is not null.",
                        cls.getName()
                    )
                );
            }
            return configDef;
        }
Return {@link ConfigDef} from {@code cls}, which is expected to be a non-null {@code Class<T>},
by instantiating it and invoking {@link #config(T)}.
@param key
@param cls The subclass of the baseclass.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java