Merge pull request #63 from rozza/k203

Fixed sink NPE issue when using with confluent connect 6.1.0

Author: RWaltersMA

CHANGELOG.md

@@ -13,7 +13,7 @@
 
 ### Bug Fixes
   - [KAFKA-195](https://jira.mongodb.org/browse/KAFKA-195) Fixed topics.regex sink validation issue for synthetic config property
-
+  - [KAFKA-203](https://jira.mongodb.org/browse/KAFKA-203) Fixed sink NPE issue when using with confluent connect 6.1.0
 
 ## 1.4.0
 

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkConfig.java

@@ -88,7 +88,7 @@ public class MongoSinkConfig extends AbstractConfig {
 
   static final String PROVIDER_CONFIG = "provider";
 
-  static final List<String> IGNORED_CONFIGS = singletonList(TOPIC_OVERRIDE_CONFIG);
+  static final List<String> INVISIBLE_CONFIGS = singletonList(TOPIC_OVERRIDE_CONFIG);
 
   private Map<String, String> originals;
   private final Optional<List<String>> topics;
@@ -195,7 +195,12 @@ private static ConfigDef createConfigDef() {
           @SuppressWarnings("unchecked")
           public Map<String, ConfigValue> validateAll(final Map<String, String> props) {
             Map<String, ConfigValue> results = super.validateAll(props);
-            IGNORED_CONFIGS.forEach(results::remove);
+            INVISIBLE_CONFIGS.forEach(
+                c -> {
+                  if (results.containsKey(c)) {
+                    results.get(c).visible(false);
+                  }
+                });
             // Don't validate child configs if the top level configs are broken
             if (results.values().stream().anyMatch((c) -> !c.errorMessages().isEmpty())) {
               return results;

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkTopicConfig.java

@@ -328,7 +328,7 @@ public String value() {
 
   public static final String ID_FIELD = "_id";
 
-  static final List<String> IGNORED_CONFIGS = singletonList(TOPIC_CONFIG);
+  static final List<String> SYNTHETIC_CONFIGS = singletonList(TOPIC_CONFIG);
 
   private static final List<Consumer<MongoSinkTopicConfig>> INITIALIZERS =
       asList(
@@ -512,7 +512,7 @@ static Map<String, ConfigValue> validateAll(final String topic, final Map<String
         .forEach(
             (k, v) -> {
               String name = topicOverrides.contains(prefix + k) ? prefix + k : k;
-              if (props.containsKey(name) && !IGNORED_CONFIGS.contains(name)) {
+              if (props.containsKey(name) && !SYNTHETIC_CONFIGS.contains(name)) {
                 results.put(
                     name,
                     new ConfigValue(name, v.value(), v.recommendedValues(), v.errorMessages()));
@@ -567,7 +567,7 @@ static Map<String, ConfigValue> validateRegexAll(final Map<String, String> props
         .validateAll(sinkTopicOriginals)
         .forEach(
             (k, v) -> {
-              if (!IGNORED_CONFIGS.contains(k)) {
+              if (!SYNTHETIC_CONFIGS.contains(k)) {
                 results.put(
                     k, new ConfigValue(k, v.value(), v.recommendedValues(), v.errorMessages()));
               }

src/test/java/com/mongodb/kafka/connect/sink/MongoSinkConfigTest.java

@@ -172,9 +172,8 @@ void testValidation() {
 
     Set<String> expectedKeys = new HashSet<>(MongoSinkConfig.CONFIG.configKeys().keySet());
     expectedKeys.addAll(MongoSinkTopicConfig.CONFIG.configKeys().keySet());
-    // Remove ignored configs
-    expectedKeys.removeAll(MongoSinkConfig.IGNORED_CONFIGS);
-    expectedKeys.removeAll(MongoSinkTopicConfig.IGNORED_CONFIGS);
+    // Remove Synthetic configs
+    expectedKeys.removeAll(MongoSinkTopicConfig.SYNTHETIC_CONFIGS);
 
     // Added declared overrides
     expectedKeys.addAll(configMap.keySet());
@@ -196,9 +195,7 @@ void testValidationRegex() {
 
     Set<String> expectedKeys = new HashSet<>(MongoSinkConfig.CONFIG.configKeys().keySet());
     expectedKeys.addAll(MongoSinkTopicConfig.CONFIG.configKeys().keySet());
-    // Remove ignored configs
-    expectedKeys.removeAll(MongoSinkConfig.IGNORED_CONFIGS);
-    expectedKeys.removeAll(MongoSinkTopicConfig.IGNORED_CONFIGS);
+    expectedKeys.removeAll(MongoSinkTopicConfig.SYNTHETIC_CONFIGS);
 
     // Added declared overrides
     expectedKeys.addAll(configMap.keySet());