[GitHub] [kafka] C0urante commented on a diff in pull request #13445: [KAFKA-14843] fix(connect): include base connector config to return

["C0urante (via GitHub)" <gi...@apache.org>]

C0urante commented on code in PR #13445:
URL: https://github.com/apache/kafka/pull/13445#discussion_r1149888198


##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java:
##########
@@ -943,35 +943,45 @@ public void testGenerateResultWithConfigValuesWithNoConfigKeysAndWithSomeErrors(
 
     @Test
     public void testSinkConnectorPluginConfig() throws ClassNotFoundException {
-        testConnectorPluginConfig("sink", SampleSinkConnector::new, SampleSinkConnector::config);
+        testConnectorPluginConfig("sink", SampleSinkConnector::new, SampleSinkConnector::config, Optional.of(SinkConnectorConfig.configDef()));
+    }
+
+    @Test
+    public void testSinkConnectorPluginConfigIncludingCommon() throws ClassNotFoundException {
+        testConnectorPluginConfig("sink", SampleSinkConnector::new, SampleSinkConnector::configWithCommon, Optional.empty());
     }
 
     @Test
     public void testSourceConnectorPluginConfig() throws ClassNotFoundException {
-        testConnectorPluginConfig("source", SampleSourceConnector::new, SampleSourceConnector::config);
+        testConnectorPluginConfig("source", SampleSourceConnector::new, SampleSourceConnector::config, Optional.of(SourceConnectorConfig.configDef()));
+    }
+
+    @Test
+    public void testSourceConnectorPluginConfigIncludingCommon() throws ClassNotFoundException {
+        testConnectorPluginConfig("source", SampleSourceConnector::new, SampleSourceConnector::configWithCommon, Optional.empty());
     }
 
     @Test
     public void testConverterPluginConfig() throws ClassNotFoundException {
-        testConnectorPluginConfig("converter", SampleConverterWithHeaders::new, SampleConverterWithHeaders::config);
+        testConnectorPluginConfig("converter", SampleConverterWithHeaders::new, SampleConverterWithHeaders::config, Optional.empty());
     }
 
     @Test
     public void testHeaderConverterPluginConfig() throws ClassNotFoundException {
-        testConnectorPluginConfig("header-converter", SampleHeaderConverter::new, SampleHeaderConverter::config);
+        testConnectorPluginConfig("header-converter", SampleHeaderConverter::new, SampleHeaderConverter::config, Optional.empty());
     }
 
     @Test
     public void testPredicatePluginConfig() throws ClassNotFoundException {
-        testConnectorPluginConfig("predicate", SamplePredicate::new, SamplePredicate::config);
+        testConnectorPluginConfig("predicate", SamplePredicate::new, SamplePredicate::config, Optional.empty());
     }
 
     @Test
     public void testTransformationPluginConfig() throws ClassNotFoundException {
-        testConnectorPluginConfig("transformation", SampleTransformation::new, SampleTransformation::config);
+        testConnectorPluginConfig("transformation", SampleTransformation::new, SampleTransformation::config, Optional.empty());
     }
 
-    private <T> void testConnectorPluginConfig(String pluginName, Supplier<T> newPluginInstance, Function<T, ConfigDef> pluginConfig) throws ClassNotFoundException {
+    private <T> void testConnectorPluginConfig(String pluginName, Supplier<T> newPluginInstance, Function<T, ConfigDef> pluginConfig, Optional<ConfigDef> baseConfig) throws ClassNotFoundException {

Review Comment:
   Nit: this line's getting a bit line, can we break it up?
   ```suggestion
       private <T> void testConnectorPluginConfig(
               String pluginName,
               Supplier<T> newPluginInstance,
               Function<T, ConfigDef> pluginConfig,
               Optional<ConfigDef> baseConfig
       ) throws ClassNotFoundException {
   ```



##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java:
##########
@@ -835,11 +835,16 @@ public List<ConfigKeyInfo> connectorPluginConfig(String pluginName) {
         try (LoaderSwap loaderSwap = p.withClassLoader(pluginClass.getClassLoader())) {
             Object plugin = p.newPlugin(pluginName);
             PluginType pluginType = PluginType.from(plugin.getClass());
+            Map<String, ConfigKey> configsMap = new HashMap<>();

Review Comment:
   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:
   
   ```java
   // 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());
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org