Logging incompatible properties no longer NPEs on null values (#147)

KAFKA-390

Co-authored-by: Valentin Kovalenko <valentin.kovalenko@mongodb.com>

Author: rozza

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## Changelog
 
+## 1.11.1
+
+### Bug Fixes
+- [KAFKA-390](https://jira.mongodb.org/browse/KAFKA-390) Logging incompatible properties no longer NPEs on null values
+
 ## 1.11.0
 
 ### Improvements

src/main/java/com/mongodb/kafka/connect/util/config/ConfigSoftValidator.java

@@ -74,18 +74,19 @@ public static void logIncompatibleProperties(
       final Consumer<String> logger) {
     // global props are considered as belonging to a topic named "" for simplicity
     String global = "";
-    Map<String, Map<String, String>> topicNameToItsStrippedProps =
+    Map<String, Map<String, Optional<String>>> topicNameToItsStrippedProps =
         props.entrySet().stream()
             .collect(
                 Collectors.groupingBy(
                     propNameAndValue -> topicNameFromPropertyName(propNameAndValue.getKey()),
                     Collectors.toMap(
                         propNameAndValue -> strippedPropertyName(propNameAndValue.getKey()),
-                        Entry::getValue)));
-    Map<String, String> globalProps =
+                        e -> Optional.ofNullable(e.getValue()))));
+
+    Map<String, Optional<String>> globalProps =
         topicNameToItsStrippedProps.getOrDefault(global, Collections.emptyMap());
     topicNameToItsStrippedProps.remove(global);
-    Map<String, Map<String, Entry<String, Boolean>>> topicNameToCombinedStrippedProps =
+    Map<String, Map<String, Entry<Optional<String>, Boolean>>> topicNameToCombinedStrippedProps =
         topicNameToItsStrippedProps.entrySet().stream()
             .collect(
                 Collectors.toMap(
@@ -95,7 +96,8 @@ public static void logIncompatibleProperties(
                             globalProps,
                             topicNameAndItsStrippedProps.getKey(),
                             topicNameAndItsStrippedProps.getValue())));
-    Map<String, Entry<String, Boolean>> globalPropsWithFalseFlags =
+
+    Map<String, Entry<Optional<String>, Boolean>> globalPropsWithFalseFlags =
         combineProperties(globalProps, null, null);
     incompatibleConfigs.forEach(
         incompatiblePair -> {
@@ -117,12 +119,12 @@ public static void logIncompatibleProperties(
    *     property value and a flag telling whether the property value came from {@code
    *     topicStrippedProps}, i.e., was overridden, or from {@code globalProps}.
    */
-  private static Map<String, Entry<String, Boolean>> combineProperties(
-      final Map<String, String> globalProps,
+  private static Map<String, Entry<Optional<String>, Boolean>> combineProperties(
+      final Map<String, Optional<String>> globalProps,
       @Nullable final String topicName,
-      @Nullable final Map<String, String> topicStrippedProps) {
+      @Nullable final Map<String, Optional<String>> topicStrippedProps) {
     assertTrue((topicName == null) ^ (topicStrippedProps != null));
-    Map<String, Entry<String, Boolean>> combinedStrippedProps = new HashMap<>();
+    Map<String, Entry<Optional<String>, Boolean>> combinedStrippedProps = new HashMap<>();
     globalProps.forEach(
         (propertyName, propertyValue) ->
             combinedStrippedProps.put(
@@ -273,16 +275,22 @@ public static IncompatiblePropertiesPair latterIgnored(
      */
     private void logIfPresent(
         @Nullable final String topicName,
-        final Map<String, Entry<String, Boolean>> combinedStrippedProps,
+        final Map<String, Entry<Optional<String>, Boolean>> combinedStrippedProps,
         final Consumer<String> logger) {
-      Entry<String, Boolean> property1ValueAndOverridden = combinedStrippedProps.get(propertyName1);
+      Entry<Optional<String>, Boolean> property1ValueAndOverridden =
+          combinedStrippedProps.get(propertyName1);
       if (property1ValueAndOverridden == null
-          || property1ValueAndOverridden.getKey().equals(defaultPropertyValue1)) {
+          || property1ValueAndOverridden
+              .getKey()
+              .equals(Optional.ofNullable(defaultPropertyValue1))) {
         return;
       }
-      Entry<String, Boolean> property2ValueAndOverridden = combinedStrippedProps.get(propertyName2);
+      Entry<Optional<String>, Boolean> property2ValueAndOverridden =
+          combinedStrippedProps.get(propertyName2);
       if (property2ValueAndOverridden == null
-          || property2ValueAndOverridden.getKey().equals(defaultPropertyValue2)) {
+          || property2ValueAndOverridden
+              .getKey()
+              .equals(Optional.ofNullable(defaultPropertyValue2))) {
         return;
       }
       logger.accept(

src/test/java/com/mongodb/kafka/connect/util/config/ConfigSoftValidatorTest.java

@@ -87,11 +87,12 @@ void logIncompatibleProperties() {
             .collect(Collectors.toSet());
     Map<String, String> props = new HashMap<>();
     props.put("a", "valueA");
-    props.put("b", "valueB");
+    props.put("b", null);
     props.put("c", "");
     props.put(topicOverridePropertyName("t", "c"), "valueC");
-    props.put("d", "valueD");
+    props.put("d", null);
     props.put("e", "valueE");
+    props.put(topicOverridePropertyName("t", "e"), null);
     props.put("f", "valueF");
     props.put("g", "valueG");
     Set<String> expectedMessages =
@@ -101,6 +102,7 @@ void logIncompatibleProperties() {
                 latterIgnoredMsg("c", "b"),
                 latterIgnoredMsg(topicOverridePropertyName("t", "c"), "b"),
                 latterIgnoredMsg("d", "e"),
+                latterIgnoredMsg("d", topicOverridePropertyName("t", "e")),
                 latterIgnoredMsg("d", "f"))
             .collect(toSet());
     Set<String> actualMessages = new HashSet<>();