Add wildcard support for starts with field name matching

KAFKA-391

Author: rozza

CHANGELOG.md

@@ -5,7 +5,8 @@
 ## 1.11.1
 
 ### Bug Fixes
-- [KAFKA-390](https://jira.mongodb.org/browse/KAFKA-390) Logging incompatible properties no longer NPEs on null values
+  - [KAFKA-391](https://jira.mongodb.org/browse/KAFKA-391) Add wildcard support for starts with field name matching
+  - [KAFKA-390](https://jira.mongodb.org/browse/KAFKA-390) Logging incompatible properties no longer NPEs on null values
 
 ## 1.11.0
 

src/main/java/com/mongodb/kafka/connect/sink/processor/field/projection/AllowListProjector.java

@@ -56,6 +56,7 @@ private void doProjection(final String field, final BsonDocument doc) {
           field.isEmpty()
               ? entry.getKey()
               : field + FieldProjector.SUB_FIELD_DOT_SEPARATOR + entry.getKey();
+
       BsonValue value = entry.getValue();
 
       if (!getFields().contains(key) && !checkForWildcardMatch(key)) {
@@ -85,6 +86,10 @@ private void doProjection(final String field, final BsonDocument doc) {
   }
 
   private boolean checkForWildcardMatch(final String key) {
+    if (checkForStartsWithWildcardMatch(key)) {
+      return true;
+    }
+
     String[] keyParts = key.split("\\" + FieldProjector.SUB_FIELD_DOT_SEPARATOR);
     String[] pattern = new String[keyParts.length];
     Arrays.fill(pattern, FieldProjector.SINGLE_WILDCARD);
@@ -105,4 +110,21 @@ private boolean checkForWildcardMatch(final String key) {
 
     return false;
   }
+
+  /**
+   * Processes any non nested field names to see if they should be kept
+   *
+   * @param key the document key to check against
+   * @return true if any of the fields match against the document key
+   */
+  private boolean checkForStartsWithWildcardMatch(final String key) {
+    return getFields().stream()
+        .filter(
+            f ->
+                !f.contains(FieldProjector.SUB_FIELD_DOT_SEPARATOR)
+                    && !f.equals(FieldProjector.SINGLE_WILDCARD)
+                    && f.endsWith(FieldProjector.SINGLE_WILDCARD))
+        .map(f -> f.substring(0, f.length() - 1))
+        .anyMatch(key::startsWith);
+  }
 }

src/main/java/com/mongodb/kafka/connect/sink/processor/field/projection/BlockListProjector.java

@@ -41,13 +41,7 @@ protected void projectDocument(final BsonDocument doc) {
 
   private void doProjection(final String field, final BsonDocument doc) {
     if (!field.contains(FieldProjector.SUB_FIELD_DOT_SEPARATOR)) {
-      if (field.equals(FieldProjector.SINGLE_WILDCARD)
-          || field.equals(FieldProjector.DOUBLE_WILDCARD)) {
-        handleWildcard(field, "", doc);
-        return;
-      }
-
-      doc.remove(field);
+      processNonNestedMatches(field, doc);
       return;
     }
 
@@ -77,6 +71,24 @@ private void doProjection(final String field, final BsonDocument doc) {
     }
   }
 
+  /**
+   * Processes any non nested field names and removes any matches from the document.
+   *
+   * @param field the field name or field name with wildcard
+   * @param doc the document to block fields from
+   */
+  private void processNonNestedMatches(final String field, final BsonDocument doc) {
+    if (field.equals(FieldProjector.SINGLE_WILDCARD)
+        || field.equals(FieldProjector.DOUBLE_WILDCARD)) {
+      handleWildcard(field, "", doc);
+    } else if (field.endsWith(FieldProjector.SINGLE_WILDCARD)) {
+      String fieldStartsWith = field.substring(0, field.length() - 1);
+      doc.entrySet().removeIf(entry -> entry.getKey().startsWith(fieldStartsWith));
+    } else {
+      doc.remove(field);
+    }
+  }
+
   private void handleWildcard(
       final String firstPart, final String otherParts, final BsonDocument doc) {
     Iterator<Map.Entry<String, BsonValue>> iter = doc.entrySet().iterator();

src/test/java/com/mongodb/kafka/connect/sink/processor/field/projection/FieldProjectorTest.java

@@ -89,13 +89,17 @@ static void setupFlatDocMaps() {
         BsonDocument.parse(
             "{myBoolean: true, myInt: 42, "
                 + "myBytes: {$binary: 'QUJD', $type: '00'}, myArray: []}");
+    BsonDocument keyDocument6 = BsonDocument.parse("{_id: 'ABC-123', myInt: 42, myArray: []}");
+    BsonDocument keyDocument7 =
+        BsonDocument.parse("{myBoolean: true, myBytes: {$binary: 'QUJD', $type: '00'}}");
 
     flatKeyFieldsMapBlockList =
         new HashMap<String, BsonDocument>() {
           {
             put("", keyDocument1);
             put("*", keyDocument2);
             put("**", keyDocument2);
+            put("myB*", keyDocument6);
             put("_id", keyDocument5);
             put("myBoolean, myInt", keyDocument3);
             put("missing1, unknown2", keyDocument1);
@@ -108,6 +112,7 @@ static void setupFlatDocMaps() {
             put("", keyDocument2);
             put("*", keyDocument1);
             put("**", keyDocument1);
+            put("myB*", keyDocument7);
             put("missing1, unknown2", keyDocument2);
             put("myBoolean, myBytes, myArray", keyDocument4);
           }
@@ -131,13 +136,15 @@ static void setupFlatDocMaps() {
         BsonDocument.parse(
             "{ myLong: {$numberLong: '42'}, myDouble: 23.23, myString: 'BSON', "
                 + "myBytes: {$binary: 'eHl6', $type: '00'}, myArray: []}");
+    BsonDocument valueDocument6 = BsonDocument.parse("{_id: 'XYZ-789'}");
 
     flatValueFieldsMapBlockList =
         new HashMap<String, BsonDocument>() {
           {
             put("", valueDocument1);
             put("*", valueDocument2);
             put("**", valueDocument2);
+            put("my*", valueDocument6);
             put("_id", valueDocument5);
             put("myLong, myDouble", valueDocument3);
             put("missing1,unknown2", valueDocument1);
@@ -150,6 +157,8 @@ static void setupFlatDocMaps() {
             put("", valueDocument2);
             put("*", valueDocument1);
             put("**", valueDocument1);
+            put("my*", valueDocument5);
+            put("_id", valueDocument6);
             put("missing1,unknown2", valueDocument2);
             put("myDouble, myBytes,myArray", valueDocument4);
           }
@@ -364,7 +373,7 @@ List<DynamicTest> testProjectorSettingsOnNestedStructure() {
                   entry.getKey()));
       tests.add(
           buildDynamicTestFor(
-              entry, buildSinkDocumentNestedStruct(), new WhitelistKeyProjector(cfg)));
+              entry, buildSinkDocumentNestedStruct(), new AllowListKeyProjector(cfg)));
     }
 
     for (Map.Entry<String, BsonDocument> entry : nestedValueFieldsMapBlockList.entrySet()) {
@@ -378,7 +387,7 @@ List<DynamicTest> testProjectorSettingsOnNestedStructure() {
                   entry.getKey()));
       tests.add(
           buildDynamicTestFor(
-              entry, buildSinkDocumentNestedStruct(), new BlacklistValueProjector(cfg)));
+              entry, buildSinkDocumentNestedStruct(), new BlockListValueProjector(cfg)));
     }
 
     for (Map.Entry<String, BsonDocument> entry : nestedValueFieldsMapAllowList.entrySet()) {
@@ -392,7 +401,7 @@ List<DynamicTest> testProjectorSettingsOnNestedStructure() {
                   entry.getKey()));
       tests.add(
           buildDynamicTestFor(
-              entry, buildSinkDocumentNestedStruct(), new WhitelistValueProjector(cfg)));
+              entry, buildSinkDocumentNestedStruct(), new AllowListValueProjector(cfg)));
     }
 
     return tests;