[KAFKA-14843] fix(connect): include base connector config to return

[jeqo]

Issue: https://issues.apache.org/jira/browse/KAFKA-14843

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[gharris1727]

Hey, thanks for the change!

I think that providing the standard sink and source ConfigDefs is a good idea, since that allows the API to be self-describing. Since the existing connector configuration does not separate the namespace for standard sink and source configurations from plugin-specific ones, I think it's acceptable to merge them together in this endpoint.

I only had some nits regarding repeated code. Thanks in advance for humoring me!

[jeqo]

Sure, thanks @gharris1727! let me know how it looks now.

[jeqo]

Used Set to build the result and avoid duplicates in case the configdef comes with common configs -- which could be the case? Happy to adjust if doesn't make sense.

[gharris1727]

It looks like ConfigKeyInfo uses very strict equality, such that if a connector exports a ConfigKeyInfo with the same name but with some other difference, the Set would consider them distinct. This means that the list and set are largely equivalent here.

It looks like in the validate call, connector config keys with the same name as framework keys override the framework provided key: https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java#L511

We could replicate the same behavior here so that users don't get two distinct keys with the same name.

[gharris1727]

LGTM

[C0urante]

Thanks @jeqo! Two minor comments and then this should be good to go.

[C0urante]

I think we can make this part a bit cleaner with some small tweaks:

1. Instead of instantiating configsMap here, declare a ConfigDef baseConfigDefs and initialize it to null (possibly with a comment about how this contains definitions for properties that are automatically supported by the Connect framework, instead of declared by the specific plugin class)
2. Rename configDefs to pluginConfigDefs (and possibly add a comment about how this contains definitions for properties that are explicitly declared by the plugin class)
3. Set baseConfigDefs to SinkConnectorConfig.configDef() or SourceConnectorConfig.configDef() in the switch statement
4. Combine everything together after the switch statement with something like this:

// Track config properties by name and, if the same property is defined in multiple places, give precedence to the one defined by the plugin class
Map<String, ConfigKey> configsMap = new HashMap<>();
if (baseConfigDefs != null)
    configsMap.putAll(baseConfigDefs.configKeys());
configsMap.putAll(pluginConfigDefs.configKeys());

[jeqo]

Thank you all for your feedback. It's looking good! Let's see what CI says :)

[C0urante]

Thanks @jeqo! I've just noticed one more thing that we may want to change but I promise it's the last thing before we can merge this.

[C0urante]

LGTM, thanks @jeqo!