Fix custom Delete write model strategy support

KAFKA-395

Co-authored-by: Ross Lawley <ross@mongodb.com>

Author: tlsdnwn55

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

@@ -525,22 +525,22 @@ public Optional<WriteModelStrategy> getDeleteWriteModelStrategy() {
       return Optional.empty();
     }
     if (deleteOneWriteModelStrategy == null) {
-      if (!getString(DELETE_WRITEMODEL_STRATEGY_CONFIG).isEmpty()) {
-        deleteOneWriteModelStrategy =
-            getWriteModelStrategyFromConfig(DELETE_WRITEMODEL_STRATEGY_CONFIG);
-      } else {
-        /*
-        NOTE: DeleteOneModel requires the key document which means that the only reasonable ID generation strategies are those
-        which refer to/operate on the key document. Thus currently this means the IdStrategy must be either:
-
-        FullKeyStrategy
-        PartialKeyStrategy
-        ProvidedInKeyStrategy
-        */
+      if (DELETE_WRITEMODEL_STRATEGY_DEFAULT.equals(getString(DELETE_WRITEMODEL_STRATEGY_CONFIG))) {
+
         IdStrategy idStrategy = getIdStrategy();
-        if (!(idStrategy instanceof FullKeyStrategy)
-            && !(idStrategy instanceof PartialKeyStrategy)
-            && !(idStrategy instanceof ProvidedInKeyStrategy)) {
+        if (idStrategy instanceof FullKeyStrategy
+            || idStrategy instanceof PartialKeyStrategy
+            || idStrategy instanceof ProvidedInKeyStrategy) {
+          deleteOneWriteModelStrategy = new DeleteOneDefaultStrategy(idStrategy);
+        } else {
+          /*
+          NOTE: DeleteOneModel requires the key document which means that the only reasonable ID generation strategies are those
+          which refer to/operate on the key document. Thus currently this means the IdStrategy must be either:
+
+          FullKeyStrategy
+          PartialKeyStrategy
+          ProvidedInKeyStrategy
+            */
           throw new ConnectConfigException(
               DELETE_ON_NULL_VALUES_CONFIG,
               getBoolean(DELETE_ON_NULL_VALUES_CONFIG),
@@ -551,7 +551,9 @@ public Optional<WriteModelStrategy> getDeleteWriteModelStrategy() {
                   PartialKeyStrategy.class.getSimpleName(),
                   ProvidedInKeyStrategy.class.getSimpleName()));
         }
-        deleteOneWriteModelStrategy = new DeleteOneDefaultStrategy(idStrategy);
+      } else {
+        deleteOneWriteModelStrategy =
+            getWriteModelStrategyFromConfig(DELETE_WRITEMODEL_STRATEGY_CONFIG);
       }
     }
 

src/main/java/com/mongodb/kafka/connect/sink/writemodel/strategy/DeleteOneDefaultStrategy.java

@@ -31,6 +31,7 @@
 
 import com.mongodb.kafka.connect.sink.converter.SinkDocument;
 import com.mongodb.kafka.connect.sink.processor.id.strategy.IdStrategy;
+import com.mongodb.kafka.connect.util.VisibleForTesting;
 
 public class DeleteOneDefaultStrategy implements WriteModelStrategy {
   private IdStrategy idStrategy;
@@ -43,6 +44,11 @@ public DeleteOneDefaultStrategy(final IdStrategy idStrategy) {
     this.idStrategy = idStrategy;
   }
 
+  @VisibleForTesting(otherwise = VisibleForTesting.AccessModifier.PRIVATE)
+  public IdStrategy getIdStrategy() {
+    return this.idStrategy;
+  }
+
   @Override
   public WriteModel<BsonDocument> createWriteModel(final SinkDocument document) {
 

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

@@ -72,6 +72,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -103,6 +104,7 @@
 import com.mongodb.kafka.connect.sink.processor.id.strategy.ProvidedInKeyStrategy;
 import com.mongodb.kafka.connect.sink.processor.id.strategy.ProvidedInValueStrategy;
 import com.mongodb.kafka.connect.sink.processor.id.strategy.UuidStrategy;
+import com.mongodb.kafka.connect.sink.writemodel.strategy.CustomDeleteWriteModelStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.DefaultWriteModelStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneBusinessKeyStrategy;
 import com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneDefaultStrategy;
@@ -833,6 +835,44 @@ Collection<DynamicTest> testGetSingleValidDeleteWriteModelStrategy() {
     return tests;
   }
 
+  @Test
+  @DisplayName("test Default DELETE_WRITEMODEL_STRATEGY_CONFIG")
+  void testDefaultDeleteWriteModelStrategyConfig() {
+    Map<String, String> map = createConfigMap();
+
+    map.put(DELETE_ON_NULL_VALUES_CONFIG, "true");
+    map.put(DOCUMENT_ID_STRATEGY_CONFIG, FullKeyStrategy.class.getName());
+    MongoSinkConfig cfg = new MongoSinkConfig(map);
+
+    Optional<WriteModelStrategy> optionalDeleteWriteModelStrategy =
+        cfg.getMongoSinkTopicConfig(TEST_TOPIC).getDeleteWriteModelStrategy();
+    assertTrue(optionalDeleteWriteModelStrategy.isPresent());
+
+    WriteModelStrategy writeModelStrategy = optionalDeleteWriteModelStrategy.get();
+    assertTrue(writeModelStrategy instanceof DeleteOneDefaultStrategy);
+
+    assertTrue(
+        ((DeleteOneDefaultStrategy) writeModelStrategy).getIdStrategy() instanceof FullKeyStrategy,
+        "IdStrategy is FullKeyStrategy");
+  }
+
+  @Test
+  @DisplayName("test Custom DELETE_WRITEMODEL_STRATEGY_CONFIG")
+  void testCustomDeleteWriteModelStrategyConfig() {
+    Map<String, String> map = createConfigMap();
+
+    map.put(DELETE_ON_NULL_VALUES_CONFIG, "true");
+    map.put(DELETE_WRITEMODEL_STRATEGY_CONFIG, CustomDeleteWriteModelStrategy.class.getName());
+    MongoSinkConfig cfg = new MongoSinkConfig(map);
+
+    Optional<WriteModelStrategy> optionalDeleteWriteModelStrategy =
+        cfg.getMongoSinkTopicConfig(TEST_TOPIC).getDeleteWriteModelStrategy();
+    assertTrue(optionalDeleteWriteModelStrategy.isPresent());
+
+    WriteModelStrategy writeModelStrategy = optionalDeleteWriteModelStrategy.get();
+    assertTrue(writeModelStrategy instanceof CustomDeleteWriteModelStrategy);
+  }
+
   @Test
   @DisplayName("test timeseries validation")
   void testTimeseries() {

src/test/java/com/mongodb/kafka/connect/sink/writemodel/strategy/CustomDeleteWriteModelStrategy.java

@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+package com.mongodb.kafka.connect.sink.writemodel.strategy;
+
+import org.bson.BsonDocument;
+
+import com.mongodb.client.model.WriteModel;
+
+import com.mongodb.kafka.connect.sink.converter.SinkDocument;
+
+public final class CustomDeleteWriteModelStrategy implements WriteModelStrategy {
+  @Override
+  public WriteModel<BsonDocument> createWriteModel(final SinkDocument document) {
+    return null;
+  }
+}