Added AllowList and BlockList field projectors

Deprecated: `Blacklist[Key|Value]Projector`, `Whitelist[Key|Value]Projector`.
Use: `BlockListKeyProjector`, `BlockListValueProjector`,
     `AllowListKeyProjector` or `AllowListValueProjector` instead.

KAFKA-112

Author: rozza

CHANGELOG.md

@@ -11,6 +11,9 @@
   - [KAFKA-118](https://jira.mongodb.org/browse/KAFKA-118) Made UuidStrategy configurable so can output BsonBinary Uuid values
   - [KAFKA-101](https://jira.mongodb.org/browse/KAFKA-101) Added `UuidProvidedInKeyStrategy` & `UuidProvidedInValueStrategy`
   - [KAFKA-114](https://jira.mongodb.org/browse/KAFKA-114) Added `UpdateOneBusinessKeyTimestampStrategy` write model strategy`
+  - [KAFKA-112](https://jira.mongodb.org/browse/KAFKA-112) Added `BlockList` and `AllowList` field projector type configurations and
+    `BlockListKeyProjector`, `BlockListValueProjector`, `AllowListKeyProjector`and `AllowListValueProjector` Post processors.
+    Deprecated: `BlacklistKeyProjector`, `BlacklistValueProjector`, `WhitelistKeyProjector` and `WhitelistValueProjector`.
 
 ## 1.1.0
   - [KAFKA-45](https://jira.mongodb.org/browse/KAFKA-45) Allow the Sink connector to ignore unused source record key or value fields.

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

@@ -12,13 +12,10 @@
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
- *
  */
 
 package com.mongodb.kafka.connect.sink;
 
-import org.apache.kafka.common.config.AbstractConfig;
-
 public interface Configurable {
-  void configure(AbstractConfig configuration);
+  void configure(MongoSinkTopicConfig configuration);
 }

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

@@ -28,7 +28,6 @@
 import static java.util.Collections.singletonList;
 import static org.apache.kafka.common.config.ConfigDef.NO_DEFAULT_VALUE;
 
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -45,16 +44,10 @@
 import com.mongodb.MongoNamespace;
 
 import com.mongodb.kafka.connect.sink.cdc.CdcHandler;
-import com.mongodb.kafka.connect.sink.processor.BlacklistKeyProjector;
-import com.mongodb.kafka.connect.sink.processor.BlacklistValueProjector;
 import com.mongodb.kafka.connect.sink.processor.PostProcessors;
-import com.mongodb.kafka.connect.sink.processor.WhitelistKeyProjector;
-import com.mongodb.kafka.connect.sink.processor.WhitelistValueProjector;
-import com.mongodb.kafka.connect.sink.processor.field.projection.FieldProjector;
 import com.mongodb.kafka.connect.sink.processor.id.strategy.FullKeyStrategy;
 import com.mongodb.kafka.connect.sink.processor.id.strategy.IdStrategy;
 import com.mongodb.kafka.connect.sink.processor.id.strategy.PartialKeyStrategy;
-import com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy;
 import com.mongodb.kafka.connect.sink.processor.id.strategy.ProvidedInKeyStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneDefaultStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.WriteModelStrategy;
@@ -67,7 +60,9 @@ public class MongoSinkTopicConfig extends AbstractConfig {
   public enum FieldProjectionType {
     NONE,
     BLACKLIST,
-    WHITELIST
+    WHITELIST,
+    ALLOWLIST,
+    BLOCKLIST
   }
 
   public enum UuidBsonFormat {
@@ -121,7 +116,7 @@ public enum UuidBsonFormat {
   private static final String DOCUMENT_ID_STRATEGY_UUID_FORMAT_DISPLAY =
       "The document id strategy uuid format";
   private static final String DOCUMENT_ID_STRATEGY_UUID_FORMAT_DOC =
-      "The bson output format for UuidStrategy.";
+      "The bson output format when using the `UuidStrategy`.";
   private static final String DOCUMENT_ID_STRATEGY_UUID_FORMAT_DEFAULT = "string";
 
   public static final String KEY_PROJECTION_TYPE_CONFIG = "key.projection.type";
@@ -278,32 +273,21 @@ public MongoNamespace getNamespace() {
     return namespace;
   }
 
-  @SuppressWarnings("unchecked")
+  private <T> T configureInstance(final T instance) {
+    if (instance instanceof Configurable) {
+      ((Configurable) instance).configure(this);
+    }
+    return instance;
+  }
+
   public IdStrategy getIdStrategy() {
     if (idStrategy == null) {
-      String className = getString(DOCUMENT_ID_STRATEGY_CONFIG);
-      Optional<FieldProjector> fieldProjector = getFieldProjector(className);
       idStrategy =
-          createInstance(
-              DOCUMENT_ID_STRATEGY_CONFIG,
-              className,
-              IdStrategy.class,
-              () -> {
-                Class<IdStrategy> clazz = (Class<IdStrategy>) Class.forName(className);
-                boolean hasFieldProjectorConstructor =
-                    fieldProjector.isPresent()
-                        && Arrays.stream(clazz.getConstructors())
-                            .anyMatch(
-                                i ->
-                                    i.getParameterTypes().length == 1
-                                        && i.getParameterTypes()[0].equals(FieldProjector.class));
-                if (hasFieldProjectorConstructor) {
-                  return clazz
-                      .getConstructor(FieldProjector.class)
-                      .newInstance(fieldProjector.get());
-                }
-                return clazz.getConstructor().newInstance();
-              });
+          configureInstance(
+              createInstance(
+                  DOCUMENT_ID_STRATEGY_CONFIG,
+                  getString(DOCUMENT_ID_STRATEGY_CONFIG),
+                  IdStrategy.class));
     }
     return idStrategy;
   }
@@ -318,10 +302,11 @@ PostProcessors getPostProcessors() {
   WriteModelStrategy getWriteModelStrategy() {
     if (writeModelStrategy == null) {
       writeModelStrategy =
-          createInstance(
-              WRITEMODEL_STRATEGY_CONFIG,
-              getString(WRITEMODEL_STRATEGY_CONFIG),
-              WriteModelStrategy.class);
+          configureInstance(
+              createInstance(
+                  WRITEMODEL_STRATEGY_CONFIG,
+                  getString(WRITEMODEL_STRATEGY_CONFIG),
+                  WriteModelStrategy.class));
     }
     return writeModelStrategy;
   }
@@ -490,41 +475,6 @@ private static Map<String, String> createSinkTopicOriginals(final Map<String, St
     return topicConfig;
   }
 
-  private Optional<FieldProjector> getFieldProjector(final String strategyClassName) {
-    FieldProjectionType keyProjectionType =
-        FieldProjectionType.valueOf(getString(KEY_PROJECTION_TYPE_CONFIG).toUpperCase());
-    FieldProjectionType valueProjectionType =
-        FieldProjectionType.valueOf(getString(VALUE_PROJECTION_TYPE_CONFIG).toUpperCase());
-
-    if (keyProjectionType.equals(FieldProjectionType.NONE)
-        && strategyClassName.equals(PartialKeyStrategy.class.getName())) {
-      throw new ConnectConfigException(
-          DOCUMENT_ID_STRATEGY_CONFIG,
-          strategyClassName,
-          format("Invalid %s value", KEY_PROJECTION_TYPE_CONFIG));
-    } else if (valueProjectionType.equals(FieldProjectionType.NONE)
-        && strategyClassName.equals(PartialValueStrategy.class.getName())) {
-      throw new ConnectConfigException(
-          DOCUMENT_ID_STRATEGY_CONFIG,
-          strategyClassName,
-          format("Invalid %s value", VALUE_PROJECTION_TYPE_CONFIG));
-    }
-
-    if (keyProjectionType.equals(FieldProjectionType.BLACKLIST)) {
-      return Optional.of(new BlacklistKeyProjector(this));
-    } else if (keyProjectionType.equals(FieldProjectionType.WHITELIST)) {
-      return Optional.of(new WhitelistKeyProjector(this));
-    }
-
-    if (valueProjectionType.equals(FieldProjectionType.BLACKLIST)) {
-      return Optional.of(new BlacklistValueProjector(this));
-    } else if (valueProjectionType.equals(FieldProjectionType.WHITELIST)) {
-      return Optional.of(new WhitelistValueProjector(this));
-    }
-
-    return Optional.empty();
-  }
-
   private static ConfigDef createConfigDef() {
 
     ConfigDef configDef = new ConfigDef();

src/main/java/com/mongodb/kafka/connect/sink/processor/AllowListKeyProjector.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Original Work: Apache License, Version 2.0, Copyright 2017 Hans-Peter Grahsl.
+ */
+
+package com.mongodb.kafka.connect.sink.processor;
+
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.FieldProjectionType.ALLOWLIST;
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.KEY_PROJECTION_LIST_CONFIG;
+
+import com.mongodb.kafka.connect.sink.MongoSinkTopicConfig;
+import com.mongodb.kafka.connect.sink.processor.field.projection.AllowListProjector;
+
+public class AllowListKeyProjector extends AllowListProjector {
+
+  public AllowListKeyProjector(final MongoSinkTopicConfig config) {
+    this(config, config.getString(KEY_PROJECTION_LIST_CONFIG));
+  }
+
+  public AllowListKeyProjector(final MongoSinkTopicConfig config, final String fieldList) {
+    super(config, buildProjectionList(ALLOWLIST, fieldList), SinkDocumentField.KEY);
+  }
+}

src/main/java/com/mongodb/kafka/connect/sink/processor/AllowListValueProjector.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Original Work: Apache License, Version 2.0, Copyright 2017 Hans-Peter Grahsl.
+ */
+
+package com.mongodb.kafka.connect.sink.processor;
+
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.FieldProjectionType.ALLOWLIST;
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.VALUE_PROJECTION_LIST_CONFIG;
+
+import com.mongodb.kafka.connect.sink.MongoSinkTopicConfig;
+import com.mongodb.kafka.connect.sink.processor.field.projection.AllowListProjector;
+
+public class AllowListValueProjector extends AllowListProjector {
+
+  public AllowListValueProjector(final MongoSinkTopicConfig config) {
+    this(config, config.getString(VALUE_PROJECTION_LIST_CONFIG));
+  }
+
+  public AllowListValueProjector(final MongoSinkTopicConfig config, final String fieldList) {
+    super(config, buildProjectionList(ALLOWLIST, fieldList), SinkDocumentField.VALUE);
+  }
+}

src/main/java/com/mongodb/kafka/connect/sink/processor/BlacklistKeyProjector.java

@@ -18,22 +18,13 @@
 
 package com.mongodb.kafka.connect.sink.processor;
 
-import org.apache.kafka.connect.sink.SinkRecord;
-
 import com.mongodb.kafka.connect.sink.MongoSinkTopicConfig;
-import com.mongodb.kafka.connect.sink.converter.SinkDocument;
-import com.mongodb.kafka.connect.sink.processor.field.projection.BlacklistProjector;
 
-public class BlacklistKeyProjector extends BlacklistProjector {
+/** @deprecated Use {@link BlockListKeyProjector} instead */
+@Deprecated
+public class BlacklistKeyProjector extends BlockListKeyProjector {
 
   public BlacklistKeyProjector(final MongoSinkTopicConfig config) {
-    super(config, getKeyFields(config));
-  }
-
-  @Override
-  public void process(final SinkDocument doc, final SinkRecord orig) {
-    if (isUsingBlacklistKeyProjection()) {
-      doc.getKeyDoc().ifPresent(kd -> getFields().forEach(f -> doProjection(f, kd)));
-    }
+    super(config);
   }
 }

src/main/java/com/mongodb/kafka/connect/sink/processor/BlacklistValueProjector.java

@@ -18,22 +18,13 @@
 
 package com.mongodb.kafka.connect.sink.processor;
 
-import org.apache.kafka.connect.sink.SinkRecord;
-
 import com.mongodb.kafka.connect.sink.MongoSinkTopicConfig;
-import com.mongodb.kafka.connect.sink.converter.SinkDocument;
-import com.mongodb.kafka.connect.sink.processor.field.projection.BlacklistProjector;
 
-public class BlacklistValueProjector extends BlacklistProjector {
+/** @deprecated Use {@link BlockListValueProjector} instead */
+@Deprecated
+public class BlacklistValueProjector extends BlockListValueProjector {
 
   public BlacklistValueProjector(final MongoSinkTopicConfig config) {
-    super(config, getValueFields(config));
-  }
-
-  @Override
-  public void process(final SinkDocument doc, final SinkRecord orig) {
-    if (isUsingBlacklistValueProjection()) {
-      doc.getValueDoc().ifPresent(vd -> getFields().forEach(f -> doProjection(f, vd)));
-    }
+    super(config);
   }
 }

src/main/java/com/mongodb/kafka/connect/sink/processor/BlockListKeyProjector.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Original Work: Apache License, Version 2.0, Copyright 2017 Hans-Peter Grahsl.
+ */
+
+package com.mongodb.kafka.connect.sink.processor;
+
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.FieldProjectionType.BLOCKLIST;
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.KEY_PROJECTION_LIST_CONFIG;
+
+import com.mongodb.kafka.connect.sink.MongoSinkTopicConfig;
+import com.mongodb.kafka.connect.sink.processor.field.projection.BlockListProjector;
+
+public class BlockListKeyProjector extends BlockListProjector {
+
+  public BlockListKeyProjector(final MongoSinkTopicConfig config) {
+    this(config, config.getString(KEY_PROJECTION_LIST_CONFIG));
+  }
+
+  public BlockListKeyProjector(final MongoSinkTopicConfig config, final String fieldList) {
+    super(config, buildProjectionList(BLOCKLIST, fieldList), SinkDocumentField.KEY);
+  }
+}

src/main/java/com/mongodb/kafka/connect/sink/processor/BlockListValueProjector.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Original Work: Apache License, Version 2.0, Copyright 2017 Hans-Peter Grahsl.
+ */
+
+package com.mongodb.kafka.connect.sink.processor;
+
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.FieldProjectionType.BLOCKLIST;
+import static com.mongodb.kafka.connect.sink.MongoSinkTopicConfig.VALUE_PROJECTION_LIST_CONFIG;
+
+import com.mongodb.kafka.connect.sink.MongoSinkTopicConfig;
+import com.mongodb.kafka.connect.sink.processor.field.projection.BlockListProjector;
+
+public class BlockListValueProjector extends BlockListProjector {
+
+  public BlockListValueProjector(final MongoSinkTopicConfig config) {
+    this(config, config.getString(VALUE_PROJECTION_LIST_CONFIG));
+  }
+
+  public BlockListValueProjector(final MongoSinkTopicConfig config, final String fieldList) {
+    super(config, buildProjectionList(BLOCKLIST, fieldList), SinkDocumentField.VALUE);
+  }
+}